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('requests.post')
 155     def _test_pushdb_version(self, mock_function=None, **kwargs):
 156         payload = self._get_data()
 157         payload["version"] = kwargs.get("version", "unknown")
 158         self.assertEqual(self.test.push_to_db(), testcase.TestCase.EX_OK)
 159         mock_function.assert_called_once_with(
 160             os.environ['TEST_DB_URL'],
 161             data=json.dumps(payload, sort_keys=True),
 162             headers=self._headers)
 163
 164     def test_pushdb_daily_job(self):
 165         self._test_pushdb_version(version="master")
 166
 167     def test_pushdb_weekly_job(self):
 168         os.environ['BUILD_TAG'] = 'foo-weekly-master-bar'
 169         self._test_pushdb_version(version="master")
 170
 171     def test_pushdb_random_build_tag(self):
 172         os.environ['BUILD_TAG'] = 'whatever'
 173         self._test_pushdb_version(version="unknown")
 174
 175     @mock.patch('requests.post', return_value=mock.Mock(
 176         raise_for_status=mock.Mock(
 177             side_effect=requests.exceptions.HTTPError)))
 178     def test_pushdb_http_errors(self, mock_function=None):
 179         self.assertEqual(
 180             self.test.push_to_db(),
 181             testcase.TestCase.EX_PUSH_TO_DB_ERROR)
 182         mock_function.assert_called_once_with(
 183             os.environ['TEST_DB_URL'],
 184             data=json.dumps(self._get_data(), sort_keys=True),
 185             headers=self._headers)
 186
 187     def test_check_requirements(self):
 188         self.test.check_requirements()
 189         self.assertEqual(self.test.is_skipped, False)
 190
 191     def test_check_criteria_missing(self):
 192         self.test.criteria = None
 193         self.assertEqual(self.test.is_successful(),
 194                          testcase.TestCase.EX_TESTCASE_FAILED)
 195
 196     def test_check_result_missing(self):
 197         self.test.result = None
 198         self.assertEqual(self.test.is_successful(),
 199                          testcase.TestCase.EX_TESTCASE_FAILED)
 200
 201     def test_check_result_failed(self):
 202         # Backward compatibility
 203         # It must be removed as soon as TestCase subclasses
 204         # stop setting result = 'PASS' or 'FAIL'.
 205         self.test.result = 'FAIL'
 206         self.assertEqual(self.test.is_successful(),
 207                          testcase.TestCase.EX_TESTCASE_FAILED)
 208
 209     def test_check_result_pass(self):
 210         # Backward compatibility
 211         # It must be removed as soon as TestCase subclasses
 212         # stop setting result = 'PASS' or 'FAIL'.
 213         self.test.result = 'PASS'
 214         self.assertEqual(self.test.is_successful(),
 215                          testcase.TestCase.EX_OK)
 216
 217     def test_check_result_skip(self):
 218         self.test.is_skipped = True
 219         self.assertEqual(self.test.is_successful(),
 220                          testcase.TestCase.EX_TESTCASE_SKIPPED)
 221
 222     def test_check_result_lt(self):
 223         self.test.result = 50
 224         self.assertEqual(self.test.is_successful(),
 225                          testcase.TestCase.EX_TESTCASE_FAILED)
 226
 227     def test_check_result_eq(self):
 228         self.test.result = 100
 229         self.assertEqual(self.test.is_successful(),
 230                          testcase.TestCase.EX_OK)
 231
 232     def test_check_result_gt(self):
 233         self.test.criteria = 50
 234         self.test.result = 100
 235         self.assertEqual(self.test.is_successful(),
 236                          testcase.TestCase.EX_OK)
 237
 238     def test_check_result_zero(self):
 239         self.test.criteria = 0
 240         self.test.result = 0
 241         self.assertEqual(self.test.is_successful(),
 242                          testcase.TestCase.EX_TESTCASE_FAILED)
 243
 244     def test_get_duration_start_ko(self):
 245         self.test.start_time = None
 246         self.assertEqual(self.test.get_duration(), "XX:XX")
 247         self.test.start_time = 0
 248         self.assertEqual(self.test.get_duration(), "XX:XX")
 249
 250     def test_get_duration_end_ko(self):
 251         self.test.stop_time = None
 252         self.assertEqual(self.test.get_duration(), "XX:XX")
 253         self.test.stop_time = 0
 254         self.assertEqual(self.test.get_duration(), "XX:XX")
 255
 256     def test_get_invalid_duration(self):
 257         self.test.start_time = 2
 258         self.test.stop_time = 1
 259         self.assertEqual(self.test.get_duration(), "XX:XX")
 260
 261     def test_get_zero_duration(self):
 262         self.test.start_time = 2
 263         self.test.stop_time = 2
 264         self.assertEqual(self.test.get_duration(), "00:00")
 265
 266     def test_get_duration(self):
 267         self.test.start_time = 1
 268         self.test.stop_time = 180
 269         self.assertEqual(self.test.get_duration(), "02:59")
 270
 271     def test_get_duration_skipped_test(self):
 272         self.test.is_skipped = True
 273         self.assertEqual(self.test.get_duration(), "00:00")
 274
 275     def test_str_project_name_ko(self):
 276         self.test.project_name = None
 277         self.assertIn("FakeTestCase object at", str(self.test))
 278
 279     def test_str_case_name_ko(self):
 280         self.test.case_name = None
 281         self.assertIn("FakeTestCase object at", str(self.test))
 282
 283     def test_str_pass(self):
 284         duration = '01:01'
 285         with mock.patch.object(self.test, 'get_duration',
 286                                return_value=duration), \
 287                 mock.patch.object(self.test, 'is_successful',
 288                                   return_value=testcase.TestCase.EX_OK):
 289             message = str(self.test)
 290         self.assertIn(self._project_name, message)
 291         self.assertIn(self._case_name, message)
 292         self.assertIn(duration, message)
 293         self.assertIn('PASS', message)
 294
 295     def test_str_fail(self):
 296         duration = '00:59'
 297         with mock.patch.object(self.test, 'get_duration',
 298                                return_value=duration), \
 299                 mock.patch.object(
 300                     self.test, 'is_successful',
 301                     return_value=testcase.TestCase.EX_TESTCASE_FAILED):
 302             message = str(self.test)
 303         self.assertIn(self._project_name, message)
 304         self.assertIn(self._case_name, message)
 305         self.assertIn(duration, message)
 306         self.assertIn('FAIL', message)
 307
 308     def test_str_skip(self):
 309         self.test.is_skipped = True
 310         message = str(self.test)
 311         self.assertIn(self._project_name, message)
 312         self.assertIn(self._case_name, message)
 313         self.assertIn("00:00", message)
 314         self.assertIn('SKIP', message)
 315
 316     def test_clean(self):
 317         self.assertEqual(self.test.clean(), None)
 318
 319     def _test_publish_artifacts_nokw(self, key):
 320         del os.environ[key]
 321         self.assertEqual(self.test.publish_artifacts(),
 322                          testcase.TestCase.EX_PUBLISH_ARTIFACTS_ERROR)
 323
 324     def test_publish_artifacts_exc1(self):
 325         for key in ["S3_ENDPOINT_URL", "S3_DST_URL", "HTTP_DST_URL"]:
 326             self._test_publish_artifacts_nokw(key)
 327
 328     @mock.patch('boto3.resource',
 329                 side_effect=botocore.exceptions.NoCredentialsError)
 330     def test_publish_artifacts_exc2(self, *args):
 331         self.assertEqual(self.test.publish_artifacts(),
 332                          testcase.TestCase.EX_PUBLISH_ARTIFACTS_ERROR)
 333         args[0].assert_called_once_with(
 334             's3', endpoint_url=os.environ['S3_ENDPOINT_URL'])
 335
 336     @mock.patch('boto3.resource', side_effect=Exception)
 337     def test_publish_artifacts_exc3(self, *args):
 338         self.assertEqual(self.test.publish_artifacts(),
 339                          testcase.TestCase.EX_PUBLISH_ARTIFACTS_ERROR)
 340         args[0].assert_called_once_with(
 341             's3', endpoint_url=os.environ['S3_ENDPOINT_URL'])
 342
 343     @mock.patch('boto3.resource')
 344     def test_publish_artifacts_exc4(self, *args):
 345         args[0].return_value.meta.client.head_bucket.side_effect = Exception
 346         self.assertEqual(self.test.publish_artifacts(),
 347                          testcase.TestCase.EX_PUBLISH_ARTIFACTS_ERROR)
 348         args[0].assert_called_once_with(
 349             's3', endpoint_url=os.environ['S3_ENDPOINT_URL'])
 350
 351     @mock.patch('boto3.resource')
 352     def test_publish_artifacts_exc5(self, *args):
 353         error_response = {'Error': {'Code': '403'}}
 354         mock_head_bucket = args[0].return_value.meta.client.head_bucket
 355         mock_head_bucket.side_effect = botocore.exceptions.ClientError(
 356             error_response, 'Foo')
 357         self.assertEqual(self.test.publish_artifacts(),
 358                          testcase.TestCase.EX_PUBLISH_ARTIFACTS_ERROR)
 359         args[0].assert_called_once_with(
 360             's3', endpoint_url=os.environ['S3_ENDPOINT_URL'])
 361
 362     @mock.patch('mimetypes.guess_type', return_value=(None, None))
 363     @mock.patch('boto3.resource')
 364     @mock.patch('os.walk', return_value=[])
 365     def test_publish_artifacts1(self, *args):
 366         self.assertEqual(self.test.publish_artifacts(),
 367                          testcase.TestCase.EX_OK)
 368         args[0].assert_called_once_with(self.test.res_dir)
 369         args[1].assert_called_once_with(
 370             's3', endpoint_url=os.environ['S3_ENDPOINT_URL'])
 371
 372     @mock.patch('mimetypes.guess_type', return_value=(None, None))
 373     @mock.patch('boto3.resource')
 374     @mock.patch('os.walk', return_value=[])
 375     def test_publish_artifacts2(self, *args):
 376         error_response = {'Error': {'Code': '404'}}
 377         mock_head_bucket = args[1].return_value.meta.client.head_bucket
 378         mock_head_bucket.side_effect = botocore.exceptions.ClientError(
 379             error_response, 'NoSuchBucket')
 380         self.assertEqual(self.test.publish_artifacts(),
 381                          testcase.TestCase.EX_OK)
 382         args[0].assert_called_once_with(self.test.res_dir)
 383         args[1].assert_called_once_with(
 384             's3', endpoint_url=os.environ['S3_ENDPOINT_URL'])
 385
 386     @mock.patch('mimetypes.guess_type', return_value=(None, None))
 387     @mock.patch('os.path.exists', return_value=True)
 388     @mock.patch('boto3.resource')
 389     @mock.patch('os.walk',
 390                 return_value=[
 391                     (testcase.TestCase.dir_results, ('',), ('bar',))])
 392     def test_publish_artifacts3(self, *args):
 393         self.assertEqual(self.test.publish_artifacts(),
 394                          testcase.TestCase.EX_OK)
 395         args[0].assert_called_once_with(self.test.res_dir)
 396         expected = [
 397             mock.call('s3', endpoint_url=os.environ['S3_ENDPOINT_URL']),
 398             mock.call().meta.client.head_bucket(Bucket='xtesting'),
 399             mock.call().Bucket('xtesting'),
 400             mock.call().Bucket().upload_file(
 401                 '/var/lib/xtesting/results/xtesting.log',
 402                 'prefix/xtesting.log',
 403                 ExtraArgs={'ContentType': 'application/octet-stream'}),
 404             mock.call().Bucket('xtesting'),
 405             mock.call().Bucket().upload_file(
 406                 '/var/lib/xtesting/results/xtesting.debug.log',
 407                 'prefix/xtesting.debug.log',
 408                 ExtraArgs={'ContentType': 'application/octet-stream'}),
 409             mock.call().Bucket('xtesting'),
 410             mock.call().Bucket().upload_file(
 411                 '/var/lib/xtesting/results/bar', 'prefix/bar',
 412                 ExtraArgs={'ContentType': 'application/octet-stream'})]
 413         self.assertEqual(args[1].mock_calls, expected)
 414
 415     @mock.patch('mimetypes.guess_type', return_value=('text/plain', None))
 416     @mock.patch('os.path.exists', return_value=True)
 417     @mock.patch('boto3.resource')
 418     @mock.patch('os.walk',
 419                 return_value=[
 420                     (testcase.TestCase.dir_results, ('',), ('bar',))])
 421     def test_publish_artifacts4(self, *args):
 422         self.assertEqual(self.test.publish_artifacts(),
 423                          testcase.TestCase.EX_OK)
 424         args[0].assert_called_once_with(self.test.res_dir)
 425         expected = [
 426             mock.call('s3', endpoint_url=os.environ['S3_ENDPOINT_URL']),
 427             mock.call().meta.client.head_bucket(Bucket='xtesting'),
 428             mock.call().Bucket('xtesting'),
 429             mock.call().Bucket().upload_file(
 430                 '/var/lib/xtesting/results/xtesting.log',
 431                 'prefix/xtesting.log',
 432                 ExtraArgs={'ContentType': 'text/plain'}),
 433             mock.call().Bucket('xtesting'),
 434             mock.call().Bucket().upload_file(
 435                 '/var/lib/xtesting/results/xtesting.debug.log',
 436                 'prefix/xtesting.debug.log',
 437                 ExtraArgs={'ContentType': 'text/plain'}),
 438             mock.call().Bucket('xtesting'),
 439             mock.call().Bucket().upload_file(
 440                 '/var/lib/xtesting/results/bar', 'prefix/bar',
 441                 ExtraArgs={'ContentType': 'text/plain'})]
 442         self.assertEqual(args[1].mock_calls, expected)
 443
 444 if __name__ == "__main__":
 445     logging.disable(logging.CRITICAL)
 446     unittest.main(verbosity=2)