3 # Copyright (c) 2016 Orange and others.
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
10 # pylint: disable=missing-docstring
20 from functest.utils import functest_utils
23 class FunctestUtilsTesting(unittest.TestCase):
24 # pylint: disable=too-many-instance-attributes,too-many-public-methods
27 test_ip = ['10.1.23.4', '10.1.14.15', '10.1.16.15']
30 self.url = 'http://www.opnfv.org/'
32 self.dest_path = 'test_path'
33 self.repo_path = 'test_repo_path'
34 self.installer = 'test_installer'
35 self.scenario = 'test_scenario'
36 self.build_tag = 'jenkins-functest-fuel-opnfv-jump-2-daily-master-190'
37 self.build_tag_week = 'jenkins-functest-fuel-baremetal-weekly-master-8'
38 self.version = 'master'
39 self.node_name = 'test_node_name'
40 self.project = 'test_project'
41 self.case_name = 'test_case_name'
42 self.status = 'test_status'
43 self.details = 'test_details'
44 self.db_url = 'test_db_url'
47 self.start_date = 1482624000
48 self.stop_date = 1482624000
49 self.start_time = time.time()
50 self.stop_time = time.time()
52 self.test_ip = ['10.1.23.4', '10.1.14.15', '10.1.16.15']
53 self.test_file = 'test_file'
54 self.error_msg = 'test_error_msg'
56 self.output_file = 'test_output_file'
57 self.testname = 'testname'
58 self.parameter = 'general.openstack.image_name'
59 self.config_yaml = pkg_resources.resource_filename(
60 'functest', 'ci/config_functest.yaml')
61 self.db_url_env = 'http://foo/testdb'
62 self.testcases_yaml = "test_testcases_yaml"
63 self.file_yaml = {'general': {'openstack': {'image_name':
66 def _get_env_dict(self, var):
67 dic = {'INSTALLER_TYPE': self.installer,
68 'DEPLOY_SCENARIO': self.scenario,
69 'NODE_NAME': self.node_name,
70 'BUILD_TAG': self.build_tag}
76 if FunctestUtilsTesting.readline == \
77 len(FunctestUtilsTesting.test_ip) - 1:
79 FunctestUtilsTesting.readline += 1
80 return FunctestUtilsTesting.test_ip[FunctestUtilsTesting.readline]
82 def _get_environ(self, var, *args): # pylint: disable=unused-argument
83 if var == 'INSTALLER_TYPE':
85 if var == 'DEPLOY_SCENARIO':
93 @mock.patch('functest.utils.functest_utils.LOGGER.error')
94 @mock.patch('functest.utils.functest_utils.LOGGER.info')
95 def test_exec_cmd_args_present_ko(self, mock_logger_info,
97 with mock.patch('functest.utils.functest_utils.subprocess.Popen') \
98 as mock_subproc_open, \
99 mock.patch('six.moves.builtins.open',
100 mock.mock_open()) as mopen:
101 stream = six.BytesIO()
102 stream.write(self.cmd_readline().encode("utf-8"))
104 'return_value.__enter__.return_value.stdout': stream,
105 'return_value.__enter__.return_value.wait.return_value': 1}
106 mock_subproc_open.configure_mock(**attrs)
107 resp = functest_utils.execute_command(
108 self.cmd, info=True, error_msg=self.error_msg, verbose=True,
109 output_file=self.output_file)
110 self.assertEqual(resp, 1)
111 msg_exec = f"Executing command: '{self.cmd}'"
112 mock_logger_info.assert_called_once_with(msg_exec)
113 mopen.assert_called_once_with(
114 self.output_file, "w", encoding='utf-8')
115 mock_logger_error.assert_called_once_with(self.error_msg)
117 @mock.patch('functest.utils.functest_utils.LOGGER.info')
118 def test_exec_cmd_args_present_ok(self, mock_logger_info):
119 with mock.patch('functest.utils.functest_utils.subprocess.Popen') \
120 as mock_subproc_open, \
121 mock.patch('six.moves.builtins.open',
122 mock.mock_open()) as mopen:
123 stream = six.BytesIO()
124 stream.write(self.cmd_readline().encode("utf-8"))
126 'return_value.__enter__.return_value.stdout': stream,
127 'return_value.__enter__.return_value.wait.return_value': 0}
128 mock_subproc_open.configure_mock(**attrs)
129 resp = functest_utils.execute_command(
130 self.cmd, info=True, error_msg=self.error_msg, verbose=True,
131 output_file=self.output_file)
132 self.assertEqual(resp, 0)
133 msg_exec = (f"Executing command: '{self.cmd}'")
134 mock_logger_info.assert_called_once_with(msg_exec)
135 mopen.assert_called_once_with(
136 self.output_file, "w", encoding='utf-8')
138 @mock.patch('sys.stdout')
139 def test_exec_cmd_args_missing_ok(self, stdout=None):
140 # pylint: disable=unused-argument
141 with mock.patch('functest.utils.functest_utils.subprocess.Popen') \
142 as mock_subproc_open:
143 stream = six.BytesIO()
144 stream.write(self.cmd_readline().encode("utf-8"))
146 'return_value.__enter__.return_value.stdout': stream,
147 'return_value.__enter__.return_value.wait.return_value': 0}
148 mock_subproc_open.configure_mock(**attrs)
149 resp = functest_utils.execute_command(
150 self.cmd, info=False, error_msg="", verbose=False,
152 self.assertEqual(resp, 0)
154 @mock.patch('sys.stdout')
155 def test_exec_cmd_args_missing_ko(self, stdout=None):
156 # pylint: disable=unused-argument
157 with mock.patch('functest.utils.functest_utils.subprocess.Popen') \
158 as mock_subproc_open:
160 stream = six.BytesIO()
161 stream.write(self.cmd_readline().encode("utf-8"))
163 'return_value.__enter__.return_value.stdout': stream,
164 'return_value.__enter__.return_value.wait.return_value': 1}
165 mock_subproc_open.configure_mock(**attrs)
166 resp = functest_utils.execute_command(
167 self.cmd, info=False, error_msg="", verbose=False,
169 self.assertEqual(resp, 1)
171 def test_get_param_from_yaml_failed(self):
172 self.file_yaml['general'] = None
173 with mock.patch('six.moves.builtins.open', mock.mock_open()), \
174 mock.patch('functest.utils.functest_utils.yaml.safe_load') \
176 self.assertRaises(ValueError) as excep:
177 mock_yaml.return_value = self.file_yaml
178 functest_utils.get_parameter_from_yaml(self.parameter,
180 self.assertTrue((f"The parameter {self.parameter} is not"
181 " defined in config_functest.yaml"
182 ) in excep.exception)
184 def test_get_param_from_yaml_def(self):
185 with mock.patch('six.moves.builtins.open', mock.mock_open()), \
186 mock.patch('functest.utils.functest_utils.yaml.safe_load') \
188 mock_yaml.return_value = self.file_yaml
189 self.assertEqual(functest_utils.
190 get_parameter_from_yaml(self.parameter,
194 def test_nova_version_exc1(self):
195 # pylint: disable=protected-access
197 cloud._compute_client.request.return_value = None
198 self.assertEqual(functest_utils.get_nova_version(cloud), None)
199 cloud._compute_client.request.assert_called_once_with('/', 'GET')
201 def test_nova_version_exc2(self):
202 # pylint: disable=protected-access
204 cloud._compute_client.request.return_value = {"version": None}
205 self.assertEqual(functest_utils.get_nova_version(cloud), None)
206 cloud._compute_client.request.assert_called_once_with('/', 'GET')
208 def test_nova_version_exc3(self):
209 # pylint: disable=protected-access
211 cloud._compute_client.request.return_value = {
212 "version": {"version": None}}
213 self.assertEqual(functest_utils.get_nova_version(cloud), None)
214 cloud._compute_client.request.assert_called_once_with('/', 'GET')
216 def test_nova_version_exc4(self):
217 # pylint: disable=protected-access
219 cloud._compute_client.request.return_value = {
220 "version": {"version": "a.b"}}
221 self.assertEqual(functest_utils.get_nova_version(cloud), None)
222 cloud._compute_client.request.assert_called_once_with('/', 'GET')
224 def test_nova_version(self):
225 # pylint: disable=protected-access
227 cloud._compute_client.request.return_value = {
228 "version": {"version": "2.1"}}
229 self.assertEqual(functest_utils.get_nova_version(cloud), (2, 1))
230 cloud._compute_client.request.assert_called_once_with('/', 'GET')
232 @mock.patch('functest.utils.functest_utils.get_nova_version',
233 return_value=(2, 61))
234 def test_openstack_version1(self, *args):
236 self.assertEqual(functest_utils.get_openstack_version(
238 args[0].assert_called_once_with(cloud)
240 @mock.patch('functest.utils.functest_utils.get_nova_version',
241 return_value=(2, 60))
242 def test_openstack_version2(self, *args):
244 self.assertEqual(functest_utils.get_openstack_version(cloud), "Queens")
245 args[0].assert_called_once_with(cloud)
247 @mock.patch('functest.utils.functest_utils.get_nova_version',
248 return_value=(2, 43))
249 def test_openstack_version3(self, *args):
251 self.assertEqual(functest_utils.get_openstack_version(cloud), "Pike")
252 args[0].assert_called_once_with(cloud)
254 @mock.patch('functest.utils.functest_utils.get_nova_version',
255 return_value=(2, 39))
256 def test_openstack_version4(self, *args):
258 self.assertEqual(functest_utils.get_openstack_version(cloud), "Ocata")
259 args[0].assert_called_once_with(cloud)
261 @mock.patch('functest.utils.functest_utils.get_nova_version',
262 return_value=(2, 26))
263 def test_openstack_version5(self, *args):
265 self.assertEqual(functest_utils.get_openstack_version(cloud), "Newton")
266 args[0].assert_called_once_with(cloud)
268 @mock.patch('functest.utils.functest_utils.get_nova_version',
269 return_value=(2, 13))
270 def test_openstack_version6(self, *args):
272 self.assertEqual(functest_utils.get_openstack_version(cloud), "Mitaka")
273 args[0].assert_called_once_with(cloud)
275 @mock.patch('functest.utils.functest_utils.get_nova_version',
277 def test_openstack_version7(self, *args):
280 functest_utils.get_openstack_version(cloud), "Liberty")
281 args[0].assert_called_once_with(cloud)
283 @mock.patch('functest.utils.functest_utils.get_nova_version',
285 def test_openstack_version8(self, *args):
287 self.assertEqual(functest_utils.get_openstack_version(cloud), "Kilo")
288 args[0].assert_called_once_with(cloud)
290 @mock.patch('functest.utils.functest_utils.get_nova_version',
292 def test_openstack_version9(self, *args):
295 functest_utils.get_openstack_version(cloud), "Unknown")
296 args[0].assert_called_once_with(cloud)
298 @mock.patch('functest.utils.functest_utils.get_nova_version',
300 def test_openstack_version10(self, *args):
303 functest_utils.get_openstack_version(cloud), "Master")
304 args[0].assert_called_once_with(cloud)
306 @mock.patch('functest.utils.functest_utils.get_nova_version',
307 return_value=(2, 66))
308 def test_openstack_version11(self, *args):
310 self.assertEqual(functest_utils.get_openstack_version(
312 args[0].assert_called_once_with(cloud)
314 @mock.patch('functest.utils.functest_utils.get_nova_version',
315 return_value=(2, 78))
316 def test_openstack_version12(self, *args):
318 self.assertEqual(functest_utils.get_openstack_version(
320 args[0].assert_called_once_with(cloud)
322 @mock.patch('functest.utils.functest_utils.get_nova_version',
323 return_value=(2, 87))
324 def test_openstack_version13(self, *args):
326 self.assertEqual(functest_utils.get_openstack_version(
328 args[0].assert_called_once_with(cloud)
330 @mock.patch('functest.utils.functest_utils.get_nova_version',
331 return_value=(2, 88))
332 def test_openstack_version14(self, *args):
334 self.assertEqual(functest_utils.get_openstack_version(
336 args[0].assert_called_once_with(cloud)
338 @mock.patch('functest.utils.functest_utils.get_nova_version',
339 return_value=(2, 92))
340 def test_openstack_version14(self, *args):
342 self.assertEqual(functest_utils.get_openstack_version(
344 args[0].assert_called_once_with(cloud)
346 @mock.patch('functest.utils.functest_utils.get_nova_version',
347 return_value=(2, 89))
348 def test_openstack_version14(self, *args):
350 self.assertEqual(functest_utils.get_openstack_version(
352 args[0].assert_called_once_with(cloud)
354 @mock.patch('functest.utils.functest_utils.get_nova_version',
356 def test_openstack_version_exc(self, *args):
359 functest_utils.get_openstack_version(cloud), "Unknown")
360 args[0].assert_called_once_with(cloud)
362 def test_convert_dict_to_ini(self):
364 functest_utils.convert_dict_to_ini({}), "")
366 functest_utils.convert_dict_to_ini({"a": "b"}), "a:b")
367 value = functest_utils.convert_dict_to_ini({"a": "b", "c": "d"})
368 self.assertTrue(value in ('a:b,c:d', 'c:d,a:b'))
369 with self.assertRaises(AssertionError):
370 functest_utils.convert_list_to_ini("")
372 def test_convert_list_to_ini(self):
374 functest_utils.convert_list_to_ini([]), "")
376 functest_utils.convert_list_to_ini(["a"]), "a")
378 functest_utils.convert_list_to_ini(["a", "b"]), "a,b")
379 with self.assertRaises(AssertionError):
380 functest_utils.convert_list_to_ini("")
382 def test_convert_ini_to_dict(self):
384 functest_utils.convert_ini_to_dict(""), {})
386 functest_utils.convert_ini_to_dict("a:b"), {"a": "b"})
388 functest_utils.convert_ini_to_dict(
389 "a:b,c:d"), {"a": "b", "c": "d"})
391 functest_utils.convert_ini_to_dict(
392 "a:b:c,d:e:f"), {"a:b": "c", "d:e": "f"})
393 with self.assertRaises(AssertionError):
394 functest_utils.convert_list_to_ini({})
396 def test_convert_ini_to_list(self):
398 functest_utils.convert_ini_to_list(""), [])
400 functest_utils.convert_ini_to_list("a"), ["a"])
402 functest_utils.convert_ini_to_list("a,b"), ["a", "b"])
403 with self.assertRaises(AssertionError):
404 functest_utils.convert_ini_to_list([])
407 if __name__ == "__main__":
408 logging.disable(logging.CRITICAL)
409 unittest.main(verbosity=2)