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