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             if urllib.parse.urlparse(url).scheme != "file":
 236                 res_url = req.json()["href"]
 237                 if env.get('TEST_DB_EXT_URL'):
 238                     res_url = res_url.replace(
 239                         env.get('TEST_DB_URL'), env.get('TEST_DB_EXT_URL'))
 240                 self.__logger.info(
 241                     "The results were successfully pushed to DB: \n\n%s\n",
 242                     res_url)
 243         except AssertionError:
 244             self.__logger.exception(
 245                 "Please run test before publishing the results")
 246             return TestCase.EX_PUSH_TO_DB_ERROR
 247         except requests.exceptions.HTTPError:
 248             self.__logger.exception("The HTTP request raises issues")
 249             return TestCase.EX_PUSH_TO_DB_ERROR
 250         except Exception:  # pylint: disable=broad-except
 251             self.__logger.exception("The results cannot be pushed to DB")
 252             return TestCase.EX_PUSH_TO_DB_ERROR
 253         return TestCase.EX_OK
 254
 255     def publish_artifacts(self):  # pylint: disable=too-many-locals
 256         """Push the artifacts to the S3 repository.
 257
 258         It allows publishing the artifacts.
 259
 260         It could be overriden if the common implementation is not
 261         suitable.
 262
 263         The credentials must be configured before publishing the artifacts:
 264
 265             * fill ~/.aws/credentials or ~/.boto,
 266             * set AWS_ACCESS_KEY_ID and AWS_SECRET_ACCESS_KEY in env.
 267
 268         The next vars must be set in env:
 269
 270             * S3_ENDPOINT_URL (http://127.0.0.1:9000),
 271             * S3_DST_URL (s3://xtesting/prefix),
 272             * HTTP_DST_URL (http://127.0.0.1/prefix).
 273
 274         Returns:
 275             TestCase.EX_OK if artifacts were published to repository.
 276             TestCase.EX_PUBLISH_ARTIFACTS_ERROR otherwise.
 277         """
 278         try:
 279             b3resource = boto3.resource(
 280                 's3', endpoint_url=os.environ["S3_ENDPOINT_URL"])
 281             dst_s3_url = os.environ["S3_DST_URL"]
 282             bucket_name = urllib.parse.urlparse(dst_s3_url).netloc
 283             try:
 284                 b3resource.meta.client.head_bucket(Bucket=bucket_name)
 285             except botocore.exceptions.ClientError as exc:
 286                 error_code = exc.response['Error']['Code']
 287                 if error_code == '404':
 288                     # pylint: disable=no-member
 289                     b3resource.create_bucket(Bucket=bucket_name)
 290                 else:
 291                     typ, value, traceback = sys.exc_info()
 292                     six.reraise(typ, value, traceback)
 293             except Exception:  # pylint: disable=broad-except
 294                 typ, value, traceback = sys.exc_info()
 295                 six.reraise(typ, value, traceback)
 296             path = urllib.parse.urlparse(dst_s3_url).path.strip("/")
 297             output_str = "\n"
 298             for root, _, files in os.walk(self.dir_results):
 299                 for pub_file in files:
 300                     # pylint: disable=no-member
 301                     b3resource.Bucket(bucket_name).upload_file(
 302                         os.path.join(root, pub_file),
 303                         os.path.join(path, os.path.relpath(
 304                             os.path.join(root, pub_file),
 305                             start=self.dir_results)))
 306                     dst_http_url = os.environ["HTTP_DST_URL"]
 307                     output_str += "\n{}".format(
 308                         os.path.join(dst_http_url, os.path.relpath(
 309                             os.path.join(root, pub_file),
 310                             start=self.dir_results)))
 311             self.__logger.info(
 312                 "All artifacts were successfully published: %s\n", output_str)
 313             return TestCase.EX_OK
 314         except KeyError as ex:
 315             self.__logger.error("Please check env var: %s", str(ex))
 316             return TestCase.EX_PUBLISH_ARTIFACTS_ERROR
 317         except botocore.exceptions.NoCredentialsError:
 318             self.__logger.error(
 319                 "Please fill ~/.aws/credentials, ~/.boto or set "
 320                 "AWS_ACCESS_KEY_ID and AWS_SECRET_ACCESS_KEY in env")
 321             return TestCase.EX_PUBLISH_ARTIFACTS_ERROR
 322         except Exception:  # pylint: disable=broad-except
 323             self.__logger.exception("Cannot publish the artifacts")
 324             return TestCase.EX_PUBLISH_ARTIFACTS_ERROR
 325
 326     def clean(self):
 327         """Clean the resources.
 328
 329         It can be overriden if resources must be deleted after
 330         running the test case.
 331         """