xtesting/core/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 """Define the parent class of all Xtesting TestCases."""
  11
  12 import abc
  13 from datetime import datetime
  14 import json
  15 import logging
  16 import os
  17 import re
  18 import sys
  19
  20 import boto3
  21 import botocore
  22 import prettytable
  23 import requests
  24 import six
  25 from six.moves import urllib
  26
  27 from xtesting.utils import decorators
  28 from xtesting.utils import env
  29
  30 __author__ = "Cedric Ollivier <cedric.ollivier@orange.com>"
  31
  32
  33 @six.add_metaclass(abc.ABCMeta)
  34 class TestCase():
  35     # pylint: disable=too-many-instance-attributes
  36     """Base model for single test case."""
  37
  38     EX_OK = os.EX_OK
  39     """everything is OK"""
  40
  41     EX_RUN_ERROR = os.EX_SOFTWARE
  42     """run() failed"""
  43
  44     EX_PUSH_TO_DB_ERROR = os.EX_SOFTWARE - 1
  45     """push_to_db() failed"""
  46
  47     EX_TESTCASE_FAILED = os.EX_SOFTWARE - 2
  48     """results are false"""
  49
  50     EX_TESTCASE_SKIPPED = os.EX_SOFTWARE - 3
  51     """requirements are unmet"""
  52
  53     EX_PUBLISH_ARTIFACTS_ERROR = os.EX_SOFTWARE - 4
  54     """publish_artifacts() failed"""
  55
  56     dir_results = "/var/lib/xtesting/results"
  57     _job_name_rule = "(dai|week)ly-(.+?)-[0-9]*"
  58     _headers = {'Content-Type': 'application/json'}
  59     __logger = logging.getLogger(__name__)
  60
  61     def __init__(self, **kwargs):
  62         self.details = {}
  63         self.project_name = kwargs.get('project_name', 'xtesting')
  64         self.case_name = kwargs.get('case_name', '')
  65         self.criteria = kwargs.get('criteria', 100)
  66         self.result = 0
  67         self.start_time = 0
  68         self.stop_time = 0
  69         self.is_skipped = False
  70         self.res_dir = "{}/{}".format(self.dir_results, self.case_name)
  71
  72     def __str__(self):
  73         try:
  74             assert self.project_name
  75             assert self.case_name
  76             if self.is_skipped:
  77                 result = 'SKIP'
  78             else:
  79                 result = 'PASS' if(self.is_successful(
  80                     ) == TestCase.EX_OK) else 'FAIL'
  81             msg = prettytable.PrettyTable(
  82                 header_style='upper', padding_width=5,
  83                 field_names=['test case', 'project', 'duration',
  84                              'result'])
  85             msg.add_row([self.case_name, self.project_name,
  86                          self.get_duration(), result])
  87             return msg.get_string()
  88         except AssertionError:
  89             self.__logger.error("We cannot print invalid objects")
  90             return super(TestCase, self).__str__()
  91
  92     def get_duration(self):
  93         """Return the duration of the test case.
  94
  95         Returns:
  96             duration if start_time and stop_time are set
  97             "XX:XX" otherwise.
  98         """
  99         try:
 100             if self.is_skipped:
 101                 return "00:00"
 102             assert self.start_time
 103             assert self.stop_time
 104             if self.stop_time < self.start_time:
 105                 return "XX:XX"
 106             return "{0[0]:02.0f}:{0[1]:02.0f}".format(divmod(
 107                 self.stop_time - self.start_time, 60))
 108         except Exception:  # pylint: disable=broad-except
 109             self.__logger.error("Please run test before getting the duration")
 110             return "XX:XX"
 111
 112     def is_successful(self):
 113         """Interpret the result of the test case.
 114
 115         It allows getting the result of TestCase. It completes run()
 116         which only returns the execution status.
 117
 118         It can be overriden if checking result is not suitable.
 119
 120         Returns:
 121             TestCase.EX_OK if result is 'PASS'.
 122             TestCase.EX_TESTCASE_SKIPPED if test case is skipped.
 123             TestCase.EX_TESTCASE_FAILED otherwise.
 124         """
 125         try:
 126             if self.is_skipped:
 127                 return TestCase.EX_TESTCASE_SKIPPED
 128             assert self.criteria
 129             assert self.result is not None
 130             if (not isinstance(self.result, str) and
 131                     not isinstance(self.criteria, str)):
 132                 if self.result >= self.criteria:
 133                     return TestCase.EX_OK
 134             else:
 135                 # Backward compatibility
 136                 # It must be removed as soon as TestCase subclasses
 137                 # stop setting result = 'PASS' or 'FAIL'.
 138                 # In this case criteria is unread.
 139                 self.__logger.warning(
 140                     "Please update result which must be an int!")
 141                 if self.result == 'PASS':
 142                     return TestCase.EX_OK
 143         except AssertionError:
 144             self.__logger.error("Please run test before checking the results")
 145         return TestCase.EX_TESTCASE_FAILED
 146
 147     def check_requirements(self):  # pylint: disable=no-self-use
 148         """Check the requirements of the test case.
 149
 150         It can be overriden on purpose.
 151         """
 152         self.is_skipped = False
 153
 154     @abc.abstractmethod
 155     def run(self, **kwargs):
 156         """Run the test case.
 157
 158         It allows running TestCase and getting its execution
 159         status.
 160
 161         The subclasses must override the default implementation which
 162         is false on purpose.
 163
 164         The new implementation must set the following attributes to
 165         push the results to DB:
 166
 167             * result,
 168             * start_time,
 169             * stop_time.
 170
 171         Args:
 172             kwargs: Arbitrary keyword arguments.
 173         """
 174
 175     @decorators.can_dump_request_to_file
 176     def push_to_db(self):
 177         """Push the results of the test case to the DB.
 178
 179         It allows publishing the results and checking the status.
 180
 181         It could be overriden if the common implementation is not
 182         suitable.
 183
 184         The following attributes must be set before pushing the results to DB:
 185
 186             * project_name,
 187             * case_name,
 188             * result,
 189             * start_time,
 190             * stop_time.
 191
 192         The next vars must be set in env:
 193
 194             * TEST_DB_URL,
 195             * INSTALLER_TYPE,
 196             * DEPLOY_SCENARIO,
 197             * NODE_NAME,
 198             * BUILD_TAG.
 199
 200         Returns:
 201             TestCase.EX_OK if results were pushed to DB.
 202             TestCase.EX_PUSH_TO_DB_ERROR otherwise.
 203         """
 204         try:
 205             if self.is_skipped:
 206                 return TestCase.EX_PUSH_TO_DB_ERROR
 207             assert self.project_name
 208             assert self.case_name
 209             assert self.start_time
 210             assert self.stop_time
 211             url = env.get('TEST_DB_URL')
 212             data = {"project_name": self.project_name,
 213                     "case_name": self.case_name,
 214                     "details": self.details}
 215             data["installer"] = env.get('INSTALLER_TYPE')
 216             data["scenario"] = env.get('DEPLOY_SCENARIO')
 217             data["pod_name"] = env.get('NODE_NAME')
 218             data["build_tag"] = env.get('BUILD_TAG')
 219             data["criteria"] = 'PASS' if self.is_successful(
 220                 ) == TestCase.EX_OK else 'FAIL'
 221             data["start_date"] = datetime.fromtimestamp(
 222                 self.start_time).strftime('%Y-%m-%d %H:%M:%S')
 223             data["stop_date"] = datetime.fromtimestamp(
 224                 self.stop_time).strftime('%Y-%m-%d %H:%M:%S')
 225             try:
 226                 data["version"] = re.search(
 227                     TestCase._job_name_rule,
 228                     env.get('BUILD_TAG')).group(2)
 229             except Exception:  # pylint: disable=broad-except
 230                 data["version"] = "unknown"
 231             req = requests.post(
 232                 url, data=json.dumps(data, sort_keys=True),
 233                 headers=self._headers)
 234             req.raise_for_status()
 235             self.__logger.info(
 236                 "The results were successfully pushed to DB")
 237         except AssertionError:
 238             self.__logger.exception(
 239                 "Please run test before publishing the results")
 240             return TestCase.EX_PUSH_TO_DB_ERROR
 241         except requests.exceptions.HTTPError:
 242             self.__logger.exception("The HTTP request raises issues")
 243             return TestCase.EX_PUSH_TO_DB_ERROR
 244         except Exception:  # pylint: disable=broad-except
 245             self.__logger.exception("The results cannot be pushed to DB")
 246             return TestCase.EX_PUSH_TO_DB_ERROR
 247         return TestCase.EX_OK
 248
 249     def publish_artifacts(self):  # pylint: disable=too-many-locals
 250         """Push the artifacts to the S3 repository.
 251
 252         It allows publishing the artifacts.
 253
 254         It could be overriden if the common implementation is not
 255         suitable.
 256
 257         The credentials must be configured before publishing the artifacts:
 258
 259             * fill ~/.aws/credentials or ~/.boto,
 260             * set AWS_ACCESS_KEY_ID and AWS_SECRET_ACCESS_KEY in env.
 261
 262         The next vars must be set in env:
 263
 264             * S3_ENDPOINT_URL (http://127.0.0.1:9000),
 265             * S3_DST_URL (s3://xtesting/prefix),
 266             * HTTP_DST_URL (http://127.0.0.1/prefix).
 267
 268         Returns:
 269             TestCase.EX_OK if artifacts were published to repository.
 270             TestCase.EX_PUBLISH_ARTIFACTS_ERROR otherwise.
 271         """
 272         try:
 273             b3resource = boto3.resource(
 274                 's3', endpoint_url=os.environ["S3_ENDPOINT_URL"])
 275             dst_s3_url = os.environ["S3_DST_URL"]
 276             bucket_name = urllib.parse.urlparse(dst_s3_url).netloc
 277             try:
 278                 b3resource.meta.client.head_bucket(Bucket=bucket_name)
 279             except botocore.exceptions.ClientError as exc:
 280                 error_code = exc.response['Error']['Code']
 281                 if error_code == '404':
 282                     # pylint: disable=no-member
 283                     b3resource.create_bucket(Bucket=bucket_name)
 284                 else:
 285                     typ, value, traceback = sys.exc_info()
 286                     six.reraise(typ, value, traceback)
 287             except Exception:  # pylint: disable=broad-except
 288                 typ, value, traceback = sys.exc_info()
 289                 six.reraise(typ, value, traceback)
 290             path = urllib.parse.urlparse(dst_s3_url).path.strip("/")
 291             output_str = "\n"
 292             for root, _, files in os.walk(self.dir_results):
 293                 for pub_file in files:
 294                     # pylint: disable=no-member
 295                     b3resource.Bucket(bucket_name).upload_file(
 296                         os.path.join(root, pub_file),
 297                         os.path.join(path, os.path.relpath(
 298                             os.path.join(root, pub_file),
 299                             start=self.dir_results)))
 300                     dst_http_url = os.environ["HTTP_DST_URL"]
 301                     output_str += "\n{}".format(
 302                         os.path.join(dst_http_url, os.path.relpath(
 303                             os.path.join(root, pub_file),
 304                             start=self.dir_results)))
 305             self.__logger.info(
 306                 "All artifacts were successfully published: %s\n", output_str)
 307             return TestCase.EX_OK
 308         except KeyError as ex:
 309             self.__logger.error("Please check env var: %s", str(ex))
 310             return TestCase.EX_PUBLISH_ARTIFACTS_ERROR
 311         except botocore.exceptions.NoCredentialsError:
 312             self.__logger.error(
 313                 "Please fill ~/.aws/credentials, ~/.boto or set "
 314                 "AWS_ACCESS_KEY_ID and AWS_SECRET_ACCESS_KEY in env")
 315             return TestCase.EX_PUBLISH_ARTIFACTS_ERROR
 316         except Exception:  # pylint: disable=broad-except
 317             self.__logger.exception("Cannot publish the artifacts")
 318             return TestCase.EX_PUBLISH_ARTIFACTS_ERROR
 319
 320     def clean(self):
 321         """Clean the resources.
 322
 323         It can be overriden if resources must be deleted after
 324         running the test case.
 325         """