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
21 from functest.utils import functest_utils
24 class FunctestUtilsTesting(unittest.TestCase):
25 # pylint: disable=too-many-instance-attributes
28 test_ip = ['10.1.23.4', '10.1.14.15', '10.1.16.15']
31 self.url = 'http://www.opnfv.org/'
33 self.dest_path = 'test_path'
34 self.repo_path = 'test_repo_path'
35 self.installer = 'test_installer'
36 self.scenario = 'test_scenario'
37 self.build_tag = 'jenkins-functest-fuel-opnfv-jump-2-daily-master-190'
38 self.build_tag_week = 'jenkins-functest-fuel-baremetal-weekly-master-8'
39 self.version = 'master'
40 self.node_name = 'test_node_name'
41 self.project = 'test_project'
42 self.case_name = 'test_case_name'
43 self.status = 'test_status'
44 self.details = 'test_details'
45 self.db_url = 'test_db_url'
48 self.start_date = 1482624000
49 self.stop_date = 1482624000
50 self.start_time = time.time()
51 self.stop_time = time.time()
53 self.test_ip = ['10.1.23.4', '10.1.14.15', '10.1.16.15']
54 self.test_file = 'test_file'
55 self.error_msg = 'test_error_msg'
57 self.output_file = 'test_output_file'
58 self.testname = 'testname'
59 self.parameter = 'general.openstack.image_name'
60 self.config_yaml = pkg_resources.resource_filename(
61 'functest', 'ci/config_functest.yaml')
62 self.db_url_env = 'http://foo/testdb'
63 self.testcases_yaml = "test_testcases_yaml"
64 self.file_yaml = {'general': {'openstack': {'image_name':
67 def _get_env_dict(self, var):
68 dic = {'INSTALLER_TYPE': self.installer,
69 'DEPLOY_SCENARIO': self.scenario,
70 'NODE_NAME': self.node_name,
71 'BUILD_TAG': self.build_tag}
77 if FunctestUtilsTesting.readline == \
78 len(FunctestUtilsTesting.test_ip) - 1:
80 FunctestUtilsTesting.readline += 1
81 return FunctestUtilsTesting.test_ip[FunctestUtilsTesting.readline]
83 def _get_environ(self, var, *args): # pylint: disable=unused-argument
84 if var == 'INSTALLER_TYPE':
86 elif var == 'DEPLOY_SCENARIO':
94 @mock.patch('functest.utils.functest_utils.LOGGER.error')
95 @mock.patch('functest.utils.functest_utils.LOGGER.info')
96 def test_exec_cmd_args_present_ko(self, mock_logger_info,
98 with mock.patch('functest.utils.functest_utils.subprocess.Popen') \
99 as mock_subproc_open, \
100 mock.patch('six.moves.builtins.open',
101 mock.mock_open()) as mopen:
103 FunctestUtilsTesting.readline = 0
105 mock_obj = mock.Mock()
106 attrs = {'readline.side_effect': self.cmd_readline()}
107 mock_obj.configure_mock(**attrs)
109 mock_obj2 = mock.Mock()
110 attrs = {'stdout': mock_obj, 'wait.return_value': 1}
111 mock_obj2.configure_mock(**attrs)
113 mock_subproc_open.return_value = mock_obj2
115 resp = functest_utils.execute_command(self.cmd, info=True,
116 error_msg=self.error_msg,
118 output_file=self.output_file)
119 self.assertEqual(resp, 1)
120 msg_exec = ("Executing command: '%s'" % self.cmd)
121 mock_logger_info.assert_called_once_with(msg_exec)
122 mopen.assert_called_once_with(self.output_file, "w")
123 mock_logger_error.assert_called_once_with(self.error_msg)
125 @mock.patch('functest.utils.functest_utils.LOGGER.info')
126 def test_exec_cmd_args_present_ok(self, mock_logger_info):
127 with mock.patch('functest.utils.functest_utils.subprocess.Popen') \
128 as mock_subproc_open, \
129 mock.patch('six.moves.builtins.open',
130 mock.mock_open()) as mopen:
132 FunctestUtilsTesting.readline = 0
134 mock_obj = mock.Mock()
135 attrs = {'readline.side_effect': self.cmd_readline()}
136 mock_obj.configure_mock(**attrs)
138 mock_obj2 = mock.Mock()
139 attrs = {'stdout': mock_obj, 'wait.return_value': 0}
140 mock_obj2.configure_mock(**attrs)
142 mock_subproc_open.return_value = mock_obj2
144 resp = functest_utils.execute_command(self.cmd, info=True,
145 error_msg=self.error_msg,
147 output_file=self.output_file)
148 self.assertEqual(resp, 0)
149 msg_exec = ("Executing command: '%s'" % self.cmd)
150 mock_logger_info.assert_called_once_with(msg_exec)
151 mopen.assert_called_once_with(self.output_file, "w")
153 @mock.patch('sys.stdout')
154 def test_exec_cmd_args_missing_ok(self, stdout=None):
155 # pylint: disable=unused-argument
156 with mock.patch('functest.utils.functest_utils.subprocess.Popen') \
157 as mock_subproc_open:
159 FunctestUtilsTesting.readline = 2
161 mock_obj = mock.Mock()
162 attrs = {'readline.side_effect': self.cmd_readline()}
163 mock_obj.configure_mock(**attrs)
165 mock_obj2 = mock.Mock()
166 attrs = {'stdout': mock_obj, 'wait.return_value': 0}
167 mock_obj2.configure_mock(**attrs)
169 mock_subproc_open.return_value = mock_obj2
171 resp = functest_utils.execute_command(self.cmd, info=False,
175 self.assertEqual(resp, 0)
177 @mock.patch('sys.stdout')
178 def test_exec_cmd_args_missing_ko(self, stdout=None):
179 # pylint: disable=unused-argument
180 with mock.patch('functest.utils.functest_utils.subprocess.Popen') \
181 as mock_subproc_open:
183 FunctestUtilsTesting.readline = 2
184 mock_obj = mock.Mock()
185 attrs = {'readline.side_effect': self.cmd_readline()}
186 mock_obj.configure_mock(**attrs)
188 mock_obj2 = mock.Mock()
189 attrs = {'stdout': mock_obj, 'wait.return_value': 1}
190 mock_obj2.configure_mock(**attrs)
192 mock_subproc_open.return_value = mock_obj2
194 resp = functest_utils.execute_command(self.cmd, info=False,
198 self.assertEqual(resp, 1)
200 def test_get_param_from_yaml_failed(self):
201 self.file_yaml['general'] = None
202 with mock.patch('six.moves.builtins.open', mock.mock_open()), \
203 mock.patch('functest.utils.functest_utils.yaml.safe_load') \
205 self.assertRaises(ValueError) as excep:
206 mock_yaml.return_value = self.file_yaml
207 functest_utils.get_parameter_from_yaml(self.parameter,
209 self.assertTrue(("The parameter %s is not"
210 " defined in config_functest.yaml" %
211 self.parameter) in excep.exception)
213 def test_get_param_from_yaml_def(self):
214 with mock.patch('six.moves.builtins.open', mock.mock_open()), \
215 mock.patch('functest.utils.functest_utils.yaml.safe_load') \
217 mock_yaml.return_value = self.file_yaml
218 self.assertEqual(functest_utils.
219 get_parameter_from_yaml(self.parameter,
223 def test_get_extnetwork_nocloud(self):
224 with self.assertRaises(AssertionError):
225 functest_utils.get_external_network(None)
227 def test_get_extnetwork_env_ok1(self):
229 cloud.get_network.return_value = munch.Munch(name="dummy")
230 os.environ["EXTERNAL_NETWORK"] = 'dummy'
232 functest_utils.get_external_network(cloud),
233 cloud.get_network.return_value)
234 cloud.get_network.assert_called_once_with(
235 'dummy', {'router:external': True})
236 cloud.list_networks.assert_not_called()
238 def test_get_extnetwork_env_ok2(self):
240 cloud.get_network.return_value = None
241 cloud.list_networks.return_value = None
242 os.environ["EXTERNAL_NETWORK"] = 'dummy'
243 self.assertEqual(functest_utils.get_external_network(cloud), None)
244 cloud.get_network.assert_called_once_with(
245 'dummy', {'router:external': True})
246 cloud.list_networks.assert_called_once_with(
247 {'router:external': True})
249 def test_get_extnetwork_env_ko(self):
251 cloud.get_network.return_value = None
252 cloud.list_networks.return_value = [munch.Munch(name="dummy")]
253 os.environ["EXTERNAL_NETWORK"] = 'dummy'
255 functest_utils.get_external_network(cloud),
256 cloud.list_networks.return_value[0])
257 cloud.get_network.assert_called_once_with(
258 'dummy', {'router:external': True})
259 cloud.list_networks.assert_called_once_with(
260 {'router:external': True})
262 def test_get_extnetwork_noenv_ko(self):
264 del os.environ["EXTERNAL_NETWORK"]
265 except Exception: # pylint: disable=broad-except
268 cloud.list_networks.return_value = None
269 self.assertEqual(functest_utils.get_external_network(cloud), None)
270 cloud.get_network.assert_not_called()
271 cloud.list_networks.assert_called_once_with(
272 {'router:external': True})
274 def test_get_extnetwork_noenv_ok(self):
276 del os.environ["EXTERNAL_NETWORK"]
277 except Exception: # pylint: disable=broad-except
280 cloud.list_networks.return_value = [munch.Munch(name="dummy")]
282 functest_utils.get_external_network(cloud),
283 cloud.list_networks.return_value[0])
284 cloud.get_network.assert_not_called()
285 cloud.list_networks.assert_called_once_with(
286 {'router:external': True})
288 if __name__ == "__main__":
289 logging.disable(logging.CRITICAL)
290 unittest.main(verbosity=2)