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