Use StringIO in odl.py
[functest.git] / 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 """Define the classes required to fully cover odl."""
11
12 import errno
13 import logging
14 import os
15 import unittest
16
17 from keystoneauth1.exceptions import auth_plugins
18 import mock
19 from robot.errors import DataError, RobotError
20 from robot.result import model
21 from robot.utils.robottime import timestamp_to_secs
22 import six
23
24 from functest.core import testcase
25 from functest.opnfv_tests.sdn.odl import odl
26
27 __author__ = "Cedric Ollivier <cedric.ollivier@orange.com>"
28
29
30 class ODLVisitorTesting(unittest.TestCase):
31
32     """The class testing ODLResultVisitor."""
33     # pylint: disable=missing-docstring
34
35     logging.disable(logging.CRITICAL)
36
37     def setUp(self):
38         self.visitor = odl.ODLResultVisitor()
39
40     def test_empty(self):
41         self.assertFalse(self.visitor.get_data())
42
43     def test_ok(self):
44         data = {'name': 'foo',
45                 'parent': 'bar',
46                 'status': 'PASS',
47                 'starttime': "20161216 16:00:00.000",
48                 'endtime': "20161216 16:00:01.000",
49                 'elapsedtime': 1000,
50                 'text': 'Hello, World!',
51                 'critical': True}
52         test = model.TestCase(
53             name=data['name'], status=data['status'], message=data['text'],
54             starttime=data['starttime'], endtime=data['endtime'])
55         test.parent = mock.Mock()
56         config = {'name': data['parent'],
57                   'criticality.test_is_critical.return_value': data[
58                       'critical']}
59         test.parent.configure_mock(**config)
60         self.visitor.visit_test(test)
61         self.assertEqual(self.visitor.get_data(), [data])
62
63
64 class ODLTesting(unittest.TestCase):
65
66     """The super class which testing classes could inherit."""
67     # pylint: disable=missing-docstring
68
69     logging.disable(logging.CRITICAL)
70
71     _keystone_ip = "127.0.0.1"
72     _neutron_ip = "127.0.0.2"
73     _sdn_controller_ip = "127.0.0.3"
74     _os_auth_url = "http://{}:5000/v2.0".format(_keystone_ip)
75     _os_tenantname = "admin"
76     _os_username = "admin"
77     _os_password = "admin"
78     _odl_webport = "8080"
79     _odl_restconfport = "8181"
80     _odl_username = "admin"
81     _odl_password = "admin"
82
83     def setUp(self):
84         for var in ("INSTALLER_TYPE", "SDN_CONTROLLER", "SDN_CONTROLLER_IP"):
85             if var in os.environ:
86                 del os.environ[var]
87         os.environ["OS_AUTH_URL"] = self._os_auth_url
88         os.environ["OS_USERNAME"] = self._os_username
89         os.environ["OS_PASSWORD"] = self._os_password
90         os.environ["OS_TENANT_NAME"] = self._os_tenantname
91         self.test = odl.ODLTests(case_name='odl', project_name='functest')
92         self.defaultargs = {'odlusername': self._odl_username,
93                             'odlpassword': self._odl_password,
94                             'neutronip': self._keystone_ip,
95                             'osauthurl': self._os_auth_url,
96                             'osusername': self._os_username,
97                             'ostenantname': self._os_tenantname,
98                             'ospassword': self._os_password,
99                             'odlip': self._keystone_ip,
100                             'odlwebport': self._odl_webport,
101                             'odlrestconfport': self._odl_restconfport,
102                             'pushtodb': False}
103
104
105 class ODLParseResultTesting(ODLTesting):
106
107     """The class testing ODLTests.parse_results()."""
108     # pylint: disable=missing-docstring
109
110     _config = {'name': 'dummy', 'starttime': '20161216 16:00:00.000',
111                'endtime': '20161216 16:00:01.000'}
112
113     @mock.patch('robot.api.ExecutionResult', side_effect=DataError)
114     def test_raises_exc(self, mock_method):
115         with self.assertRaises(DataError):
116             self.test.parse_results()
117         mock_method.assert_called_once_with(
118             os.path.join(odl.ODLTests.res_dir, 'output.xml'))
119
120     def _test_result(self, config, result):
121         suite = mock.Mock()
122         suite.configure_mock(**config)
123         with mock.patch('robot.api.ExecutionResult',
124                         return_value=mock.Mock(suite=suite)):
125             self.test.parse_results()
126             self.assertEqual(self.test.result, result)
127             self.assertEqual(self.test.start_time,
128                              timestamp_to_secs(config['starttime']))
129             self.assertEqual(self.test.stop_time,
130                              timestamp_to_secs(config['endtime']))
131             self.assertEqual(self.test.details,
132                              {'description': config['name'], 'tests': []})
133
134     def test_null_passed(self):
135         self._config.update({'statistics.critical.passed': 0,
136                              'statistics.critical.total': 20})
137         self._test_result(self._config, 0)
138
139     def test_no_test(self):
140         self._config.update({'statistics.critical.passed': 20,
141                              'statistics.critical.total': 0})
142         self._test_result(self._config, 0)
143
144     def test_half_success(self):
145         self._config.update({'statistics.critical.passed': 10,
146                              'statistics.critical.total': 20})
147         self._test_result(self._config, 50)
148
149     def test_success(self):
150         self._config.update({'statistics.critical.passed': 20,
151                              'statistics.critical.total': 20})
152         self._test_result(self._config, 100)
153
154
155 class ODLRobotTesting(ODLTesting):
156
157     """The class testing ODLTests.set_robotframework_vars()."""
158     # pylint: disable=missing-docstring
159
160     @mock.patch('fileinput.input', side_effect=Exception())
161     def test_set_vars_ko(self, mock_method):
162         self.assertFalse(self.test.set_robotframework_vars())
163         mock_method.assert_called_once_with(
164             os.path.join(odl.ODLTests.odl_test_repo,
165                          'csit/variables/Variables.py'), inplace=True)
166
167     @mock.patch('fileinput.input', return_value=[])
168     def test_set_vars_empty(self, mock_method):
169         self.assertTrue(self.test.set_robotframework_vars())
170         mock_method.assert_called_once_with(
171             os.path.join(odl.ODLTests.odl_test_repo,
172                          'csit/variables/Variables.py'), inplace=True)
173
174     @mock.patch('sys.stdout', new_callable=six.StringIO)
175     def _test_set_vars(self, msg1, msg2, *args):
176         line = mock.MagicMock()
177         line.__iter__.return_value = [msg1]
178         with mock.patch('fileinput.input', return_value=line) as mock_method:
179             self.assertTrue(self.test.set_robotframework_vars())
180             mock_method.assert_called_once_with(
181                 os.path.join(odl.ODLTests.odl_test_repo,
182                              'csit/variables/Variables.py'), inplace=True)
183             self.assertEqual(args[0].getvalue(), "{}\n".format(msg2))
184
185     def test_set_vars_auth_default(self):
186         self._test_set_vars("AUTH = []",
187                             "AUTH = [u'admin', u'admin']")
188
189     def test_set_vars_auth1(self):
190         self._test_set_vars("AUTH1 = []", "AUTH1 = []")
191
192     @mock.patch('sys.stdout', new_callable=six.StringIO)
193     def test_set_vars_auth_foo(self, *args):
194         line = mock.MagicMock()
195         line.__iter__.return_value = ["AUTH = []"]
196         with mock.patch('fileinput.input', return_value=line) as mock_method:
197             self.assertTrue(self.test.set_robotframework_vars('foo', 'bar'))
198             mock_method.assert_called_once_with(
199                 os.path.join(odl.ODLTests.odl_test_repo,
200                              'csit/variables/Variables.py'), inplace=True)
201             self.assertEqual(args[0].getvalue(),
202                              "AUTH = [u'{}', u'{}']\n".format('foo', 'bar'))
203
204
205 class ODLMainTesting(ODLTesting):
206
207     """The class testing ODLTests.main()."""
208     # pylint: disable=missing-docstring
209
210     def _get_main_kwargs(self, key=None):
211         kwargs = {'odlusername': self._odl_username,
212                   'odlpassword': self._odl_password,
213                   'neutronip': self._neutron_ip,
214                   'osauthurl': self._os_auth_url,
215                   'osusername': self._os_username,
216                   'ostenantname': self._os_tenantname,
217                   'ospassword': self._os_password,
218                   'odlip': self._sdn_controller_ip,
219                   'odlwebport': self._odl_webport,
220                   'odlrestconfport': self._odl_restconfport}
221         if key:
222             del kwargs[key]
223         return kwargs
224
225     def _test_main(self, status, *args):
226         kwargs = self._get_main_kwargs()
227         self.assertEqual(self.test.main(**kwargs), status)
228         if len(args) > 0:
229             args[0].assert_called_once_with(
230                 odl.ODLTests.res_dir)
231         if len(args) > 1:
232             variable = ['KEYSTONE:{}'.format(self._keystone_ip),
233                         'NEUTRON:{}'.format(self._neutron_ip),
234                         'OS_AUTH_URL:"{}"'.format(self._os_auth_url),
235                         'OSUSERNAME:"{}"'.format(self._os_username),
236                         'OSTENANTNAME:"{}"'.format(self._os_tenantname),
237                         'OSPASSWORD:"{}"'.format(self._os_password),
238                         'ODL_SYSTEM_IP:{}'.format(self._sdn_controller_ip),
239                         'PORT:{}'.format(self._odl_webport),
240                         'RESTCONFPORT:{}'.format(self._odl_restconfport)]
241             args[1].assert_called_once_with(
242                 odl.ODLTests.basic_suite_dir,
243                 odl.ODLTests.neutron_suite_dir,
244                 log='NONE',
245                 output=os.path.join(odl.ODLTests.res_dir, 'output.xml'),
246                 report='NONE',
247                 stdout=mock.ANY,
248                 variable=variable)
249         if len(args) > 2:
250             args[2].assert_called_with(
251                 os.path.join(odl.ODLTests.res_dir, 'stdout.txt'))
252
253     def _test_no_keyword(self, key):
254         kwargs = self._get_main_kwargs(key)
255         self.assertEqual(self.test.main(**kwargs),
256                          testcase.TestCase.EX_RUN_ERROR)
257
258     def test_no_odlusername(self):
259         self._test_no_keyword('odlusername')
260
261     def test_no_odlpassword(self):
262         self._test_no_keyword('odlpassword')
263
264     def test_no_neutronip(self):
265         self._test_no_keyword('neutronip')
266
267     def test_no_osauthurl(self):
268         self._test_no_keyword('osauthurl')
269
270     def test_no_osusername(self):
271         self._test_no_keyword('osusername')
272
273     def test_no_ostenantname(self):
274         self._test_no_keyword('ostenantname')
275
276     def test_no_ospassword(self):
277         self._test_no_keyword('ospassword')
278
279     def test_no_odlip(self):
280         self._test_no_keyword('odlip')
281
282     def test_no_odlwebport(self):
283         self._test_no_keyword('odlwebport')
284
285     def test_no_odlrestconfport(self):
286         self._test_no_keyword('odlrestconfport')
287
288     def test_set_vars_ko(self):
289         with mock.patch.object(self.test, 'set_robotframework_vars',
290                                return_value=False) as mock_object:
291             self._test_main(testcase.TestCase.EX_RUN_ERROR)
292             mock_object.assert_called_once_with(
293                 self._odl_username, self._odl_password)
294
295     @mock.patch('os.makedirs', side_effect=Exception)
296     def test_makedirs_exc(self, mock_method):
297         with mock.patch.object(self.test, 'set_robotframework_vars',
298                                return_value=True), \
299                 self.assertRaises(Exception):
300             self._test_main(testcase.TestCase.EX_RUN_ERROR,
301                             mock_method)
302
303     @mock.patch('os.makedirs', side_effect=OSError)
304     def test_makedirs_oserror(self, mock_method):
305         with mock.patch.object(self.test, 'set_robotframework_vars',
306                                return_value=True):
307             self._test_main(testcase.TestCase.EX_RUN_ERROR,
308                             mock_method)
309
310     @mock.patch('robot.run', side_effect=RobotError)
311     @mock.patch('os.makedirs')
312     def test_run_ko(self, *args):
313         with mock.patch.object(self.test, 'set_robotframework_vars',
314                                return_value=True), \
315                 self.assertRaises(RobotError):
316             self._test_main(testcase.TestCase.EX_RUN_ERROR, *args)
317
318     @mock.patch('robot.run')
319     @mock.patch('os.makedirs')
320     def test_parse_results_ko(self, *args):
321         with mock.patch.object(self.test, 'set_robotframework_vars',
322                                return_value=True), \
323                 mock.patch.object(self.test, 'parse_results',
324                                   side_effect=RobotError):
325             self._test_main(testcase.TestCase.EX_RUN_ERROR, *args)
326
327     @mock.patch('robot.run')
328     @mock.patch('os.makedirs')
329     def test_ok(self, *args):
330         with mock.patch.object(self.test, 'set_robotframework_vars',
331                                return_value=True), \
332                 mock.patch.object(self.test, 'parse_results'):
333             self._test_main(testcase.TestCase.EX_OK, *args)
334
335     @mock.patch('robot.run')
336     @mock.patch('os.makedirs', side_effect=OSError(errno.EEXIST, ''))
337     def test_makedirs_oserror17(self, *args):
338         with mock.patch.object(self.test, 'set_robotframework_vars',
339                                return_value=True), \
340                 mock.patch.object(self.test, 'parse_results'):
341             self._test_main(testcase.TestCase.EX_OK, *args)
342
343     @mock.patch('robot.run', return_value=1)
344     @mock.patch('os.makedirs')
345     def test_testcases_in_failure(self, *args):
346         with mock.patch.object(self.test, 'set_robotframework_vars',
347                                return_value=True), \
348                 mock.patch.object(self.test, 'parse_results'):
349             self._test_main(testcase.TestCase.EX_OK, *args)
350
351
352 class ODLRunTesting(ODLTesting):
353
354     """The class testing ODLTests.run()."""
355     # pylint: disable=missing-docstring
356
357     def _test_no_env_var(self, var):
358         with mock.patch('functest.utils.openstack_utils.get_endpoint',
359                         return_value="http://{}:9696".format(
360                             ODLTesting._neutron_ip)):
361             del os.environ[var]
362             self.assertEqual(self.test.run(),
363                              testcase.TestCase.EX_RUN_ERROR)
364
365     def _test_run(self, status=testcase.TestCase.EX_OK,
366                   exception=None, **kwargs):
367         odlip = kwargs['odlip'] if 'odlip' in kwargs else '127.0.0.3'
368         odlwebport = kwargs['odlwebport'] if 'odlwebport' in kwargs else '8080'
369         odlrestconfport = (kwargs['odlrestconfport']
370                            if 'odlrestconfport' in kwargs else '8181')
371
372         with mock.patch('functest.utils.openstack_utils.get_endpoint',
373                         return_value="http://{}:9696".format(
374                             ODLTesting._neutron_ip)):
375             if exception:
376                 self.test.main = mock.Mock(side_effect=exception)
377             else:
378                 self.test.main = mock.Mock(return_value=status)
379             self.assertEqual(self.test.run(), status)
380             self.test.main.assert_called_once_with(
381                 odl.ODLTests.default_suites,
382                 neutronip=self._neutron_ip,
383                 odlip=odlip, odlpassword=self._odl_password,
384                 odlrestconfport=odlrestconfport,
385                 odlusername=self._odl_username, odlwebport=odlwebport,
386                 osauthurl=self._os_auth_url,
387                 ospassword=self._os_password, ostenantname=self._os_tenantname,
388                 osusername=self._os_username)
389
390     def _test_multiple_suites(self, suites,
391                               status=testcase.TestCase.EX_OK, **kwargs):
392         odlip = kwargs['odlip'] if 'odlip' in kwargs else '127.0.0.3'
393         odlwebport = kwargs['odlwebport'] if 'odlwebport' in kwargs else '8080'
394         odlrestconfport = (kwargs['odlrestconfport']
395                            if 'odlrestconfport' in kwargs else '8181')
396         with mock.patch('functest.utils.openstack_utils.get_endpoint',
397                         return_value="http://{}:9696".format(
398                             ODLTesting._neutron_ip)):
399             self.test.main = mock.Mock(return_value=status)
400             self.assertEqual(self.test.run(suites=suites), status)
401             self.test.main.assert_called_once_with(
402                 suites,
403                 neutronip=self._neutron_ip,
404                 odlip=odlip, odlpassword=self._odl_password,
405                 odlrestconfport=odlrestconfport,
406                 odlusername=self._odl_username, odlwebport=odlwebport,
407                 osauthurl=self._os_auth_url,
408                 ospassword=self._os_password, ostenantname=self._os_tenantname,
409                 osusername=self._os_username)
410
411     def test_exc(self):
412         with mock.patch('functest.utils.openstack_utils.get_endpoint',
413                         side_effect=auth_plugins.MissingAuthPlugin()):
414             self.assertEqual(self.test.run(),
415                              testcase.TestCase.EX_RUN_ERROR)
416
417     def test_no_os_auth_url(self):
418         self._test_no_env_var("OS_AUTH_URL")
419
420     def test_no_os_username(self):
421         self._test_no_env_var("OS_USERNAME")
422
423     def test_no_os_password(self):
424         self._test_no_env_var("OS_PASSWORD")
425
426     def test_no_os_tenant_name(self):
427         self._test_no_env_var("OS_TENANT_NAME")
428
429     def test_main_false(self):
430         os.environ["SDN_CONTROLLER_IP"] = self._sdn_controller_ip
431         self._test_run(testcase.TestCase.EX_RUN_ERROR,
432                        odlip=self._sdn_controller_ip,
433                        odlwebport=self._odl_webport)
434
435     def test_main_exc(self):
436         with self.assertRaises(Exception):
437             os.environ["SDN_CONTROLLER_IP"] = self._sdn_controller_ip
438             self._test_run(status=testcase.TestCase.EX_RUN_ERROR,
439                            exception=Exception(),
440                            odlip=self._sdn_controller_ip,
441                            odlwebport=self._odl_webport)
442
443     def test_no_sdn_controller_ip(self):
444         with mock.patch('functest.utils.openstack_utils.get_endpoint',
445                         return_value="http://{}:9696".format(
446                             ODLTesting._neutron_ip)):
447             self.assertEqual(self.test.run(),
448                              testcase.TestCase.EX_RUN_ERROR)
449
450     def test_without_installer_type(self):
451         os.environ["SDN_CONTROLLER_IP"] = self._sdn_controller_ip
452         self._test_run(testcase.TestCase.EX_OK,
453                        odlip=self._sdn_controller_ip,
454                        odlwebport=self._odl_webport)
455
456     def test_suites(self):
457         os.environ["SDN_CONTROLLER_IP"] = self._sdn_controller_ip
458         self._test_multiple_suites(
459             [odl.ODLTests.basic_suite_dir],
460             testcase.TestCase.EX_OK,
461             odlip=self._sdn_controller_ip,
462             odlwebport=self._odl_webport)
463
464     def test_fuel(self):
465         os.environ["INSTALLER_TYPE"] = "fuel"
466         self._test_run(testcase.TestCase.EX_OK,
467                        odlip=self._neutron_ip, odlwebport='8282')
468
469     def test_apex_no_controller_ip(self):
470         with mock.patch('functest.utils.openstack_utils.get_endpoint',
471                         return_value="http://{}:9696".format(
472                             ODLTesting._neutron_ip)):
473             os.environ["INSTALLER_TYPE"] = "apex"
474             self.assertEqual(self.test.run(),
475                              testcase.TestCase.EX_RUN_ERROR)
476
477     def test_apex(self):
478         os.environ["SDN_CONTROLLER_IP"] = self._sdn_controller_ip
479         os.environ["INSTALLER_TYPE"] = "apex"
480         self._test_run(testcase.TestCase.EX_OK,
481                        odlip=self._sdn_controller_ip, odlwebport='8081',
482                        odlrestconfport='8081')
483
484     def test_netvirt_no_controller_ip(self):
485         with mock.patch('functest.utils.openstack_utils.get_endpoint',
486                         return_value="http://{}:9696".format(
487                             ODLTesting._neutron_ip)):
488             os.environ["INSTALLER_TYPE"] = "netvirt"
489             self.assertEqual(self.test.run(),
490                              testcase.TestCase.EX_RUN_ERROR)
491
492     def test_netvirt(self):
493         os.environ["SDN_CONTROLLER_IP"] = self._sdn_controller_ip
494         os.environ["INSTALLER_TYPE"] = "netvirt"
495         self._test_run(testcase.TestCase.EX_OK,
496                        odlip=self._sdn_controller_ip, odlwebport='8081',
497                        odlrestconfport='8081')
498
499     def test_joid_no_controller_ip(self):
500         with mock.patch('functest.utils.openstack_utils.get_endpoint',
501                         return_value="http://{}:9696".format(
502                             ODLTesting._neutron_ip)):
503             os.environ["INSTALLER_TYPE"] = "joid"
504             self.assertEqual(self.test.run(),
505                              testcase.TestCase.EX_RUN_ERROR)
506
507     def test_joid(self):
508         os.environ["SDN_CONTROLLER"] = self._sdn_controller_ip
509         os.environ["INSTALLER_TYPE"] = "joid"
510         self._test_run(testcase.TestCase.EX_OK,
511                        odlip=self._sdn_controller_ip, odlwebport='8080')
512
513     def test_compass(self):
514         os.environ["INSTALLER_TYPE"] = "compass"
515         self._test_run(testcase.TestCase.EX_OK,
516                        odlip=self._neutron_ip, odlwebport='8181')
517
518
519 class ODLArgParserTesting(ODLTesting):
520
521     """The class testing ODLParser."""
522     # pylint: disable=missing-docstring
523
524     def setUp(self):
525         self.parser = odl.ODLParser()
526         super(ODLArgParserTesting, self).setUp()
527
528     def test_default(self):
529         self.assertEqual(self.parser.parse_args(), self.defaultargs)
530
531     def test_basic(self):
532         self.defaultargs['neutronip'] = self._neutron_ip
533         self.defaultargs['odlip'] = self._sdn_controller_ip
534         self.assertEqual(
535             self.parser.parse_args(
536                 ["--neutronip={}".format(self._neutron_ip),
537                  "--odlip={}".format(self._sdn_controller_ip)]),
538             self.defaultargs)
539
540     @mock.patch('sys.stderr', new_callable=six.StringIO)
541     def test_fail(self, mock_method):
542         self.defaultargs['foo'] = 'bar'
543         with self.assertRaises(SystemExit):
544             self.parser.parse_args(["--foo=bar"])
545         self.assertTrue(mock_method.getvalue().startswith("usage:"))
546
547     def _test_arg(self, arg, value):
548         self.defaultargs[arg] = value
549         self.assertEqual(
550             self.parser.parse_args(["--{}={}".format(arg, value)]),
551             self.defaultargs)
552
553     def test_odlusername(self):
554         self._test_arg('odlusername', 'foo')
555
556     def test_odlpassword(self):
557         self._test_arg('odlpassword', 'foo')
558
559     def test_osauthurl(self):
560         self._test_arg('osauthurl', 'http://127.0.0.4:5000/v2')
561
562     def test_neutronip(self):
563         self._test_arg('neutronip', '127.0.0.4')
564
565     def test_osusername(self):
566         self._test_arg('osusername', 'foo')
567
568     def test_ostenantname(self):
569         self._test_arg('ostenantname', 'foo')
570
571     def test_ospassword(self):
572         self._test_arg('ospassword', 'foo')
573
574     def test_odlip(self):
575         self._test_arg('odlip', '127.0.0.4')
576
577     def test_odlwebport(self):
578         self._test_arg('odlwebport', '80')
579
580     def test_odlrestconfport(self):
581         self._test_arg('odlrestconfport', '80')
582
583     def test_pushtodb(self):
584         self.defaultargs['pushtodb'] = True
585         self.assertEqual(self.parser.parse_args(["--{}".format('pushtodb')]),
586                          self.defaultargs)
587
588     def test_multiple_args(self):
589         self.defaultargs['neutronip'] = self._neutron_ip
590         self.defaultargs['odlip'] = self._sdn_controller_ip
591         self.assertEqual(
592             self.parser.parse_args(
593                 ["--neutronip={}".format(self._neutron_ip),
594                  "--odlip={}".format(self._sdn_controller_ip)]),
595             self.defaultargs)
596
597
598 if __name__ == "__main__":
599     unittest.main(verbosity=2)