summaryrefslogtreecommitdiffstats
path: root/testcases
diff options
context:
space:
mode:
Diffstat (limited to 'testcases')
-rwxr-xr-xtestcases/Controllers/ODL/OpenDaylightTesting.py158
-rwxr-xr-xtestcases/Controllers/ODL/OpenDaylightUnitTesting.py341
2 files changed, 425 insertions, 74 deletions
diff --git a/testcases/Controllers/ODL/OpenDaylightTesting.py b/testcases/Controllers/ODL/OpenDaylightTesting.py
index 35a2ee117..e302b569e 100755
--- a/testcases/Controllers/ODL/OpenDaylightTesting.py
+++ b/testcases/Controllers/ODL/OpenDaylightTesting.py
@@ -1,6 +1,7 @@
#!/usr/bin/python
import argparse
+import errno
import fileinput
import os
import re
@@ -8,13 +9,13 @@ import shutil
import sys
import urlparse
-from robot import run
from robot.api import ExecutionResult, ResultVisitor
from robot.errors import RobotError
+import robot.run
from robot.utils.robottime import timestamp_to_secs
+from functest.core import TestCasesBase
import functest.utils.functest_logger as ft_logger
-import functest.utils.functest_utils as ft_utils
import functest.utils.openstack_utils as op_utils
@@ -39,7 +40,7 @@ class ODLResultVisitor(ResultVisitor):
return self._data
-class ODLTestCases:
+class ODLTestCases(TestCasesBase.TestCasesBase):
repos = "/home/opnfv/repos/"
odl_test_repo = repos + "odl_test/"
@@ -48,13 +49,16 @@ class ODLTestCases:
res_dir = '/home/opnfv/functest/results/odl/'
logger = ft_logger.Logger("opendaylight").getLogger()
+ def __init__(self):
+ self.case_name = "odl"
+
@classmethod
def copy_opnf_testcases(cls):
opnfv_testcases_dir = (os.path.dirname(os.path.abspath(__file__)) +
"/custom_tests/neutron/")
- file = opnfv_testcases_dir + "001__reachability.robot"
+ f = opnfv_testcases_dir + "001__reachability.robot"
try:
- shutil.copy(file, cls.neutron_suite_dir)
+ shutil.copy(f, cls.neutron_suite_dir)
except IOError as e:
cls.logger.error(
"Cannot copy OPNFV's testcase to ODL directory: %s" % str(e))
@@ -76,9 +80,19 @@ class ODLTestCases:
cls.logger.error("Cannot set ODL creds: %s" % str(e))
return False
- @classmethod
- def run(cls, **kwargs):
- dirs = [cls.basic_suite_dir, cls.neutron_suite_dir]
+ def parse_results(self):
+ result = ExecutionResult(self.res_dir + 'output.xml')
+ visitor = ODLResultVisitor()
+ result.visit(visitor)
+ self.criteria = result.suite.status
+ self.start_time = timestamp_to_secs(result.suite.starttime)
+ self.stop_time = timestamp_to_secs(result.suite.endtime)
+ self.details = {}
+ self.details['description'] = result.suite.name
+ self.details['tests'] = visitor.get_data()
+
+ def main(self, **kwargs):
+ dirs = [self.basic_suite_dir, self.neutron_suite_dir]
try:
odlusername = kwargs['odlusername']
odlpassword = kwargs['odlpassword']
@@ -91,49 +105,60 @@ class ODLTestCases:
'PORT:' + kwargs['odlwebport'],
'RESTCONFPORT:' + kwargs['odlrestconfport']]
except KeyError as e:
- cls.logger.error("Cannot run ODL testcases. Please check "
- "%s" % str(e))
- return False
- if (cls.copy_opnf_testcases() and
- cls.set_robotframework_vars(odlusername, odlpassword)):
+ self.logger.error("Cannot run ODL testcases. Please check "
+ "%s" % str(e))
+ return self.EX_RUN_ERROR
+ if (self.copy_opnf_testcases() and
+ self.set_robotframework_vars(odlusername, odlpassword)):
try:
- os.makedirs(cls.res_dir)
- except OSError:
- pass
- stdout_file = cls.res_dir + 'stdout.txt'
+ os.makedirs(self.res_dir)
+ except OSError as e:
+ if e.errno != errno.EEXIST:
+ self.logger.exception(
+ "Cannot create {}".format(self.res_dir))
+ return self.EX_RUN_ERROR
+ stdout_file = self.res_dir + 'stdout.txt'
with open(stdout_file, 'w+') as stdout:
- run(*dirs, variable=variables,
- output=cls.res_dir + 'output.xml',
- log='NONE',
- report='NONE',
- stdout=stdout)
+ robot.run(*dirs, variable=variables,
+ output=self.res_dir + 'output.xml',
+ log='NONE',
+ report='NONE',
+ stdout=stdout)
stdout.seek(0, 0)
- cls.logger.info("\n" + stdout.read())
- cls.logger.info("ODL results were successfully generated")
+ self.logger.info("\n" + stdout.read())
+ self.logger.info("ODL results were successfully generated")
+ try:
+ self.parse_results()
+ self.logger.info("ODL results were successfully parsed")
+ except RobotError as e:
+ self.logger.error("Run tests before publishing: %s" %
+ e.message)
+ return self.EX_RUN_ERROR
try:
os.remove(stdout_file)
except OSError:
- pass
- return True
+ self.logger.warning("Cannot remove {}".format(stdout_file))
+ return self.EX_OK
else:
- return False
+ return self.EX_RUN_ERROR
- @classmethod
- def functest_run(cls):
- kclient = op_utils.get_keystone_client()
- keystone_url = kclient.service_catalog.url_for(
- service_type='identity', endpoint_type='publicURL')
- neutron_url = kclient.service_catalog.url_for(
- service_type='network', endpoint_type='publicURL')
- kwargs = {'keystoneip': urlparse.urlparse(keystone_url).hostname}
- kwargs['neutronip'] = urlparse.urlparse(neutron_url).hostname
- kwargs['odlip'] = kwargs['neutronip']
- kwargs['odlwebport'] = '8080'
- kwargs['odlrestconfport'] = '8181'
- kwargs['odlusername'] = 'admin'
- kwargs['odlpassword'] = 'admin'
+ def run(self):
try:
- installer_type = os.environ['INSTALLER_TYPE']
+ kclient = op_utils.get_keystone_client()
+ keystone_url = kclient.service_catalog.url_for(
+ service_type='identity', endpoint_type='publicURL')
+ neutron_url = kclient.service_catalog.url_for(
+ service_type='network', endpoint_type='publicURL')
+ kwargs = {'keystoneip': urlparse.urlparse(keystone_url).hostname}
+ kwargs['neutronip'] = urlparse.urlparse(neutron_url).hostname
+ kwargs['odlip'] = kwargs['neutronip']
+ kwargs['odlwebport'] = '8080'
+ kwargs['odlrestconfport'] = '8181'
+ kwargs['odlusername'] = 'admin'
+ kwargs['odlpassword'] = 'admin'
+ installer_type = None
+ if 'INSTALLER_TYPE' in os.environ:
+ installer_type = os.environ['INSTALLER_TYPE']
kwargs['osusername'] = os.environ['OS_USERNAME']
kwargs['ostenantname'] = os.environ['OS_TENANT_NAME']
kwargs['ospassword'] = os.environ['OS_PASSWORD']
@@ -149,34 +174,15 @@ class ODLTestCases:
else:
kwargs['odlip'] = os.environ['SDN_CONTROLLER_IP']
except KeyError as e:
- cls.logger.error("Cannot run ODL testcases. Please check env var: "
- "%s" % str(e))
- return False
-
- return cls.run(**kwargs)
-
- @classmethod
- def push_to_db(cls):
- try:
- result = ExecutionResult(cls.res_dir + 'output.xml')
- visitor = ODLResultVisitor()
- result.visit(visitor)
- start_time = timestamp_to_secs(result.suite.starttime)
- stop_time = timestamp_to_secs(result.suite.endtime)
- details = {}
- details['description'] = result.suite.name
- details['tests'] = visitor.get_data()
- if not ft_utils.push_results_to_db(
- "functest", "odl", start_time, stop_time,
- result.suite.status, details):
- cls.logger.error("Cannot push ODL results to DB")
- return False
- else:
- cls.logger.info("ODL results were successfully pushed to DB")
- return True
- except RobotError as e:
- cls.logger.error("Run tests before publishing: %s" % e.message)
+ self.logger.error("Cannot run ODL testcases. "
+ "Please check env var: "
+ "%s" % str(e))
+ return self.EX_RUN_ERROR
+ except Exception:
+ self.logger.exception("Cannot run ODL testcases.")
+ return self.EX_RUN_ERROR
+ return self.main(**kwargs)
if __name__ == '__main__':
parser = argparse.ArgumentParser()
@@ -215,8 +221,12 @@ if __name__ == '__main__':
action='store_true')
args = vars(parser.parse_args())
- if not ODLTestCases.run(**args):
- sys.exit(os.EX_SOFTWARE)
- if args['pushtodb']:
- sys.exit(not ODLTestCases.push_to_db())
- sys.exit(os.EX_OK)
+ odl = ODLTestCases()
+ try:
+ result = odl.main(**args)
+ if result != TestCasesBase.TestCasesBase.EX_OK:
+ sys.exit(result)
+ if args['pushtodb']:
+ sys.exit(odl.push_to_db())
+ except Exception:
+ sys.exit(TestCasesBase.TestCasesBase.EX_RUN_ERROR)
diff --git a/testcases/Controllers/ODL/OpenDaylightUnitTesting.py b/testcases/Controllers/ODL/OpenDaylightUnitTesting.py
new file mode 100755
index 000000000..500888b22
--- /dev/null
+++ b/testcases/Controllers/ODL/OpenDaylightUnitTesting.py
@@ -0,0 +1,341 @@
+#!/usr/bin/env python
+
+import errno
+import logging
+import mock
+import os
+import unittest
+
+from robot.errors import RobotError
+
+from functest.core import TestCasesBase
+from functest.testcases.Controllers.ODL import OpenDaylightTesting
+
+
+class ODLTestCasesTesting(unittest.TestCase):
+
+ logging.disable(logging.CRITICAL)
+
+ _keystone_ip = "127.0.0.1"
+ _neutron_ip = "127.0.0.2"
+ _sdn_controller_ip = "127.0.0.3"
+ _os_tenantname = "admin"
+ _os_username = "admin"
+ _os_password = "admin"
+ _odl_webport = "8080"
+ _odl_restconfport = "8181"
+ _odl_username = "admin"
+ _odl_password = "admin"
+
+ def setUp(self):
+ if "INSTALLER_TYPE" in os.environ:
+ del os.environ["INSTALLER_TYPE"]
+ os.environ["OS_USERNAME"] = self._os_username
+ os.environ["OS_PASSWORD"] = self._os_password
+ os.environ["OS_TENANT_NAME"] = self._os_tenantname
+ self.test = OpenDaylightTesting.ODLTestCases()
+
+ @mock.patch('shutil.copy', side_effect=Exception())
+ def test_copy_opnf_testcases_exception(self, *args):
+ with self.assertRaises(Exception):
+ self.test.copy_opnf_testcases()
+
+ @mock.patch('shutil.copy', side_effect=IOError())
+ def test_copy_opnf_testcases_ioerror(self, *args):
+ self.assertFalse(self.test.copy_opnf_testcases())
+
+ @mock.patch('shutil.copy')
+ def test_copy_opnf_testcases(self, *args):
+ self.assertTrue(self.test.copy_opnf_testcases())
+
+ @mock.patch('fileinput.input', side_effect=Exception())
+ def test_set_robotframework_vars_failed(self, *args):
+ self.assertFalse(self.test.set_robotframework_vars())
+
+ @mock.patch('fileinput.input', return_value=[])
+ def test_set_robotframework_vars(self, args):
+ self.assertTrue(self.test.set_robotframework_vars())
+
+ def _test_run_missing_env_var(self, var):
+ del os.environ[var]
+ self.assertEqual(self.test.run(),
+ TestCasesBase.TestCasesBase.EX_RUN_ERROR)
+
+ @classmethod
+ def _fake_url_for(cls, service_type='identity', **kwargs):
+ if service_type == 'identity':
+ return "http://{}:5000/v2.0".format(
+ ODLTestCasesTesting._keystone_ip)
+ elif service_type == 'network':
+ return "http://{}:9696".format(ODLTestCasesTesting._neutron_ip)
+ else:
+ return None
+
+ @classmethod
+ def _get_fake_keystone_client(cls):
+ kclient = mock.Mock()
+ kclient.service_catalog = mock.Mock()
+ kclient.service_catalog.url_for = mock.Mock(
+ side_effect=cls._fake_url_for)
+ return kclient
+
+ def _get_main_kwargs(self, key=None):
+ kwargs = {'odlusername': self._odl_username,
+ 'odlpassword': self._odl_password,
+ 'keystoneip': self._keystone_ip,
+ 'neutronip': self._neutron_ip,
+ 'osusername': self._os_username,
+ 'ostenantname': self._os_tenantname,
+ 'ospassword': self._os_password,
+ 'odlip': self._sdn_controller_ip,
+ 'odlwebport': self._odl_webport,
+ 'odlrestconfport': self._odl_restconfport}
+ if key:
+ del kwargs[key]
+ return kwargs
+
+ def _test_main(self, status, *args):
+ kwargs = self._get_main_kwargs()
+ self.assertEqual(self.test.main(**kwargs), status)
+ if len(args) > 0:
+ args[0].assert_called_once_with(
+ OpenDaylightTesting.ODLTestCases.res_dir)
+ if len(args) > 1:
+ variable = ['KEYSTONE:{}'.format(self._keystone_ip),
+ 'NEUTRON:{}'.format(self._neutron_ip),
+ 'OSUSERNAME:"{}"'.format(self._os_username),
+ 'OSTENANTNAME:"{}"'.format(self._os_tenantname),
+ 'OSPASSWORD:"{}"'.format(self._os_password),
+ 'ODL_SYSTEM_IP:{}'.format(self._sdn_controller_ip),
+ 'PORT:{}'.format(self._odl_webport),
+ 'RESTCONFPORT:{}'.format(self._odl_restconfport)]
+ args[1].assert_called_once_with(
+ OpenDaylightTesting.ODLTestCases.basic_suite_dir,
+ OpenDaylightTesting.ODLTestCases.neutron_suite_dir,
+ log='NONE',
+ output=OpenDaylightTesting.ODLTestCases.res_dir + 'output.xml',
+ report='NONE',
+ stdout=mock.ANY,
+ variable=variable)
+ if len(args) > 2:
+ args[2].assert_called_with(
+ OpenDaylightTesting.ODLTestCases.res_dir + 'stdout.txt')
+
+ def _test_main_missing_keyword(self, key):
+ kwargs = self._get_main_kwargs(key)
+ self.assertEqual(self.test.main(**kwargs),
+ TestCasesBase.TestCasesBase.EX_RUN_ERROR)
+
+ def test_main_missing_odlusername(self):
+ self._test_main_missing_keyword('odlusername')
+
+ def test_main_missing_odlpassword(self):
+ self._test_main_missing_keyword('odlpassword')
+
+ def test_main_missing_keystoneip(self):
+ self._test_main_missing_keyword('keystoneip')
+
+ def test_main_missing_neutronip(self):
+ self._test_main_missing_keyword('neutronip')
+
+ def test_main_missing_osusername(self):
+ self._test_main_missing_keyword('osusername')
+
+ def test_main_missing_ostenantname(self):
+ self._test_main_missing_keyword('ostenantname')
+
+ def test_main_missing_ospassword(self):
+ self._test_main_missing_keyword('ospassword')
+
+ def test_main_missing_odlip(self):
+ self._test_main_missing_keyword('odlip')
+
+ def test_main_missing_odlwebport(self):
+ self._test_main_missing_keyword('odlwebport')
+
+ def test_main_missing_odlrestconfport(self):
+ self._test_main_missing_keyword('odlrestconfport')
+
+ def test_main_copy_opnf_testcases_failed(self):
+ with mock.patch.object(self.test, 'copy_opnf_testcases',
+ return_value=False):
+ self._test_main(TestCasesBase.TestCasesBase.EX_RUN_ERROR)
+ self.test.copy_opnf_testcases.assert_called_once_with()
+
+ def test_main_set_robotframework_vars_failed(self):
+ with mock.patch.object(self.test, 'copy_opnf_testcases',
+ return_value=True), \
+ mock.patch.object(self.test, 'set_robotframework_vars',
+ return_value=False):
+ self._test_main(TestCasesBase.TestCasesBase.EX_RUN_ERROR)
+ self.test.set_robotframework_vars.assert_called_once_with(
+ self._odl_username, self._odl_password)
+
+ @mock.patch('os.makedirs', side_effect=Exception)
+ def test_main_makedirs_exception(self, mock_method):
+ with mock.patch.object(self.test,
+ 'copy_opnf_testcases', return_value=True), \
+ mock.patch.object(self.test, 'set_robotframework_vars',
+ return_value=True), \
+ self.assertRaises(Exception):
+ self._test_main(TestCasesBase.TestCasesBase.EX_RUN_ERROR,
+ mock_method)
+
+ @mock.patch('os.makedirs', side_effect=OSError)
+ def test_main_makedirs_oserror(self, mock_method):
+ with mock.patch.object(self.test,
+ 'copy_opnf_testcases', return_value=True), \
+ mock.patch.object(self.test, 'set_robotframework_vars',
+ return_value=True):
+ self._test_main(TestCasesBase.TestCasesBase.EX_RUN_ERROR,
+ mock_method)
+
+ @mock.patch('robot.run', side_effect=RobotError)
+ @mock.patch('os.makedirs')
+ def test_main_robot_run_failed(self, *args):
+ with mock.patch.object(self.test, 'copy_opnf_testcases',
+ return_value=True), \
+ mock.patch.object(self.test, 'set_robotframework_vars',
+ return_value=True), \
+ self.assertRaises(RobotError):
+ self._test_main(TestCasesBase.TestCasesBase.EX_RUN_ERROR, *args)
+
+ @mock.patch('robot.run')
+ @mock.patch('os.makedirs')
+ def test_main_parse_results_failed(self, *args):
+ with mock.patch.object(self.test, 'copy_opnf_testcases',
+ return_value=True), \
+ mock.patch.object(self.test, 'set_robotframework_vars',
+ return_value=True), \
+ mock.patch.object(self.test, 'parse_results',
+ side_effect=RobotError):
+ self._test_main(TestCasesBase.TestCasesBase.EX_RUN_ERROR, *args)
+
+ @mock.patch('os.remove', side_effect=Exception)
+ @mock.patch('robot.run')
+ @mock.patch('os.makedirs')
+ def test_main_remove_exception(self, *args):
+ with mock.patch.object(self.test, 'copy_opnf_testcases',
+ return_value=True), \
+ mock.patch.object(self.test, 'set_robotframework_vars',
+ return_value=True), \
+ mock.patch.object(self.test, 'parse_results'), \
+ self.assertRaises(Exception):
+ self._test_main(TestCasesBase.TestCasesBase.EX_OK, *args)
+
+ @mock.patch('os.remove')
+ @mock.patch('robot.run')
+ @mock.patch('os.makedirs')
+ def test_main(self, *args):
+ with mock.patch.object(self.test, 'copy_opnf_testcases',
+ return_value=True), \
+ mock.patch.object(self.test, 'set_robotframework_vars',
+ return_value=True), \
+ mock.patch.object(self.test, 'parse_results'):
+ self._test_main(TestCasesBase.TestCasesBase.EX_OK, *args)
+
+ @mock.patch('os.remove')
+ @mock.patch('robot.run')
+ @mock.patch('os.makedirs', side_effect=OSError(errno.EEXIST, ''))
+ def test_main_makedirs_oserror17(self, *args):
+ with mock.patch.object(self.test, 'copy_opnf_testcases',
+ return_value=True), \
+ mock.patch.object(self.test, 'set_robotframework_vars',
+ return_value=True), \
+ mock.patch.object(self.test, 'parse_results'):
+ self._test_main(TestCasesBase.TestCasesBase.EX_OK, *args)
+
+ @mock.patch('os.remove')
+ @mock.patch('robot.run', return_value=1)
+ @mock.patch('os.makedirs')
+ def test_main_testcases_in_failure(self, *args):
+ with mock.patch.object(self.test, 'copy_opnf_testcases',
+ return_value=True), \
+ mock.patch.object(self.test, 'set_robotframework_vars',
+ return_value=True), \
+ mock.patch.object(self.test, 'parse_results'):
+ self._test_main(TestCasesBase.TestCasesBase.EX_OK, *args)
+
+ @mock.patch('os.remove', side_effect=OSError)
+ @mock.patch('robot.run')
+ @mock.patch('os.makedirs')
+ def test_main_remove_oserror(self, *args):
+ with mock.patch.object(self.test, 'copy_opnf_testcases',
+ return_value=True), \
+ mock.patch.object(self.test, 'set_robotframework_vars',
+ return_value=True), \
+ mock.patch.object(self.test, 'parse_results'):
+ self._test_main(TestCasesBase.TestCasesBase.EX_OK, *args)
+
+ def _test_run(self, status=TestCasesBase.TestCasesBase.EX_OK,
+ exception=None, odlip="127.0.0.3", odlwebport="8080"):
+ with mock.patch('functest.utils.openstack_utils.get_keystone_client',
+ return_value=self._get_fake_keystone_client()):
+ if exception:
+ self.test.main = mock.Mock(side_effect=exception)
+ else:
+ self.test.main = mock.Mock(return_value=status)
+ self.assertEqual(self.test.run(), status)
+ self.test.main.assert_called_once_with(
+ keystoneip=self._keystone_ip, neutronip=self._neutron_ip,
+ odlip=odlip, odlpassword=self._odl_password,
+ odlrestconfport=self._odl_restconfport,
+ odlusername=self._odl_username, odlwebport=odlwebport,
+ ospassword=self._os_password, ostenantname=self._os_tenantname,
+ osusername=self._os_username)
+
+ def test_run_missing_os_username(self):
+ self._test_run_missing_env_var("OS_USERNAME")
+
+ def test_run_missing_os_password(self):
+ self._test_run_missing_env_var("OS_PASSWORD")
+
+ def test_run_missing_os_tenant_name(self):
+ self._test_run_missing_env_var("OS_TENANT_NAME")
+
+ def test_run_main_false(self):
+ os.environ["SDN_CONTROLLER_IP"] = self._sdn_controller_ip
+ self._test_run(TestCasesBase.TestCasesBase.EX_RUN_ERROR,
+ odlip=self._sdn_controller_ip,
+ odlwebport=self._odl_webport)
+
+ def test_run_main_exception(self):
+ os.environ["SDN_CONTROLLER_IP"] = self._sdn_controller_ip
+ with self.assertRaises(Exception):
+ self._test_run(status=TestCasesBase.TestCasesBase.EX_RUN_ERROR,
+ exception=Exception(),
+ odlip=self._sdn_controller_ip,
+ odlwebport=self._odl_webport)
+
+ def test_run_without_installer_type(self):
+ os.environ["SDN_CONTROLLER_IP"] = self._sdn_controller_ip
+ self._test_run(TestCasesBase.TestCasesBase.EX_OK,
+ odlip=self._sdn_controller_ip,
+ odlwebport=self._odl_webport)
+
+ def test_run_fuel(self):
+ os.environ["INSTALLER_TYPE"] = "fuel"
+ self._test_run(TestCasesBase.TestCasesBase.EX_OK,
+ odlip=self._neutron_ip, odlwebport='8282')
+
+ def test_run_apex(self):
+ os.environ["SDN_CONTROLLER_IP"] = self._sdn_controller_ip
+ os.environ["INSTALLER_TYPE"] = "apex"
+ self._test_run(TestCasesBase.TestCasesBase.EX_OK,
+ odlip=self._sdn_controller_ip, odlwebport='8181')
+
+ def test_run_joid(self):
+ os.environ["SDN_CONTROLLER"] = self._sdn_controller_ip
+ os.environ["INSTALLER_TYPE"] = "joid"
+ self._test_run(TestCasesBase.TestCasesBase.EX_OK,
+ odlip=self._sdn_controller_ip,
+ odlwebport=self._odl_webport)
+
+ def test_run_compass(self, *args):
+ os.environ["INSTALLER_TYPE"] = "compass"
+ self._test_run(TestCasesBase.TestCasesBase.EX_OK,
+ odlip=self._neutron_ip, odlwebport='8181')
+
+
+if __name__ == "__main__":
+ unittest.main(verbosity=2)