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
19 from six.moves import urllib
21 from functest.utils import functest_utils
24 class FunctestUtilsTesting(unittest.TestCase):
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 @mock.patch('six.moves.urllib.request.urlopen',
67 side_effect=urllib.error.URLError('no host given'))
68 def test_check_internet_connectivity_failed(self, mock_method):
69 self.assertFalse(functest_utils.check_internet_connectivity())
70 mock_method.assert_called_once_with(self.url, timeout=self.timeout)
72 @mock.patch('six.moves.urllib.request.urlopen')
73 def test_check_internet_connectivity_default(self, mock_method):
74 self.assertTrue(functest_utils.check_internet_connectivity())
75 mock_method.assert_called_once_with(self.url, timeout=self.timeout)
77 @mock.patch('six.moves.urllib.request.urlopen')
78 def test_check_internet_connectivity_debian(self, mock_method):
79 self.url = "https://www.debian.org/"
80 self.assertTrue(functest_utils.check_internet_connectivity(self.url))
81 mock_method.assert_called_once_with(self.url, timeout=self.timeout)
83 @mock.patch('six.moves.urllib.request.urlopen',
84 side_effect=urllib.error.URLError('no host given'))
85 def test_download_url_failed(self, mock_url):
86 self.assertFalse(functest_utils.download_url(self.url, self.dest_path))
88 @mock.patch('six.moves.urllib.request.urlopen')
89 def test_download_url_default(self, mock_url):
90 with mock.patch("six.moves.builtins.open", mock.mock_open()) as m, \
91 mock.patch('functest.utils.functest_utils.shutil.copyfileobj')\
93 name = self.url.rsplit('/')[-1]
94 dest = self.dest_path + "/" + name
95 self.assertTrue(functest_utils.download_url(self.url,
97 m.assert_called_once_with(dest, 'wb')
98 self.assertTrue(mock_sh.called)
100 def _get_env_dict(self, var):
101 dic = {'INSTALLER_TYPE': self.installer,
102 'DEPLOY_SCENARIO': self.scenario,
103 'NODE_NAME': self.node_name,
104 'BUILD_TAG': self.build_tag}
110 if FunctestUtilsTesting.readline == \
111 len(FunctestUtilsTesting.test_ip) - 1:
113 FunctestUtilsTesting.readline += 1
114 return FunctestUtilsTesting.test_ip[FunctestUtilsTesting.readline]
116 # TODO: get_resolvconf_ns
117 @mock.patch('functest.utils.functest_utils.dns.resolver.Resolver')
118 def test_get_resolvconf_ns_default(self, mock_dns_resolve):
119 attrs = {'query.return_value': ["test"]}
120 mock_dns_resolve.configure_mock(**attrs)
123 attrs = {'readline.side_effect': self.readline_side}
124 m.configure_mock(**attrs)
126 with mock.patch("six.moves.builtins.open") as mo:
128 self.assertEqual(functest_utils.get_resolvconf_ns(),
131 def _get_environ(self, var):
132 if var == 'INSTALLER_TYPE':
133 return self.installer
134 elif var == 'DEPLOY_SCENARIO':
138 def test_get_ci_envvars_default(self):
139 with mock.patch('os.environ.get',
140 side_effect=self._get_environ):
141 dic = {"installer": self.installer,
142 "scenario": self.scenario}
143 self.assertDictEqual(functest_utils.get_ci_envvars(), dic)
145 def cmd_readline(self):
146 return 'test_value\n'
148 @mock.patch('functest.utils.functest_utils.LOGGER.error')
149 @mock.patch('functest.utils.functest_utils.LOGGER.info')
150 def test_execute_command_args_present_with_error(self, mock_logger_info,
152 with mock.patch('functest.utils.functest_utils.subprocess.Popen') \
153 as mock_subproc_open, \
154 mock.patch('six.moves.builtins.open',
155 mock.mock_open()) as mopen:
157 FunctestUtilsTesting.readline = 0
159 mock_obj = mock.Mock()
160 attrs = {'readline.side_effect': self.cmd_readline()}
161 mock_obj.configure_mock(**attrs)
163 mock_obj2 = mock.Mock()
164 attrs = {'stdout': mock_obj, 'wait.return_value': 1}
165 mock_obj2.configure_mock(**attrs)
167 mock_subproc_open.return_value = mock_obj2
169 resp = functest_utils.execute_command(self.cmd, info=True,
170 error_msg=self.error_msg,
172 output_file=self.output_file)
173 self.assertEqual(resp, 1)
174 msg_exec = ("Executing command: '%s'" % self.cmd)
175 mock_logger_info.assert_called_once_with(msg_exec)
176 mopen.assert_called_once_with(self.output_file, "w")
177 mock_logger_error.assert_called_once_with(self.error_msg)
179 @mock.patch('functest.utils.functest_utils.LOGGER.info')
180 def test_execute_command_args_present_with_success(self, mock_logger_info,
182 with mock.patch('functest.utils.functest_utils.subprocess.Popen') \
183 as mock_subproc_open, \
184 mock.patch('six.moves.builtins.open',
185 mock.mock_open()) as mopen:
187 FunctestUtilsTesting.readline = 0
189 mock_obj = mock.Mock()
190 attrs = {'readline.side_effect': self.cmd_readline()}
191 mock_obj.configure_mock(**attrs)
193 mock_obj2 = mock.Mock()
194 attrs = {'stdout': mock_obj, 'wait.return_value': 0}
195 mock_obj2.configure_mock(**attrs)
197 mock_subproc_open.return_value = mock_obj2
199 resp = functest_utils.execute_command(self.cmd, info=True,
200 error_msg=self.error_msg,
202 output_file=self.output_file)
203 self.assertEqual(resp, 0)
204 msg_exec = ("Executing command: '%s'" % self.cmd)
205 mock_logger_info.assert_called_once_with(msg_exec)
206 mopen.assert_called_once_with(self.output_file, "w")
208 @mock.patch('sys.stdout')
209 def test_execute_command_args_missing_with_success(self, stdout=None):
210 with mock.patch('functest.utils.functest_utils.subprocess.Popen') \
211 as mock_subproc_open:
213 FunctestUtilsTesting.readline = 2
215 mock_obj = mock.Mock()
216 attrs = {'readline.side_effect': self.cmd_readline()}
217 mock_obj.configure_mock(**attrs)
219 mock_obj2 = mock.Mock()
220 attrs = {'stdout': mock_obj, 'wait.return_value': 0}
221 mock_obj2.configure_mock(**attrs)
223 mock_subproc_open.return_value = mock_obj2
225 resp = functest_utils.execute_command(self.cmd, info=False,
229 self.assertEqual(resp, 0)
231 @mock.patch('sys.stdout')
232 def test_execute_command_args_missing_with_error(self, stdout=None):
233 with mock.patch('functest.utils.functest_utils.subprocess.Popen') \
234 as mock_subproc_open:
236 FunctestUtilsTesting.readline = 2
237 mock_obj = mock.Mock()
238 attrs = {'readline.side_effect': self.cmd_readline()}
239 mock_obj.configure_mock(**attrs)
241 mock_obj2 = mock.Mock()
242 attrs = {'stdout': mock_obj, 'wait.return_value': 1}
243 mock_obj2.configure_mock(**attrs)
245 mock_subproc_open.return_value = mock_obj2
247 resp = functest_utils.execute_command(self.cmd, info=False,
251 self.assertEqual(resp, 1)
253 def _get_functest_config(self, var):
256 def test_get_parameter_from_yaml_failed(self):
257 self.file_yaml['general'] = None
258 with mock.patch('six.moves.builtins.open', mock.mock_open()), \
259 mock.patch('functest.utils.functest_utils.yaml.safe_load') \
261 self.assertRaises(ValueError) as excep:
262 mock_yaml.return_value = self.file_yaml
263 functest_utils.get_parameter_from_yaml(self.parameter,
265 self.assertTrue(("The parameter %s is not"
266 " defined in config_functest.yaml" %
267 self.parameter) in excep.exception)
269 def test_get_parameter_from_yaml_default(self):
270 with mock.patch('six.moves.builtins.open', mock.mock_open()), \
271 mock.patch('functest.utils.functest_utils.yaml.safe_load') \
273 mock_yaml.return_value = self.file_yaml
274 self.assertEqual(functest_utils.
275 get_parameter_from_yaml(self.parameter,
279 @mock.patch('functest.utils.functest_utils.get_parameter_from_yaml')
280 def test_get_functest_config_default(self, mock_get_parameter_from_yaml):
281 with mock.patch.dict(os.environ,
282 {'CONFIG_FUNCTEST_YAML': self.config_yaml}):
283 functest_utils.get_functest_config(self.parameter)
284 mock_get_parameter_from_yaml. \
285 assert_called_once_with(self.parameter,
288 def test_get_functest_yaml(self):
289 with mock.patch('six.moves.builtins.open', mock.mock_open()), \
290 mock.patch('functest.utils.functest_utils.yaml.safe_load') \
292 mock_yaml.return_value = self.file_yaml
293 resp = functest_utils.get_functest_yaml()
294 self.assertEqual(resp, self.file_yaml)
296 @mock.patch('functest.utils.functest_utils.LOGGER.info')
297 def test_print_separator(self, mock_logger_info):
298 functest_utils.print_separator()
299 mock_logger_info.assert_called_once_with("======================="
300 "=======================")
303 if __name__ == "__main__":
304 logging.disable(logging.CRITICAL)
305 unittest.main(verbosity=2)