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
12 """Define the class required to fully cover testcase."""
14 from datetime import datetime
24 from xtesting.core import testcase
25 from xtesting.utils import constants
27 __author__ = "Cedric Ollivier <cedric.ollivier@orange.com>"
30 class FakeTestCase(testcase.TestCase):
31 # pylint: disable=too-many-instance-attributes
33 def run(self, **kwargs):
34 return testcase.TestCase.EX_OK
37 class AbstractTestCaseTesting(unittest.TestCase):
39 def test_run_unimplemented(self):
40 # pylint: disable=abstract-class-instantiated
41 with self.assertRaises(TypeError):
42 testcase.TestCase(case_name="base", project_name="xtesting")
45 class TestCaseTesting(unittest.TestCase):
46 # pylint: disable=too-many-instance-attributes,too-many-public-methods
49 _project_name = "xtesting"
50 _published_result = "PASS"
51 _test_db_url = "http://testresults.opnfv.org/test/api/v1/results"
52 _headers = {'Content-Type': 'application/json'}
55 self.test = FakeTestCase(
56 case_name=self._case_name, project_name=self._project_name)
57 self.test.start_time = 1
58 self.test.stop_time = 2
59 self.test.result = 100
60 self.test.details = {"Hello": "World"}
61 os.environ['TEST_DB_URL'] = TestCaseTesting._test_db_url
62 os.environ['TEST_DB_EXT_URL'] = TestCaseTesting._test_db_url
63 os.environ['INSTALLER_TYPE'] = "installer_type"
64 os.environ['DEPLOY_SCENARIO'] = "scenario"
65 os.environ['NODE_NAME'] = "node_name"
66 os.environ['BUILD_TAG'] = "foo-daily-master-bar"
67 os.environ['S3_ENDPOINT_URL'] = "http://127.0.0.1:9000"
68 os.environ['S3_DST_URL'] = "s3://xtesting/prefix"
69 os.environ['HTTP_DST_URL'] = "http://127.0.0.1/prefix"
71 def test_run_fake(self):
72 self.assertEqual(self.test.run(), testcase.TestCase.EX_OK)
74 def _test_pushdb_missing_attribute(self):
75 self.assertEqual(self.test.push_to_db(),
76 testcase.TestCase.EX_PUSH_TO_DB_ERROR)
78 def test_pushdb_no_project_name(self):
79 self.test.project_name = None
80 self._test_pushdb_missing_attribute()
82 def test_pushdb_no_case_name(self):
83 self.test.case_name = None
84 self._test_pushdb_missing_attribute()
86 def test_pushdb_no_start_time(self):
87 self.test.start_time = None
88 self._test_pushdb_missing_attribute()
90 def test_pushdb_no_stop_time(self):
91 self.test.stop_time = None
92 self._test_pushdb_missing_attribute()
94 def _test_pushdb_missing_env(self, var):
96 self.assertEqual(self.test.push_to_db(),
97 testcase.TestCase.EX_PUSH_TO_DB_ERROR)
99 def test_pushdb_no_db_url(self):
100 self._test_pushdb_missing_env('TEST_DB_URL')
102 def test_pushdb_no_installer_type(self):
103 self._test_pushdb_missing_env('INSTALLER_TYPE')
105 def test_pushdb_no_deploy_scenario(self):
106 self._test_pushdb_missing_env('DEPLOY_SCENARIO')
108 def test_pushdb_no_node_name(self):
109 self._test_pushdb_missing_env('NODE_NAME')
111 def test_pushdb_no_build_tag(self):
112 self._test_pushdb_missing_env('BUILD_TAG')
114 @mock.patch('requests.post')
115 def test_pushdb_bad_start_time(self, mock_function=None):
116 self.test.start_time = "1"
118 self.test.push_to_db(),
119 testcase.TestCase.EX_PUSH_TO_DB_ERROR)
120 mock_function.assert_not_called()
122 @mock.patch('requests.post')
123 def test_pushdb_bad_end_time(self, mock_function=None):
124 self.test.stop_time = "2"
126 self.test.push_to_db(),
127 testcase.TestCase.EX_PUSH_TO_DB_ERROR)
128 mock_function.assert_not_called()
130 @mock.patch('requests.post')
131 def test_pushdb_skipped_test(self, mock_function=None):
132 self.test.is_skipped = True
134 self.test.push_to_db(),
135 testcase.TestCase.EX_PUSH_TO_DB_ERROR)
136 mock_function.assert_not_called()
140 "build_tag": os.environ['BUILD_TAG'],
141 "case_name": self._case_name,
142 "criteria": 'PASS' if self.test.is_successful(
143 ) == self.test.EX_OK else 'FAIL',
144 "details": self.test.details,
145 "installer": os.environ['INSTALLER_TYPE'],
146 "pod_name": os.environ['NODE_NAME'],
147 "project_name": self.test.project_name,
148 "scenario": os.environ['DEPLOY_SCENARIO'],
149 "start_date": datetime.fromtimestamp(
150 self.test.start_time).strftime('%Y-%m-%d %H:%M:%S'),
151 "stop_date": datetime.fromtimestamp(
152 self.test.stop_time).strftime('%Y-%m-%d %H:%M:%S'),
155 @mock.patch('os.path.join', return_value='')
156 @mock.patch('re.sub', return_value='')
157 @mock.patch('requests.post')
158 def _test_pushdb_version(self, *args, **kwargs):
159 payload = self._get_data()
160 payload["version"] = kwargs.get("version", "unknown")
161 self.assertEqual(self.test.push_to_db(), testcase.TestCase.EX_OK)
162 args[0].assert_called_once_with(
163 os.environ['TEST_DB_URL'],
164 data=json.dumps(payload, sort_keys=True),
165 headers=self._headers)
167 def test_pushdb_daily_job(self):
168 self._test_pushdb_version(version="master")
170 def test_pushdb_weekly_job(self):
171 os.environ['BUILD_TAG'] = 'foo-weekly-master-bar'
172 self._test_pushdb_version(version="master")
174 def test_pushdb_random_build_tag(self):
175 os.environ['BUILD_TAG'] = 'whatever'
176 self._test_pushdb_version(version="unknown")
178 @mock.patch('requests.post', return_value=mock.Mock(
179 raise_for_status=mock.Mock(
180 side_effect=requests.exceptions.HTTPError)))
181 def test_pushdb_http_errors(self, mock_function=None):
183 self.test.push_to_db(),
184 testcase.TestCase.EX_PUSH_TO_DB_ERROR)
185 mock_function.assert_called_once_with(
186 os.environ['TEST_DB_URL'],
187 data=json.dumps(self._get_data(), sort_keys=True),
188 headers=self._headers)
190 def test_check_requirements(self):
191 self.test.check_requirements()
192 self.assertEqual(self.test.is_skipped, False)
194 def test_check_criteria_missing(self):
195 self.test.criteria = None
196 self.assertEqual(self.test.is_successful(),
197 testcase.TestCase.EX_TESTCASE_FAILED)
199 def test_check_result_missing(self):
200 self.test.result = None
201 self.assertEqual(self.test.is_successful(),
202 testcase.TestCase.EX_TESTCASE_FAILED)
204 def test_check_result_failed(self):
205 # Backward compatibility
206 # It must be removed as soon as TestCase subclasses
207 # stop setting result = 'PASS' or 'FAIL'.
208 self.test.result = 'FAIL'
209 self.assertEqual(self.test.is_successful(),
210 testcase.TestCase.EX_TESTCASE_FAILED)
212 def test_check_result_pass(self):
213 # Backward compatibility
214 # It must be removed as soon as TestCase subclasses
215 # stop setting result = 'PASS' or 'FAIL'.
216 self.test.result = 'PASS'
217 self.assertEqual(self.test.is_successful(),
218 testcase.TestCase.EX_OK)
220 def test_check_result_skip(self):
221 self.test.is_skipped = True
222 self.assertEqual(self.test.is_successful(),
223 testcase.TestCase.EX_TESTCASE_SKIPPED)
225 def test_check_result_lt(self):
226 self.test.result = 50
227 self.assertEqual(self.test.is_successful(),
228 testcase.TestCase.EX_TESTCASE_FAILED)
230 def test_check_result_eq(self):
231 self.test.result = 100
232 self.assertEqual(self.test.is_successful(),
233 testcase.TestCase.EX_OK)
235 def test_check_result_gt(self):
236 self.test.criteria = 50
237 self.test.result = 100
238 self.assertEqual(self.test.is_successful(),
239 testcase.TestCase.EX_OK)
241 def test_check_result_zero(self):
242 self.test.criteria = 0
244 self.assertEqual(self.test.is_successful(),
245 testcase.TestCase.EX_TESTCASE_FAILED)
247 def test_get_duration_start_ko(self):
248 self.test.start_time = None
249 self.assertEqual(self.test.get_duration(), "XX:XX")
250 self.test.start_time = 0
251 self.assertEqual(self.test.get_duration(), "XX:XX")
253 def test_get_duration_end_ko(self):
254 self.test.stop_time = None
255 self.assertEqual(self.test.get_duration(), "XX:XX")
256 self.test.stop_time = 0
257 self.assertEqual(self.test.get_duration(), "XX:XX")
259 def test_get_invalid_duration(self):
260 self.test.start_time = 2
261 self.test.stop_time = 1
262 self.assertEqual(self.test.get_duration(), "XX:XX")
264 def test_get_zero_duration(self):
265 self.test.start_time = 2
266 self.test.stop_time = 2
267 self.assertEqual(self.test.get_duration(), "00:00")
269 def test_get_duration(self):
270 self.test.start_time = 1
271 self.test.stop_time = 180
272 self.assertEqual(self.test.get_duration(), "02:59")
274 def test_get_duration_skipped_test(self):
275 self.test.is_skipped = True
276 self.assertEqual(self.test.get_duration(), "00:00")
278 def test_str_project_name_ko(self):
279 self.test.project_name = None
280 self.assertIn("FakeTestCase object at", str(self.test))
282 def test_str_case_name_ko(self):
283 self.test.case_name = None
284 self.assertIn("FakeTestCase object at", str(self.test))
286 def test_str_pass(self):
288 with mock.patch.object(self.test, 'get_duration',
289 return_value=duration), \
290 mock.patch.object(self.test, 'is_successful',
291 return_value=testcase.TestCase.EX_OK):
292 message = str(self.test)
293 self.assertIn(self._project_name, message)
294 self.assertIn(self._case_name, message)
295 self.assertIn(duration, message)
296 self.assertIn('PASS', message)
298 def test_str_fail(self):
300 with mock.patch.object(self.test, 'get_duration',
301 return_value=duration), \
303 self.test, 'is_successful',
304 return_value=testcase.TestCase.EX_TESTCASE_FAILED):
305 message = str(self.test)
306 self.assertIn(self._project_name, message)
307 self.assertIn(self._case_name, message)
308 self.assertIn(duration, message)
309 self.assertIn('FAIL', message)
311 def test_str_skip(self):
312 self.test.is_skipped = True
313 message = str(self.test)
314 self.assertIn(self._project_name, message)
315 self.assertIn(self._case_name, message)
316 self.assertIn("00:00", message)
317 self.assertIn('SKIP', message)
319 def test_clean(self):
320 self.assertEqual(self.test.clean(), None)
322 def _test_publish_artifacts_nokw(self, key):
324 self.assertEqual(self.test.publish_artifacts(),
325 testcase.TestCase.EX_PUBLISH_ARTIFACTS_ERROR)
327 def test_publish_artifacts_exc1(self):
328 for key in ["S3_ENDPOINT_URL", "S3_DST_URL", "HTTP_DST_URL"]:
329 self._test_publish_artifacts_nokw(key)
331 @mock.patch('boto3.resource',
332 side_effect=botocore.exceptions.NoCredentialsError)
333 def test_publish_artifacts_exc2(self, *args):
334 self.assertEqual(self.test.publish_artifacts(),
335 testcase.TestCase.EX_PUBLISH_ARTIFACTS_ERROR)
336 args[0].assert_called_once_with(
337 's3', endpoint_url=os.environ['S3_ENDPOINT_URL'])
339 @mock.patch('boto3.resource', side_effect=Exception)
340 def test_publish_artifacts_exc3(self, *args):
341 self.assertEqual(self.test.publish_artifacts(),
342 testcase.TestCase.EX_PUBLISH_ARTIFACTS_ERROR)
343 args[0].assert_called_once_with(
344 's3', endpoint_url=os.environ['S3_ENDPOINT_URL'])
346 @mock.patch('boto3.resource')
347 def test_publish_artifacts_exc4(self, *args):
348 args[0].return_value.meta.client.head_bucket.side_effect = Exception
349 self.assertEqual(self.test.publish_artifacts(),
350 testcase.TestCase.EX_PUBLISH_ARTIFACTS_ERROR)
351 args[0].assert_called_once_with(
352 's3', endpoint_url=os.environ['S3_ENDPOINT_URL'])
354 @mock.patch('boto3.resource')
355 def test_publish_artifacts_exc5(self, *args):
356 error_response = {'Error': {'Code': '403'}}
357 mock_head_bucket = args[0].return_value.meta.client.head_bucket
358 mock_head_bucket.side_effect = botocore.exceptions.ClientError(
359 error_response, 'Foo')
360 self.assertEqual(self.test.publish_artifacts(),
361 testcase.TestCase.EX_PUBLISH_ARTIFACTS_ERROR)
362 args[0].assert_called_once_with(
363 's3', endpoint_url=os.environ['S3_ENDPOINT_URL'])
365 @mock.patch('mimetypes.guess_type', return_value=(None, None))
366 @mock.patch('boto3.resource')
367 @mock.patch('os.walk', return_value=[])
368 def test_publish_artifacts1(self, *args):
369 self.assertEqual(self.test.publish_artifacts(),
370 testcase.TestCase.EX_OK)
371 args[0].assert_called_once_with(self.test.res_dir)
372 args[1].assert_called_once_with(
373 's3', endpoint_url=os.environ['S3_ENDPOINT_URL'])
375 @mock.patch('mimetypes.guess_type', return_value=(None, None))
376 @mock.patch('boto3.resource')
377 @mock.patch('os.walk', return_value=[])
378 def test_publish_artifacts2(self, *args):
379 error_response = {'Error': {'Code': '404'}}
380 mock_head_bucket = args[1].return_value.meta.client.head_bucket
381 mock_head_bucket.side_effect = botocore.exceptions.ClientError(
382 error_response, 'NoSuchBucket')
383 self.assertEqual(self.test.publish_artifacts(),
384 testcase.TestCase.EX_OK)
385 args[0].assert_called_once_with(self.test.res_dir)
386 args[1].assert_called_once_with(
387 's3', endpoint_url=os.environ['S3_ENDPOINT_URL'])
389 @mock.patch('mimetypes.guess_type', return_value=(None, None))
390 @mock.patch('os.path.exists', return_value=True)
391 @mock.patch('boto3.resource')
392 @mock.patch('os.walk',
394 (testcase.TestCase.dir_results, ('',), ('bar',))])
395 def test_publish_artifacts3(self, *args):
396 self.assertEqual(self.test.publish_artifacts(),
397 testcase.TestCase.EX_OK)
398 args[0].assert_called_once_with(self.test.res_dir)
400 mock.call('s3', endpoint_url=os.environ['S3_ENDPOINT_URL']),
401 mock.call().meta.client.head_bucket(Bucket='xtesting'),
402 mock.call().Bucket('xtesting'),
403 mock.call().Bucket().upload_file(
405 os.path.join('prefix', os.path.basename(constants.LOG_PATH)),
407 ExtraArgs={'ContentType': 'application/octet-stream'}),
408 mock.call().Bucket('xtesting'),
409 mock.call().Bucket().upload_file(
410 constants.DEBUG_LOG_PATH,
411 os.path.join('prefix',
412 os.path.basename(constants.DEBUG_LOG_PATH)),
414 ExtraArgs={'ContentType': 'application/octet-stream'}),
415 mock.call().Bucket('xtesting'),
416 mock.call().Bucket().upload_file(
417 os.path.join(constants.RESULTS_DIR, 'bar'),
420 ExtraArgs={'ContentType': 'application/octet-stream'})]
421 self.assertEqual(args[1].mock_calls, expected)
423 @mock.patch('mimetypes.guess_type', return_value=('text/plain', None))
424 @mock.patch('os.path.exists', return_value=True)
425 @mock.patch('boto3.resource')
426 @mock.patch('os.walk',
428 (testcase.TestCase.dir_results, ('',), ('bar',))])
429 def test_publish_artifacts4(self, *args):
430 self.assertEqual(self.test.publish_artifacts(),
431 testcase.TestCase.EX_OK)
432 args[0].assert_called_once_with(self.test.res_dir)
434 mock.call('s3', endpoint_url=os.environ['S3_ENDPOINT_URL']),
435 mock.call().meta.client.head_bucket(Bucket='xtesting'),
436 mock.call().Bucket('xtesting'),
437 mock.call().Bucket().upload_file(
439 os.path.join('prefix', os.path.basename(constants.LOG_PATH)),
441 ExtraArgs={'ContentType': 'text/plain'}),
442 mock.call().Bucket('xtesting'),
443 mock.call().Bucket().upload_file(
444 constants.DEBUG_LOG_PATH,
445 os.path.join('prefix',
446 os.path.basename(constants.DEBUG_LOG_PATH)),
448 ExtraArgs={'ContentType': 'text/plain'}),
449 mock.call().Bucket('xtesting'),
450 mock.call().Bucket().upload_file(
451 os.path.join(constants.RESULTS_DIR, 'bar'),
454 ExtraArgs={'ContentType': 'text/plain'})]
455 self.assertEqual(args[1].mock_calls, expected)
458 if __name__ == "__main__":
459 logging.disable(logging.CRITICAL)
460 unittest.main(verbosity=2)