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