diff options
Diffstat (limited to 'functest/tests/unit/odl/test_odl.py')
| -rw-r--r-- | functest/tests/unit/odl/test_odl.py | 292 |
1 files changed, 226 insertions, 66 deletions
diff --git a/functest/tests/unit/odl/test_odl.py b/functest/tests/unit/odl/test_odl.py index ef18016b..d45d5628 100644 --- a/functest/tests/unit/odl/test_odl.py +++ b/functest/tests/unit/odl/test_odl.py @@ -11,13 +11,17 @@ import errno import logging import mock import os +import StringIO import unittest -from robot.errors import RobotError +from keystoneauth1.exceptions import auth_plugins +from robot.errors import DataError, RobotError +from robot.result import testcase +from robot.utils.robottime import timestamp_to_secs -import functest.core.testcase_base as testcase_base +mock.patch('logging.FileHandler').start() # noqa +from functest.core import testcase_base from functest.opnfv_tests.sdn.odl import odl -from functest.utils import functest_constants as ft_constants class ODLTesting(unittest.TestCase): @@ -36,19 +40,111 @@ class ODLTesting(unittest.TestCase): _odl_password = "admin" def setUp(self): - ft_constants.OS_USERNAME = self._os_username - ft_constants.OS_PASSWORD = self._os_password - ft_constants.OS_TENANT_NAME = self._os_tenantname + for var in ("INSTALLER_TYPE", "SDN_CONTROLLER", "SDN_CONTROLLER_IP"): + if var in os.environ: + del os.environ[var] + os.environ["OS_USERNAME"] = self._os_username + os.environ["OS_PASSWORD"] = self._os_password + os.environ["OS_TENANT_NAME"] = self._os_tenantname self.test = odl.ODLTests() + self.defaultargs = {'odlusername': self._odl_username, + 'odlpassword': self._odl_password, + 'keystoneip': self._keystone_ip, + 'neutronip': self._keystone_ip, + 'osusername': self._os_username, + 'ostenantname': self._os_tenantname, + 'ospassword': self._os_password, + 'odlip': self._keystone_ip, + 'odlwebport': self._odl_webport, + 'odlrestconfport': self._odl_restconfport, + 'pushtodb': False} + + def test_empty_visitor(self): + visitor = odl.ODLResultVisitor() + self.assertFalse(visitor.get_data()) + + def test_visitor(self): + visitor = odl.ODLResultVisitor() + data = {'name': 'foo', + 'parent': 'bar', + 'status': 'PASS', + 'starttime': "20161216 16:00:00.000", + 'endtime': "20161216 16:00:01.000", + 'elapsedtime': 1000, + 'text': 'Hello, World!', + 'critical': True} + test = testcase.TestCase(name=data['name'], + status=data['status'], + message=data['text'], + starttime=data['starttime'], + endtime=data['endtime']) + test.parent = mock.Mock() + config = {'name': data['parent'], + 'criticality.test_is_critical.return_value': data[ + 'critical']} + test.parent.configure_mock(**config) + visitor.visit_test(test) + self.assertEqual(visitor.get_data(), [data]) + + @mock.patch('robot.api.ExecutionResult', side_effect=DataError) + def test_parse_results_raises_exceptions(self, *args): + with self.assertRaises(DataError): + self.test.parse_results() + + def test_parse_results(self, *args): + config = {'name': 'dummy', 'starttime': '20161216 16:00:00.000', + 'endtime': '20161216 16:00:01.000', 'status': 'PASS'} + suite = mock.Mock() + suite.configure_mock(**config) + with mock.patch('robot.api.ExecutionResult', + return_value=mock.Mock(suite=suite)): + self.test.parse_results() + self.assertEqual(self.test.criteria, config['status']) + self.assertEqual(self.test.start_time, + timestamp_to_secs(config['starttime'])) + self.assertEqual(self.test.stop_time, + timestamp_to_secs(config['endtime'])) + self.assertEqual(self.test.details, + {'description': config['name'], 'tests': []}) @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): + def test_set_robotframework_vars_empty(self, args): self.assertTrue(self.test.set_robotframework_vars()) + @mock.patch('sys.stdout', new_callable=StringIO.StringIO) + def _test_set_robotframework_vars(self, msg1, msg2, *args): + line = mock.MagicMock() + line.__iter__.return_value = [msg1] + with mock.patch('fileinput.input', return_value=line) as mock_method: + self.assertTrue(self.test.set_robotframework_vars()) + mock_method.assert_called_once_with( + os.path.join(odl.ODLTests.odl_test_repo, + 'csit/variables/Variables.py'), inplace=True) + self.assertEqual(args[0].getvalue(), "{}\n".format(msg2)) + + def test_set_robotframework_vars_auth_default(self): + self._test_set_robotframework_vars("AUTH = []", + "AUTH = [u'admin', u'admin']") + + def test_set_robotframework_vars_auth1(self): + self._test_set_robotframework_vars("AUTH1 = []", "AUTH1 = []") + + @mock.patch('sys.stdout', new_callable=StringIO.StringIO) + def test_set_robotframework_vars_auth_foo(self, *args): + line = mock.MagicMock() + line.__iter__.return_value = ["AUTH = []"] + with mock.patch('fileinput.input', return_value=line) as mock_method: + self.assertTrue(self.test.set_robotframework_vars('foo', 'bar')) + mock_method.assert_called_once_with( + os.path.join(odl.ODLTests.odl_test_repo, + 'csit/variables/Variables.py'), inplace=True) + self.assertEqual(args[0].getvalue(), + "AUTH = [u'{}', u'{}']\n".format('foo', 'bar')) + @classmethod def _fake_url_for(cls, service_type='identity', **kwargs): if service_type == 'identity': @@ -59,14 +155,6 @@ class ODLTesting(unittest.TestCase): 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, @@ -85,9 +173,9 @@ class ODLTesting(unittest.TestCase): def _test_main(self, status, *args): kwargs = self._get_main_kwargs() self.assertEqual(self.test.main(**kwargs), status) - odl_res_dir = odl.ODLTests.res_dir if len(args) > 0: - args[0].assert_called_once_with(odl_res_dir) + args[0].assert_called_once_with( + odl.ODLTests.res_dir) if len(args) > 1: variable = ['KEYSTONE:{}'.format(self._keystone_ip), 'NEUTRON:{}'.format(self._neutron_ip), @@ -97,18 +185,17 @@ class ODLTesting(unittest.TestCase): 'ODL_SYSTEM_IP:{}'.format(self._sdn_controller_ip), 'PORT:{}'.format(self._odl_webport), 'RESTCONFPORT:{}'.format(self._odl_restconfport)] - output_file = os.path.join(odl_res_dir, 'output.xml') args[1].assert_called_once_with( odl.ODLTests.basic_suite_dir, odl.ODLTests.neutron_suite_dir, log='NONE', - output=output_file, + output=os.path.join(odl.ODLTests.res_dir, 'output.xml'), report='NONE', stdout=mock.ANY, variable=variable) if len(args) > 2: - stdout_file = os.path.join(odl_res_dir, 'stdout.txt') - args[2].assert_called_with(stdout_file) + args[2].assert_called_with( + os.path.join(odl.ODLTests.res_dir, 'stdout.txt')) def _test_main_missing_keyword(self, key): kwargs = self._get_main_kwargs(key) @@ -172,6 +259,8 @@ class ODLTesting(unittest.TestCase): def test_main_robot_run_failed(self, *args): with mock.patch.object(self.test, 'set_robotframework_vars', return_value=True), \ + mock.patch.object(odl, 'open', mock.mock_open(), + create=True), \ self.assertRaises(RobotError): self._test_main(testcase_base.TestcaseBase.EX_RUN_ERROR, *args) @@ -180,6 +269,8 @@ class ODLTesting(unittest.TestCase): def test_main_parse_results_failed(self, *args): with mock.patch.object(self.test, 'set_robotframework_vars', return_value=True), \ + mock.patch.object(odl, 'open', mock.mock_open(), + create=True), \ mock.patch.object(self.test, 'parse_results', side_effect=RobotError): self._test_main(testcase_base.TestcaseBase.EX_RUN_ERROR, *args) @@ -200,8 +291,9 @@ class ODLTesting(unittest.TestCase): def test_main(self, *args): with mock.patch.object(self.test, 'set_robotframework_vars', return_value=True), \ - mock.patch.object(self.test, 'parse_results', - return_value="PASS"): + mock.patch.object(odl, 'open', mock.mock_open(), + create=True), \ + mock.patch.object(self.test, 'parse_results'): self._test_main(testcase_base.TestcaseBase.EX_OK, *args) @mock.patch('os.remove') @@ -210,8 +302,9 @@ class ODLTesting(unittest.TestCase): def test_main_makedirs_oserror17(self, *args): with mock.patch.object(self.test, 'set_robotframework_vars', return_value=True), \ - mock.patch.object(self.test, 'parse_results', - return_value="PASS"): + mock.patch.object(odl, 'open', mock.mock_open(), + create=True), \ + mock.patch.object(self.test, 'parse_results'): self._test_main(testcase_base.TestcaseBase.EX_OK, *args) @mock.patch('os.remove') @@ -220,8 +313,9 @@ class ODLTesting(unittest.TestCase): def test_main_testcases_in_failure(self, *args): with mock.patch.object(self.test, 'set_robotframework_vars', return_value=True), \ - mock.patch.object(self.test, 'parse_results', - return_value="PASS"): + mock.patch.object(odl, 'open', mock.mock_open(), + create=True), \ + mock.patch.object(self.test, 'parse_results'): self._test_main(testcase_base.TestcaseBase.EX_OK, *args) @mock.patch('os.remove', side_effect=OSError) @@ -230,25 +324,22 @@ class ODLTesting(unittest.TestCase): def test_main_remove_oserror(self, *args): with mock.patch.object(self.test, 'set_robotframework_vars', return_value=True), \ - mock.patch.object(self.test, 'parse_results', - return_value="PASS"): + mock.patch.object(odl, 'open', mock.mock_open(), + create=True), \ + mock.patch.object(self.test, 'parse_results'): self._test_main(testcase_base.TestcaseBase.EX_OK, *args) def _test_run_missing_env_var(self, var): - if var == 'OS_USERNAME': - ft_constants.OS_USERNAME = None - elif var == 'OS_PASSWORD': - ft_constants.OS_PASSWORD = None - elif var == 'OS_TENANT_NAME': - ft_constants.OS_TENANT_NAME = None - - self.assertEqual(self.test.run(), - testcase_base.TestcaseBase.EX_RUN_ERROR) + with mock.patch('functest.utils.openstack_utils.get_endpoint', + side_effect=self._fake_url_for): + del os.environ[var] + self.assertEqual(self.test.run(), + testcase_base.TestcaseBase.EX_RUN_ERROR) def _test_run(self, status=testcase_base.TestcaseBase.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()): + with mock.patch('functest.utils.openstack_utils.get_endpoint', + side_effect=self._fake_url_for): if exception: self.test.main = mock.Mock(side_effect=exception) else: @@ -262,6 +353,12 @@ class ODLTesting(unittest.TestCase): ospassword=self._os_password, ostenantname=self._os_tenantname, osusername=self._os_username) + def test_run_exception(self): + with mock.patch('functest.utils.openstack_utils.get_endpoint', + side_effect=auth_plugins.MissingAuthPlugin()): + self.assertEqual(self.test.run(), + testcase_base.TestcaseBase.EX_RUN_ERROR) + def test_run_missing_os_username(self): self._test_run_missing_env_var("OS_USERNAME") @@ -272,75 +369,138 @@ class ODLTesting(unittest.TestCase): self._test_run_missing_env_var("OS_TENANT_NAME") def test_run_main_false(self): - ft_constants.CI_INSTALLER_TYPE = None - ft_constants.SDN_CONTROLLER_IP = self._sdn_controller_ip + os.environ["SDN_CONTROLLER_IP"] = self._sdn_controller_ip self._test_run(testcase_base.TestcaseBase.EX_RUN_ERROR, odlip=self._sdn_controller_ip, odlwebport=self._odl_webport) def test_run_main_exception(self): - ft_constants.CI_INSTALLER_TYPE = None - ft_constants.SDN_CONTROLLER_IP = self._sdn_controller_ip with self.assertRaises(Exception): + os.environ["SDN_CONTROLLER_IP"] = self._sdn_controller_ip self._test_run(status=testcase_base.TestcaseBase.EX_RUN_ERROR, exception=Exception(), odlip=self._sdn_controller_ip, odlwebport=self._odl_webport) def test_run_missing_sdn_controller_ip(self): - with mock.patch('functest.utils.openstack_utils.get_keystone_client', - return_value=self._get_fake_keystone_client()): - ft_constants.CI_INSTALLER_TYPE = None - ft_constants.SDN_CONTROLLER_IP = None + with mock.patch('functest.utils.openstack_utils.get_endpoint', + side_effect=self._fake_url_for): self.assertEqual(self.test.run(), testcase_base.TestcaseBase.EX_RUN_ERROR) def test_run_without_installer_type(self): - ft_constants.SDN_CONTROLLER_IP = self._sdn_controller_ip - ft_constants.CI_INSTALLER_TYPE = None + os.environ["SDN_CONTROLLER_IP"] = self._sdn_controller_ip self._test_run(testcase_base.TestcaseBase.EX_OK, odlip=self._sdn_controller_ip, odlwebport=self._odl_webport) def test_run_fuel(self): - ft_constants.CI_INSTALLER_TYPE = "fuel" + os.environ["INSTALLER_TYPE"] = "fuel" self._test_run(testcase_base.TestcaseBase.EX_OK, odlip=self._neutron_ip, odlwebport='8282') def test_run_apex_missing_sdn_controller_ip(self): - with mock.patch('functest.utils.openstack_utils.get_keystone_client', - return_value=self._get_fake_keystone_client()): - ft_constants.CI_INSTALLER_TYPE = "apex" - ft_constants.SDN_CONTROLLER_IP = None + with mock.patch('functest.utils.openstack_utils.get_endpoint', + side_effect=self._fake_url_for): + os.environ["INSTALLER_TYPE"] = "apex" self.assertEqual(self.test.run(), testcase_base.TestcaseBase.EX_RUN_ERROR) def test_run_apex(self): - ft_constants.SDN_CONTROLLER_IP = self._sdn_controller_ip - ft_constants.CI_INSTALLER_TYPE = "apex" + os.environ["SDN_CONTROLLER_IP"] = self._sdn_controller_ip + os.environ["INSTALLER_TYPE"] = "apex" self._test_run(testcase_base.TestcaseBase.EX_OK, odlip=self._sdn_controller_ip, odlwebport='8181') def test_run_joid_missing_sdn_controller(self): - with mock.patch('functest.utils.openstack_utils.get_keystone_client', - return_value=self._get_fake_keystone_client()): - ft_constants.CI_INSTALLER_TYPE = "joid" - ft_constants.SDN_CONTROLLER = None + with mock.patch('functest.utils.openstack_utils.get_endpoint', + side_effect=self._fake_url_for): + os.environ["INSTALLER_TYPE"] = "joid" self.assertEqual(self.test.run(), testcase_base.TestcaseBase.EX_RUN_ERROR) def test_run_joid(self): - ft_constants.SDN_CONTROLLER = self._sdn_controller_ip - ft_constants.CI_INSTALLER_TYPE = "joid" + os.environ["SDN_CONTROLLER"] = self._sdn_controller_ip + os.environ["INSTALLER_TYPE"] = "joid" self._test_run(testcase_base.TestcaseBase.EX_OK, - odlip=self._sdn_controller_ip, - odlwebport=self._odl_webport) + odlip=self._sdn_controller_ip, odlwebport='8080') def test_run_compass(self, *args): - ft_constants.CI_INSTALLER_TYPE = "compass" + os.environ["INSTALLER_TYPE"] = "compass" self._test_run(testcase_base.TestcaseBase.EX_OK, odlip=self._neutron_ip, odlwebport='8181') + def test_argparser_default(self): + parser = odl.ODLParser() + self.assertEqual(parser.parse_args(), self.defaultargs) + + def test_argparser_basic(self): + self.defaultargs['neutronip'] = self._neutron_ip + self.defaultargs['odlip'] = self._sdn_controller_ip + parser = odl.ODLParser() + self.assertEqual(parser.parse_args( + ["--neutronip={}".format(self._neutron_ip), + "--odlip={}".format(self._sdn_controller_ip) + ]), self.defaultargs) + + @mock.patch('sys.stderr', new_callable=StringIO.StringIO) + def test_argparser_fail(self, *args): + self.defaultargs['foo'] = 'bar' + parser = odl.ODLParser() + with self.assertRaises(SystemExit): + parser.parse_args(["--foo=bar"]) + + def _test_argparser(self, arg, value): + self.defaultargs[arg] = value + parser = odl.ODLParser() + self.assertEqual(parser.parse_args(["--{}={}".format(arg, value)]), + self.defaultargs) + + def test_argparser_odlusername(self): + self._test_argparser('odlusername', 'foo') + + def test_argparser_odlpassword(self): + self._test_argparser('odlpassword', 'foo') + + def test_argparser_keystoneip(self): + self._test_argparser('keystoneip', '127.0.0.4') + + def test_argparser_neutronip(self): + self._test_argparser('neutronip', '127.0.0.4') + + def test_argparser_osusername(self): + self._test_argparser('osusername', 'foo') + + def test_argparser_ostenantname(self): + self._test_argparser('ostenantname', 'foo') + + def test_argparser_ospassword(self): + self._test_argparser('ospassword', 'foo') + + def test_argparser_odlip(self): + self._test_argparser('odlip', '127.0.0.4') + + def test_argparser_odlwebport(self): + self._test_argparser('odlwebport', '80') + + def test_argparser_odlrestconfport(self): + self._test_argparser('odlrestconfport', '80') + + def test_argparser_pushtodb(self): + self.defaultargs['pushtodb'] = True + parser = odl.ODLParser() + self.assertEqual(parser.parse_args(["--{}".format('pushtodb')]), + self.defaultargs) + + def test_argparser_multiple_args(self): + self.defaultargs['neutronip'] = self._neutron_ip + self.defaultargs['odlip'] = self._sdn_controller_ip + parser = odl.ODLParser() + self.assertEqual(parser.parse_args( + ["--neutronip={}".format(self._neutron_ip), + "--odlip={}".format(self._sdn_controller_ip) + ]), self.defaultargs) + if __name__ == "__main__": unittest.main(verbosity=2) |