xtesting/tests/unit/core/test_testcase.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 # pylint: disable=missing-docstring
  11
  12 """Define the class required to fully cover testcase."""
  13
  14 from datetime import datetime
  15 import json
  16 import logging
  17 import os
  18 import unittest
  19
  20 import botocore
  21 import mock
  22 import requests
  23
  24 from xtesting.core import testcase
  25 from xtesting.utils import constants
  26
  27 __author__ = "Cedric Ollivier <cedric.ollivier@orange.com>"
  28
  29
  30 class FakeTestCase(testcase.TestCase):
  31     # pylint: disable=too-many-instance-attributes
  32
  33     def run(self, **kwargs):
  34         return testcase.TestCase.EX_OK
  35
  36
  37 class AbstractTestCaseTesting(unittest.TestCase):
  38
  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")
  43
  44
  45 class TestCaseTesting(unittest.TestCase):
  46     # pylint: disable=too-many-instance-attributes,too-many-public-methods
  47
  48     _case_name = "base"
  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'}
  53
  54     def setUp(self):
  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"
  70
  71     def test_run_fake(self):
  72         self.assertEqual(self.test.run(), testcase.TestCase.EX_OK)
  73
  74     def _test_pushdb_missing_attribute(self):
  75         self.assertEqual(self.test.push_to_db(),
  76                          testcase.TestCase.EX_PUSH_TO_DB_ERROR)
  77
  78     def test_pushdb_no_project_name(self):
  79         self.test.project_name = None
  80         self._test_pushdb_missing_attribute()
  81
  82     def test_pushdb_no_case_name(self):
  83         self.test.case_name = None
  84         self._test_pushdb_missing_attribute()
  85
  86     def test_pushdb_no_start_time(self):
  87         self.test.start_time = None
  88         self._test_pushdb_missing_attribute()
  89
  90     def test_pushdb_no_stop_time(self):
  91         self.test.stop_time = None
  92         self._test_pushdb_missing_attribute()
  93
  94     def _test_pushdb_missing_env(self, var):
  95         del os.environ[var]
  96         self.assertEqual(self.test.push_to_db(),
  97                          testcase.TestCase.EX_PUSH_TO_DB_ERROR)
  98
  99     def test_pushdb_no_db_url(self):
 100         self._test_pushdb_missing_env('TEST_DB_URL')
 101
 102     def test_pushdb_no_installer_type(self):
 103         self._test_pushdb_missing_env('INSTALLER_TYPE')
 104
 105     def test_pushdb_no_deploy_scenario(self):
 106         self._test_pushdb_missing_env('DEPLOY_SCENARIO')
 107
 108     def test_pushdb_no_node_name(self):
 109         self._test_pushdb_missing_env('NODE_NAME')
 110
 111     def test_pushdb_no_build_tag(self):
 112         self._test_pushdb_missing_env('BUILD_TAG')
 113
 114     @mock.patch('requests.post')
 115     def test_pushdb_bad_start_time(self, mock_function=None):
 116         self.test.start_time = "1"
 117         self.assertEqual(
 118             self.test.push_to_db(),
 119             testcase.TestCase.EX_PUSH_TO_DB_ERROR)
 120         mock_function.assert_not_called()
 121
 122     @mock.patch('requests.post')
 123     def test_pushdb_bad_end_time(self, mock_function=None):
 124         self.test.stop_time = "2"
 125         self.assertEqual(
 126             self.test.push_to_db(),
 127             testcase.TestCase.EX_PUSH_TO_DB_ERROR)
 128         mock_function.assert_not_called()
 129
 130     @mock.patch('requests.post')
 131     def test_pushdb_skipped_test(self, mock_function=None):
 132         self.test.is_skipped = True
 133         self.assertEqual(
 134             self.test.push_to_db(),
 135             testcase.TestCase.EX_PUSH_TO_DB_ERROR)
 136         mock_function.assert_not_called()
 137
 138     def _get_data(self):
 139         return {
 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'),
 153             "version": "master"}
 154
 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)
 166
 167     def test_pushdb_daily_job(self):
 168         self._test_pushdb_version(version="master")
 169
 170     def test_pushdb_weekly_job(self):
 171         os.environ['BUILD_TAG'] = 'foo-weekly-master-bar'
 172         self._test_pushdb_version(version="master")
 173
 174     def test_pushdb_random_build_tag(self):
 175         os.environ['BUILD_TAG'] = 'whatever'
 176         self._test_pushdb_version(version="unknown")
 177
 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):
 182         self.assertEqual(
 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)
 189
 190     def test_check_requirements(self):
 191         self.test.check_requirements()
 192         self.assertEqual(self.test.is_skipped, False)
 193
 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)
 198
 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)
 203
 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)
 211
 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)
 219
 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)
 224
 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)
 229
 230     def test_check_result_eq(self):
 231         self.test.result = 100
 232         self.assertEqual(self.test.is_successful(),
 233                          testcase.TestCase.EX_OK)
 234
 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)
 240
 241     def test_check_result_zero(self):
 242         self.test.criteria = 0
 243         self.test.result = 0
 244         self.assertEqual(self.test.is_successful(),
 245                          testcase.TestCase.EX_TESTCASE_FAILED)
 246
 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")
 252
 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")
 258
 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")
 263
 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")
 268
 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")
 273
 274     def test_get_duration_skipped_test(self):
 275         self.test.is_skipped = True
 276         self.assertEqual(self.test.get_duration(), "00:00")
 277
 278     def test_str_project_name_ko(self):
 279         self.test.project_name = None
 280         self.assertIn("FakeTestCase object at", str(self.test))
 281
 282     def test_str_case_name_ko(self):
 283         self.test.case_name = None
 284         self.assertIn("FakeTestCase object at", str(self.test))
 285
 286     def test_str_pass(self):
 287         duration = '01:01'
 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)
 297
 298     def test_str_fail(self):
 299         duration = '00:59'
 300         with mock.patch.object(self.test, 'get_duration',
 301                                return_value=duration), \
 302                 mock.patch.object(
 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)
 310
 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)
 318
 319     def test_clean(self):
 320         self.assertEqual(self.test.clean(), None)
 321
 322     def _test_publish_artifacts_nokw(self, key):
 323         del os.environ[key]
 324         self.assertEqual(self.test.publish_artifacts(),
 325                          testcase.TestCase.EX_PUBLISH_ARTIFACTS_ERROR)
 326
 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)
 330
 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'])
 338
 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'])
 345
 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'])
 353
 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'])
 364
 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'])
 374
 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'])
 388
 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',
 393                 return_value=[
 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)
 399         expected = [
 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(
 404                 constants.LOG_PATH,
 405                 os.path.join('prefix', os.path.basename(constants.LOG_PATH)),
 406                 Config=mock.ANY,
 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)),
 413                 Config=mock.ANY,
 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'),
 418                 'prefix/bar',
 419                 Config=mock.ANY,
 420                 ExtraArgs={'ContentType': 'application/octet-stream'})]
 421         self.assertEqual(args[1].mock_calls, expected)
 422
 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',
 427                 return_value=[
 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)
 433         expected = [
 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(
 438                 constants.LOG_PATH,
 439                 os.path.join('prefix', os.path.basename(constants.LOG_PATH)),
 440                 Config=mock.ANY,
 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)),
 447                 Config=mock.ANY,
 448                 ExtraArgs={'ContentType': 'text/plain'}),
 449             mock.call().Bucket('xtesting'),
 450             mock.call().Bucket().upload_file(
 451                 os.path.join(constants.RESULTS_DIR, 'bar'),
 452                 'prefix/bar',
 453                 Config=mock.ANY,
 454                 ExtraArgs={'ContentType': 'text/plain'})]
 455         self.assertEqual(args[1].mock_calls, expected)
 456
 457
 458 if __name__ == "__main__":
 459     logging.disable(logging.CRITICAL)
 460     unittest.main(verbosity=2)