functest/tests/unit/odl/test_odl.py

   1 #!/usr/bin/env python
   2
   3 # Copyright (c) 2016 Orange and others.
   4 #
   5 # All rights reserved. This program and the accompanying materials
   6 # are made available under the terms of the Apache License, Version 2.0
   7 # which accompanies this distribution, and is available at
   8 # http://www.apache.org/licenses/LICENSE-2.0
   9
  10 import errno
  11 import logging
  12 import mock
  13 import os
  14 import StringIO
  15 import unittest
  16
  17 from keystoneauth1.exceptions import auth_plugins
  18 from robot.errors import DataError, RobotError
  19 from robot.result import testcase
  20 from robot.utils.robottime import timestamp_to_secs
  21
  22 from functest.core import testcase_base
  23 from functest.opnfv_tests.sdn.odl import odl
  24
  25
  26 class ODLTesting(unittest.TestCase):
  27
  28     logging.disable(logging.CRITICAL)
  29
  30     _keystone_ip = "127.0.0.1"
  31     _neutron_ip = "127.0.0.2"
  32     _sdn_controller_ip = "127.0.0.3"
  33     _os_tenantname = "admin"
  34     _os_username = "admin"
  35     _os_password = "admin"
  36     _odl_webport = "8080"
  37     _odl_restconfport = "8181"
  38     _odl_username = "admin"
  39     _odl_password = "admin"
  40
  41     def setUp(self):
  42         for var in ("INSTALLER_TYPE", "SDN_CONTROLLER", "SDN_CONTROLLER_IP"):
  43             if var in os.environ:
  44                 del os.environ[var]
  45         os.environ["OS_USERNAME"] = self._os_username
  46         os.environ["OS_PASSWORD"] = self._os_password
  47         os.environ["OS_TENANT_NAME"] = self._os_tenantname
  48         self.test = odl.ODLTests()
  49         self.defaultargs = {'odlusername': self._odl_username,
  50                             'odlpassword': self._odl_password,
  51                             'keystoneip': self._keystone_ip,
  52                             'neutronip': self._keystone_ip,
  53                             'osusername': self._os_username,
  54                             'ostenantname': self._os_tenantname,
  55                             'ospassword': self._os_password,
  56                             'odlip': self._keystone_ip,
  57                             'odlwebport': self._odl_webport,
  58                             'odlrestconfport': self._odl_restconfport,
  59                             'pushtodb': False}
  60
  61     def test_empty_visitor(self):
  62         visitor = odl.ODLResultVisitor()
  63         self.assertFalse(visitor.get_data())
  64
  65     def test_visitor(self):
  66         visitor = odl.ODLResultVisitor()
  67         data = {'name': 'foo',
  68                 'parent': 'bar',
  69                 'status': 'PASS',
  70                 'starttime': "20161216 16:00:00.000",
  71                 'endtime': "20161216 16:00:01.000",
  72                 'elapsedtime': 1000,
  73                 'text': 'Hello, World!',
  74                 'critical': True}
  75         test = testcase.TestCase(name=data['name'],
  76                                  status=data['status'],
  77                                  message=data['text'],
  78                                  starttime=data['starttime'],
  79                                  endtime=data['endtime'])
  80         test.parent = mock.Mock()
  81         config = {'name': data['parent'],
  82                   'criticality.test_is_critical.return_value': data[
  83                       'critical']}
  84         test.parent.configure_mock(**config)
  85         visitor.visit_test(test)
  86         self.assertEqual(visitor.get_data(), [data])
  87
  88     @mock.patch('robot.api.ExecutionResult', side_effect=DataError)
  89     def test_parse_results_raises_exceptions(self, *args):
  90         with self.assertRaises(DataError):
  91             self.test.parse_results()
  92
  93     def test_parse_results(self, *args):
  94         config = {'name': 'dummy', 'starttime': '20161216 16:00:00.000',
  95                   'endtime': '20161216 16:00:01.000', 'status': 'PASS'}
  96         suite = mock.Mock()
  97         suite.configure_mock(**config)
  98         with mock.patch('robot.api.ExecutionResult',
  99                         return_value=mock.Mock(suite=suite)):
 100             self.test.parse_results()
 101             self.assertEqual(self.test.criteria, config['status'])
 102             self.assertEqual(self.test.start_time,
 103                              timestamp_to_secs(config['starttime']))
 104             self.assertEqual(self.test.stop_time,
 105                              timestamp_to_secs(config['endtime']))
 106             self.assertEqual(self.test.details,
 107                              {'description': config['name'], 'tests': []})
 108
 109     @mock.patch('fileinput.input', side_effect=Exception())
 110     def test_set_robotframework_vars_failed(self, *args):
 111         self.assertFalse(self.test.set_robotframework_vars())
 112
 113     @mock.patch('fileinput.input', return_value=[])
 114     def test_set_robotframework_vars_empty(self, args):
 115         self.assertTrue(self.test.set_robotframework_vars())
 116
 117     @mock.patch('sys.stdout', new_callable=StringIO.StringIO)
 118     def _test_set_robotframework_vars(self, msg1, msg2, *args):
 119         line = mock.MagicMock()
 120         line.__iter__.return_value = [msg1]
 121         with mock.patch('fileinput.input', return_value=line) as mock_method:
 122             self.assertTrue(self.test.set_robotframework_vars())
 123             mock_method.assert_called_once_with(
 124                 os.path.join(odl.ODLTests.odl_test_repo,
 125                              'csit/variables/Variables.py'), inplace=True)
 126             self.assertEqual(args[0].getvalue(), "{}\n".format(msg2))
 127
 128     def test_set_robotframework_vars_auth_default(self):
 129         self._test_set_robotframework_vars("AUTH = []",
 130                                            "AUTH = [u'admin', u'admin']")
 131
 132     def test_set_robotframework_vars_auth1(self):
 133         self._test_set_robotframework_vars("AUTH1 = []", "AUTH1 = []")
 134
 135     @mock.patch('sys.stdout', new_callable=StringIO.StringIO)
 136     def test_set_robotframework_vars_auth_foo(self, *args):
 137         line = mock.MagicMock()
 138         line.__iter__.return_value = ["AUTH = []"]
 139         with mock.patch('fileinput.input', return_value=line) as mock_method:
 140             self.assertTrue(self.test.set_robotframework_vars('foo', 'bar'))
 141             mock_method.assert_called_once_with(
 142                 os.path.join(odl.ODLTests.odl_test_repo,
 143                              'csit/variables/Variables.py'), inplace=True)
 144             self.assertEqual(args[0].getvalue(),
 145                              "AUTH = [u'{}', u'{}']\n".format('foo', 'bar'))
 146
 147     @classmethod
 148     def _fake_url_for(cls, service_type='identity', **kwargs):
 149         if service_type == 'identity':
 150             return "http://{}:5000/v2.0".format(
 151                 ODLTesting._keystone_ip)
 152         elif service_type == 'network':
 153             return "http://{}:9696".format(ODLTesting._neutron_ip)
 154         else:
 155             return None
 156
 157     def _get_main_kwargs(self, key=None):
 158         kwargs = {'odlusername': self._odl_username,
 159                   'odlpassword': self._odl_password,
 160                   'keystoneip': self._keystone_ip,
 161                   'neutronip': self._neutron_ip,
 162                   'osusername': self._os_username,
 163                   'ostenantname': self._os_tenantname,
 164                   'ospassword': self._os_password,
 165                   'odlip': self._sdn_controller_ip,
 166                   'odlwebport': self._odl_webport,
 167                   'odlrestconfport': self._odl_restconfport}
 168         if key:
 169             del kwargs[key]
 170         return kwargs
 171
 172     def _test_main(self, status, *args):
 173         kwargs = self._get_main_kwargs()
 174         self.assertEqual(self.test.main(**kwargs), status)
 175         if len(args) > 0:
 176             args[0].assert_called_once_with(
 177                 odl.ODLTests.res_dir)
 178         if len(args) > 1:
 179             variable = ['KEYSTONE:{}'.format(self._keystone_ip),
 180                         'NEUTRON:{}'.format(self._neutron_ip),
 181                         'OSUSERNAME:"{}"'.format(self._os_username),
 182                         'OSTENANTNAME:"{}"'.format(self._os_tenantname),
 183                         'OSPASSWORD:"{}"'.format(self._os_password),
 184                         'ODL_SYSTEM_IP:{}'.format(self._sdn_controller_ip),
 185                         'PORT:{}'.format(self._odl_webport),
 186                         'RESTCONFPORT:{}'.format(self._odl_restconfport)]
 187             args[1].assert_called_once_with(
 188                 odl.ODLTests.basic_suite_dir,
 189                 odl.ODLTests.neutron_suite_dir,
 190                 log='NONE',
 191                 output=os.path.join(odl.ODLTests.res_dir, 'output.xml'),
 192                 report='NONE',
 193                 stdout=mock.ANY,
 194                 variable=variable)
 195         if len(args) > 2:
 196             args[2].assert_called_with(
 197                 os.path.join(odl.ODLTests.res_dir, 'stdout.txt'))
 198
 199     def _test_main_missing_keyword(self, key):
 200         kwargs = self._get_main_kwargs(key)
 201         self.assertEqual(self.test.main(**kwargs),
 202                          testcase_base.TestcaseBase.EX_RUN_ERROR)
 203
 204     def test_main_missing_odlusername(self):
 205         self._test_main_missing_keyword('odlusername')
 206
 207     def test_main_missing_odlpassword(self):
 208         self._test_main_missing_keyword('odlpassword')
 209
 210     def test_main_missing_keystoneip(self):
 211         self._test_main_missing_keyword('keystoneip')
 212
 213     def test_main_missing_neutronip(self):
 214         self._test_main_missing_keyword('neutronip')
 215
 216     def test_main_missing_osusername(self):
 217         self._test_main_missing_keyword('osusername')
 218
 219     def test_main_missing_ostenantname(self):
 220         self._test_main_missing_keyword('ostenantname')
 221
 222     def test_main_missing_ospassword(self):
 223         self._test_main_missing_keyword('ospassword')
 224
 225     def test_main_missing_odlip(self):
 226         self._test_main_missing_keyword('odlip')
 227
 228     def test_main_missing_odlwebport(self):
 229         self._test_main_missing_keyword('odlwebport')
 230
 231     def test_main_missing_odlrestconfport(self):
 232         self._test_main_missing_keyword('odlrestconfport')
 233
 234     def test_main_set_robotframework_vars_failed(self):
 235         with mock.patch.object(self.test, 'set_robotframework_vars',
 236                                return_value=False):
 237             self._test_main(testcase_base.TestcaseBase.EX_RUN_ERROR)
 238             self.test.set_robotframework_vars.assert_called_once_with(
 239                 self._odl_username, self._odl_password)
 240
 241     @mock.patch('os.makedirs', side_effect=Exception)
 242     def test_main_makedirs_exception(self, mock_method):
 243         with mock.patch.object(self.test, 'set_robotframework_vars',
 244                                return_value=True), \
 245                 self.assertRaises(Exception):
 246             self._test_main(testcase_base.TestcaseBase.EX_RUN_ERROR,
 247                             mock_method)
 248
 249     @mock.patch('os.makedirs', side_effect=OSError)
 250     def test_main_makedirs_oserror(self, mock_method):
 251         with mock.patch.object(self.test, 'set_robotframework_vars',
 252                                return_value=True):
 253             self._test_main(testcase_base.TestcaseBase.EX_RUN_ERROR,
 254                             mock_method)
 255
 256     @mock.patch('robot.run', side_effect=RobotError)
 257     @mock.patch('os.makedirs')
 258     def test_main_robot_run_failed(self, *args):
 259         with mock.patch.object(self.test, 'set_robotframework_vars',
 260                                return_value=True), \
 261                 mock.patch.object(odl, 'open', mock.mock_open(),
 262                                   create=True), \
 263                 self.assertRaises(RobotError):
 264             self._test_main(testcase_base.TestcaseBase.EX_RUN_ERROR, *args)
 265
 266     @mock.patch('robot.run')
 267     @mock.patch('os.makedirs')
 268     def test_main_parse_results_failed(self, *args):
 269         with mock.patch.object(self.test, 'set_robotframework_vars',
 270                                return_value=True), \
 271                 mock.patch.object(odl, 'open', mock.mock_open(),
 272                                   create=True), \
 273                 mock.patch.object(self.test, 'parse_results',
 274                                   side_effect=RobotError):
 275             self._test_main(testcase_base.TestcaseBase.EX_RUN_ERROR, *args)
 276
 277     @mock.patch('os.remove', side_effect=Exception)
 278     @mock.patch('robot.run')
 279     @mock.patch('os.makedirs')
 280     def test_main_remove_exception(self, *args):
 281         with mock.patch.object(self.test, 'set_robotframework_vars',
 282                                return_value=True), \
 283                 mock.patch.object(self.test, 'parse_results'), \
 284                 self.assertRaises(Exception):
 285             self._test_main(testcase_base.TestcaseBase.EX_OK, *args)
 286
 287     @mock.patch('os.remove')
 288     @mock.patch('robot.run')
 289     @mock.patch('os.makedirs')
 290     def test_main(self, *args):
 291         with mock.patch.object(self.test, 'set_robotframework_vars',
 292                                return_value=True), \
 293                 mock.patch.object(odl, 'open', mock.mock_open(),
 294                                   create=True), \
 295                 mock.patch.object(self.test, 'parse_results'):
 296             self._test_main(testcase_base.TestcaseBase.EX_OK, *args)
 297
 298     @mock.patch('os.remove')
 299     @mock.patch('robot.run')
 300     @mock.patch('os.makedirs', side_effect=OSError(errno.EEXIST, ''))
 301     def test_main_makedirs_oserror17(self, *args):
 302         with mock.patch.object(self.test, 'set_robotframework_vars',
 303                                return_value=True), \
 304                 mock.patch.object(odl, 'open', mock.mock_open(),
 305                                   create=True), \
 306                 mock.patch.object(self.test, 'parse_results'):
 307             self._test_main(testcase_base.TestcaseBase.EX_OK, *args)
 308
 309     @mock.patch('os.remove')
 310     @mock.patch('robot.run', return_value=1)
 311     @mock.patch('os.makedirs')
 312     def test_main_testcases_in_failure(self, *args):
 313         with mock.patch.object(self.test, 'set_robotframework_vars',
 314                                return_value=True), \
 315                 mock.patch.object(odl, 'open', mock.mock_open(),
 316                                   create=True), \
 317                 mock.patch.object(self.test, 'parse_results'):
 318             self._test_main(testcase_base.TestcaseBase.EX_OK, *args)
 319
 320     @mock.patch('os.remove', side_effect=OSError)
 321     @mock.patch('robot.run')
 322     @mock.patch('os.makedirs')
 323     def test_main_remove_oserror(self, *args):
 324         with mock.patch.object(self.test, 'set_robotframework_vars',
 325                                return_value=True), \
 326                 mock.patch.object(odl, 'open', mock.mock_open(),
 327                                   create=True), \
 328                 mock.patch.object(self.test, 'parse_results'):
 329             self._test_main(testcase_base.TestcaseBase.EX_OK, *args)
 330
 331     def _test_run_missing_env_var(self, var):
 332         with mock.patch('functest.utils.openstack_utils.get_endpoint',
 333                         side_effect=self._fake_url_for):
 334             del os.environ[var]
 335             self.assertEqual(self.test.run(),
 336                              testcase_base.TestcaseBase.EX_RUN_ERROR)
 337
 338     def _test_run(self, status=testcase_base.TestcaseBase.EX_OK,
 339                   exception=None, odlip="127.0.0.3", odlwebport="8080",
 340                   odlrestconfport="8181"):
 341         with mock.patch('functest.utils.openstack_utils.get_endpoint',
 342                         side_effect=self._fake_url_for):
 343             if exception:
 344                 self.test.main = mock.Mock(side_effect=exception)
 345             else:
 346                 self.test.main = mock.Mock(return_value=status)
 347             self.assertEqual(self.test.run(), status)
 348             self.test.main.assert_called_once_with(
 349                 odl.ODLTests.default_suites,
 350                 keystoneip=self._keystone_ip, neutronip=self._neutron_ip,
 351                 odlip=odlip, odlpassword=self._odl_password,
 352                 odlrestconfport=odlrestconfport,
 353                 odlusername=self._odl_username, odlwebport=odlwebport,
 354                 ospassword=self._os_password, ostenantname=self._os_tenantname,
 355                 osusername=self._os_username)
 356
 357     def _test_run_defining_multiple_suites(
 358             self, suites,
 359             status=testcase_base.TestcaseBase.EX_OK,
 360             exception=None, odlip="127.0.0.3", odlwebport="8080",
 361             odlrestconfport="8181"):
 362         with mock.patch('functest.utils.openstack_utils.get_endpoint',
 363                         side_effect=self._fake_url_for):
 364             if exception:
 365                 self.test.main = mock.Mock(side_effect=exception)
 366             else:
 367                 self.test.main = mock.Mock(return_value=status)
 368             self.assertEqual(self.test.run(suites=suites), status)
 369             self.test.main.assert_called_once_with(
 370                 suites,
 371                 keystoneip=self._keystone_ip, neutronip=self._neutron_ip,
 372                 odlip=odlip, odlpassword=self._odl_password,
 373                 odlrestconfport=odlrestconfport,
 374                 odlusername=self._odl_username, odlwebport=odlwebport,
 375                 ospassword=self._os_password, ostenantname=self._os_tenantname,
 376                 osusername=self._os_username)
 377
 378     def test_run_exception(self):
 379         with mock.patch('functest.utils.openstack_utils.get_endpoint',
 380                         side_effect=auth_plugins.MissingAuthPlugin()):
 381             self.assertEqual(self.test.run(),
 382                              testcase_base.TestcaseBase.EX_RUN_ERROR)
 383
 384     def test_run_missing_os_username(self):
 385         self._test_run_missing_env_var("OS_USERNAME")
 386
 387     def test_run_missing_os_password(self):
 388         self._test_run_missing_env_var("OS_PASSWORD")
 389
 390     def test_run_missing_os_tenant_name(self):
 391         self._test_run_missing_env_var("OS_TENANT_NAME")
 392
 393     def test_run_main_false(self):
 394         os.environ["SDN_CONTROLLER_IP"] = self._sdn_controller_ip
 395         self._test_run(testcase_base.TestcaseBase.EX_RUN_ERROR,
 396                        odlip=self._sdn_controller_ip,
 397                        odlwebport=self._odl_webport)
 398
 399     def test_run_main_exception(self):
 400         with self.assertRaises(Exception):
 401             os.environ["SDN_CONTROLLER_IP"] = self._sdn_controller_ip
 402             self._test_run(status=testcase_base.TestcaseBase.EX_RUN_ERROR,
 403                            exception=Exception(),
 404                            odlip=self._sdn_controller_ip,
 405                            odlwebport=self._odl_webport)
 406
 407     def test_run_missing_sdn_controller_ip(self):
 408         with mock.patch('functest.utils.openstack_utils.get_endpoint',
 409                         side_effect=self._fake_url_for):
 410             self.assertEqual(self.test.run(),
 411                              testcase_base.TestcaseBase.EX_RUN_ERROR)
 412
 413     def test_run_without_installer_type(self):
 414         os.environ["SDN_CONTROLLER_IP"] = self._sdn_controller_ip
 415         self._test_run(testcase_base.TestcaseBase.EX_OK,
 416                        odlip=self._sdn_controller_ip,
 417                        odlwebport=self._odl_webport)
 418
 419     def test_run_redefining_suites(self):
 420         os.environ["SDN_CONTROLLER_IP"] = self._sdn_controller_ip
 421         self._test_run_defining_multiple_suites(
 422             [odl.ODLTests.basic_suite_dir],
 423             testcase_base.TestcaseBase.EX_OK,
 424             odlip=self._sdn_controller_ip,
 425             odlwebport=self._odl_webport)
 426
 427     def test_run_fuel(self):
 428         os.environ["INSTALLER_TYPE"] = "fuel"
 429         self._test_run(testcase_base.TestcaseBase.EX_OK,
 430                        odlip=self._neutron_ip, odlwebport='8282')
 431
 432     def test_run_apex_missing_sdn_controller_ip(self):
 433         with mock.patch('functest.utils.openstack_utils.get_endpoint',
 434                         side_effect=self._fake_url_for):
 435             os.environ["INSTALLER_TYPE"] = "apex"
 436             self.assertEqual(self.test.run(),
 437                              testcase_base.TestcaseBase.EX_RUN_ERROR)
 438
 439     def test_run_apex(self):
 440         os.environ["SDN_CONTROLLER_IP"] = self._sdn_controller_ip
 441         os.environ["INSTALLER_TYPE"] = "apex"
 442         self._test_run(testcase_base.TestcaseBase.EX_OK,
 443                        odlip=self._sdn_controller_ip, odlwebport='8081',
 444                        odlrestconfport='8081')
 445
 446     def test_run_joid_missing_sdn_controller(self):
 447         with mock.patch('functest.utils.openstack_utils.get_endpoint',
 448                         side_effect=self._fake_url_for):
 449             os.environ["INSTALLER_TYPE"] = "joid"
 450             self.assertEqual(self.test.run(),
 451                              testcase_base.TestcaseBase.EX_RUN_ERROR)
 452
 453     def test_run_joid(self):
 454         os.environ["SDN_CONTROLLER"] = self._sdn_controller_ip
 455         os.environ["INSTALLER_TYPE"] = "joid"
 456         self._test_run(testcase_base.TestcaseBase.EX_OK,
 457                        odlip=self._sdn_controller_ip, odlwebport='8080')
 458
 459     def test_run_compass(self, *args):
 460         os.environ["INSTALLER_TYPE"] = "compass"
 461         self._test_run(testcase_base.TestcaseBase.EX_OK,
 462                        odlip=self._neutron_ip, odlwebport='8181')
 463
 464     def test_argparser_default(self):
 465         parser = odl.ODLParser()
 466         self.assertEqual(parser.parse_args(), self.defaultargs)
 467
 468     def test_argparser_basic(self):
 469         self.defaultargs['neutronip'] = self._neutron_ip
 470         self.defaultargs['odlip'] = self._sdn_controller_ip
 471         parser = odl.ODLParser()
 472         self.assertEqual(parser.parse_args(
 473             ["--neutronip={}".format(self._neutron_ip),
 474              "--odlip={}".format(self._sdn_controller_ip)
 475              ]), self.defaultargs)
 476
 477     @mock.patch('sys.stderr', new_callable=StringIO.StringIO)
 478     def test_argparser_fail(self, *args):
 479         self.defaultargs['foo'] = 'bar'
 480         parser = odl.ODLParser()
 481         with self.assertRaises(SystemExit):
 482             parser.parse_args(["--foo=bar"])
 483
 484     def _test_argparser(self, arg, value):
 485         self.defaultargs[arg] = value
 486         parser = odl.ODLParser()
 487         self.assertEqual(parser.parse_args(["--{}={}".format(arg, value)]),
 488                          self.defaultargs)
 489
 490     def test_argparser_odlusername(self):
 491         self._test_argparser('odlusername', 'foo')
 492
 493     def test_argparser_odlpassword(self):
 494         self._test_argparser('odlpassword', 'foo')
 495
 496     def test_argparser_keystoneip(self):
 497         self._test_argparser('keystoneip', '127.0.0.4')
 498
 499     def test_argparser_neutronip(self):
 500         self._test_argparser('neutronip', '127.0.0.4')
 501
 502     def test_argparser_osusername(self):
 503         self._test_argparser('osusername', 'foo')
 504
 505     def test_argparser_ostenantname(self):
 506         self._test_argparser('ostenantname', 'foo')
 507
 508     def test_argparser_ospassword(self):
 509         self._test_argparser('ospassword', 'foo')
 510
 511     def test_argparser_odlip(self):
 512         self._test_argparser('odlip', '127.0.0.4')
 513
 514     def test_argparser_odlwebport(self):
 515         self._test_argparser('odlwebport', '80')
 516
 517     def test_argparser_odlrestconfport(self):
 518         self._test_argparser('odlrestconfport', '80')
 519
 520     def test_argparser_pushtodb(self):
 521         self.defaultargs['pushtodb'] = True
 522         parser = odl.ODLParser()
 523         self.assertEqual(parser.parse_args(["--{}".format('pushtodb')]),
 524                          self.defaultargs)
 525
 526     def test_argparser_multiple_args(self):
 527         self.defaultargs['neutronip'] = self._neutron_ip
 528         self.defaultargs['odlip'] = self._sdn_controller_ip
 529         parser = odl.ODLParser()
 530         self.assertEqual(parser.parse_args(
 531             ["--neutronip={}".format(self._neutron_ip),
 532              "--odlip={}".format(self._sdn_controller_ip)
 533              ]), self.defaultargs)
 534
 535
 536 if __name__ == "__main__":
 537     unittest.main(verbosity=2)