3 # Copyright (c) 2016 Orange and others.
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
10 """Define the parent class of all Xtesting TestCases."""
13 from datetime import datetime
20 from urllib.parse import urlparse
22 from boto3.s3.transfer import TransferConfig
27 from xtesting.utils import decorators
28 from xtesting.utils import env
29 from xtesting.utils import constants
31 __author__ = "Cedric Ollivier <cedric.ollivier@orange.com>"
34 class TestCase(metaclass=abc.ABCMeta):
35 # pylint: disable=too-many-instance-attributes
36 """Base model for single test case."""
39 """everything is OK"""
41 EX_RUN_ERROR = os.EX_SOFTWARE
44 EX_PUSH_TO_DB_ERROR = os.EX_SOFTWARE - 1
45 """push_to_db() failed"""
47 EX_TESTCASE_FAILED = os.EX_SOFTWARE - 2
48 """results are false"""
50 EX_TESTCASE_SKIPPED = os.EX_SOFTWARE - 3
51 """requirements are unmet"""
53 EX_PUBLISH_ARTIFACTS_ERROR = os.EX_SOFTWARE - 4
54 """publish_artifacts() failed"""
56 dir_results = constants.RESULTS_DIR
57 _job_name_rule = "(dai|week)ly-(.+?)-[0-9]*"
58 headers = {'Content-Type': 'application/json'}
59 __logger = logging.getLogger(__name__)
61 def __init__(self, **kwargs):
63 self.project_name = kwargs.get('project_name', 'xtesting')
64 self.case_name = kwargs.get('case_name', '')
65 self.criteria = kwargs.get('criteria', 100)
69 self.is_skipped = False
70 self.output_log_name = os.path.basename(constants.LOG_PATH)
71 self.output_debug_log_name = os.path.basename(constants.DEBUG_LOG_PATH)
72 self.res_dir = os.path.join(self.dir_results, self.case_name)
76 assert self.project_name
81 result = 'PASS' if(self.is_successful(
82 ) == TestCase.EX_OK) else 'FAIL'
83 msg = prettytable.PrettyTable(
84 header_style='upper', padding_width=5,
85 field_names=['test case', 'project', 'duration',
87 msg.add_row([self.case_name, self.project_name,
88 self.get_duration(), result])
89 return msg.get_string()
90 except AssertionError:
91 self.__logger.error("We cannot print invalid objects")
92 return super(TestCase, self).__str__()
94 def get_duration(self):
95 """Return the duration of the test case.
98 duration if start_time and stop_time are set
104 assert self.start_time
105 assert self.stop_time
106 if self.stop_time < self.start_time:
108 return "{}:{}".format(
109 str(int(self.stop_time - self.start_time) // 60).zfill(2),
110 str(int(self.stop_time - self.start_time) % 60).zfill(2))
111 except Exception: # pylint: disable=broad-except
112 self.__logger.error("Please run test before getting the duration")
115 def is_successful(self):
116 """Interpret the result of the test case.
118 It allows getting the result of TestCase. It completes run()
119 which only returns the execution status.
121 It can be overriden if checking result is not suitable.
124 TestCase.EX_OK if result is 'PASS'.
125 TestCase.EX_TESTCASE_SKIPPED if test case is skipped.
126 TestCase.EX_TESTCASE_FAILED otherwise.
130 return TestCase.EX_TESTCASE_SKIPPED
132 assert self.result is not None
133 if (not isinstance(self.result, str) and
134 not isinstance(self.criteria, str)):
135 if self.result >= self.criteria:
136 return TestCase.EX_OK
138 # Backward compatibility
139 # It must be removed as soon as TestCase subclasses
140 # stop setting result = 'PASS' or 'FAIL'.
141 # In this case criteria is unread.
142 self.__logger.warning(
143 "Please update result which must be an int!")
144 if self.result == 'PASS':
145 return TestCase.EX_OK
146 except AssertionError:
147 self.__logger.error("Please run test before checking the results")
148 return TestCase.EX_TESTCASE_FAILED
150 def check_requirements(self): # pylint: disable=no-self-use
151 """Check the requirements of the test case.
153 It can be overriden on purpose.
155 self.is_skipped = False
158 def run(self, **kwargs):
159 """Run the test case.
161 It allows running TestCase and getting its execution
164 The subclasses must override the default implementation which
167 The new implementation must set the following attributes to
168 push the results to DB:
175 kwargs: Arbitrary keyword arguments.
178 @decorators.can_dump_request_to_file
179 def push_to_db(self):
180 """Push the results of the test case to the DB.
182 It allows publishing the results and checking the status.
184 It could be overriden if the common implementation is not
187 The following attributes must be set before pushing the results to DB:
195 The next vars must be set in env:
204 TestCase.EX_OK if results were pushed to DB.
205 TestCase.EX_PUSH_TO_DB_ERROR otherwise.
209 return TestCase.EX_PUSH_TO_DB_ERROR
210 assert self.project_name
211 assert self.case_name
212 assert self.start_time
213 assert self.stop_time
214 url = env.get('TEST_DB_URL')
215 data = {"project_name": self.project_name,
216 "case_name": self.case_name,
217 "details": self.details}
218 data["installer"] = env.get('INSTALLER_TYPE')
219 data["scenario"] = env.get('DEPLOY_SCENARIO')
220 data["pod_name"] = env.get('NODE_NAME')
221 data["build_tag"] = env.get('BUILD_TAG')
222 data["criteria"] = 'PASS' if self.is_successful(
223 ) == TestCase.EX_OK else 'FAIL'
224 data["start_date"] = datetime.fromtimestamp(
225 self.start_time).strftime('%Y-%m-%d %H:%M:%S')
226 data["stop_date"] = datetime.fromtimestamp(
227 self.stop_time).strftime('%Y-%m-%d %H:%M:%S')
229 data["version"] = re.search(
230 TestCase._job_name_rule,
231 env.get('BUILD_TAG')).group(2)
232 except Exception: # pylint: disable=broad-except
233 data["version"] = "unknown"
235 url, data=json.dumps(data, sort_keys=True),
236 headers=self.headers)
237 req.raise_for_status()
238 if urlparse(url).scheme != "file":
239 # href must be postprocessed as OPNFV testapi is misconfigured
240 # (localhost is returned)
241 uid = re.sub(r'^.*/api/v1/results/*', '', req.json()["href"])
242 netloc = env.get('TEST_DB_EXT_URL') if env.get(
243 'TEST_DB_EXT_URL') else env.get('TEST_DB_URL')
245 "The results were successfully pushed to DB: \n\n%s\n",
246 os.path.join(netloc, uid))
247 except AssertionError:
248 self.__logger.exception(
249 "Please run test before publishing the results")
250 return TestCase.EX_PUSH_TO_DB_ERROR
251 except requests.exceptions.HTTPError:
252 self.__logger.exception("The HTTP request raises issues")
253 return TestCase.EX_PUSH_TO_DB_ERROR
254 except Exception: # pylint: disable=broad-except
255 self.__logger.exception("The results cannot be pushed to DB")
256 return TestCase.EX_PUSH_TO_DB_ERROR
257 return TestCase.EX_OK
259 def publish_artifacts(self): # pylint: disable=too-many-locals
260 """Push the artifacts to the S3 repository.
262 It allows publishing the artifacts.
264 It could be overriden if the common implementation is not
267 The credentials must be configured before publishing the artifacts:
269 * fill ~/.aws/credentials or ~/.boto,
270 * set AWS_ACCESS_KEY_ID and AWS_SECRET_ACCESS_KEY in env.
272 The next vars must be set in env:
274 * S3_ENDPOINT_URL (http://127.0.0.1:9000),
275 * S3_DST_URL (s3://xtesting/prefix),
276 * HTTP_DST_URL (http://127.0.0.1/prefix).
279 TestCase.EX_OK if artifacts were published to repository.
280 TestCase.EX_PUBLISH_ARTIFACTS_ERROR otherwise.
283 b3resource = boto3.resource(
284 's3', endpoint_url=os.environ["S3_ENDPOINT_URL"])
285 dst_s3_url = os.environ["S3_DST_URL"]
286 multipart_threshold = 5 * 1024 ** 5 if "google" in os.environ[
287 "S3_ENDPOINT_URL"] else 8 * 1024 * 1024
288 config = TransferConfig(multipart_threshold=multipart_threshold)
289 bucket_name = urlparse(dst_s3_url).netloc
291 b3resource.meta.client.head_bucket(Bucket=bucket_name)
292 except botocore.exceptions.ClientError as exc:
293 error_code = exc.response['Error']['Code']
294 if error_code == '404':
295 # pylint: disable=no-member
296 b3resource.create_bucket(Bucket=bucket_name)
299 except Exception as exc: # pylint: disable=broad-except
301 path = urlparse(dst_s3_url).path.strip("/")
302 dst_http_url = os.environ["HTTP_DST_URL"]
304 self.details["links"] = []
305 for log_file in [self.output_log_name, self.output_debug_log_name]:
306 if os.path.exists(os.path.join(self.dir_results, log_file)):
307 abs_file = os.path.join(self.dir_results, log_file)
308 mime_type = mimetypes.guess_type(abs_file)
310 "Publishing %s %s", abs_file, mime_type)
311 # pylint: disable=no-member
312 b3resource.Bucket(bucket_name).upload_file(
313 abs_file, os.path.join(path, log_file), Config=config,
314 ExtraArgs={'ContentType': mime_type[
315 0] or 'application/octet-stream'})
316 link = os.path.join(dst_http_url, log_file)
317 output_str += "\n{}".format(link)
318 self.details["links"].append(link)
319 for root, _, files in os.walk(self.res_dir):
320 for pub_file in files:
321 abs_file = os.path.join(root, pub_file)
322 mime_type = mimetypes.guess_type(abs_file)
324 "Publishing %s %s", abs_file, mime_type)
325 # pylint: disable=no-member
326 b3resource.Bucket(bucket_name).upload_file(
328 os.path.join(path, os.path.relpath(
329 os.path.join(root, pub_file),
330 start=self.dir_results)),
332 ExtraArgs={'ContentType': mime_type[
333 0] or 'application/octet-stream'})
334 link = os.path.join(dst_http_url, os.path.relpath(
335 os.path.join(root, pub_file),
336 start=self.dir_results))
337 output_str += "\n{}".format(link)
338 self.details["links"].append(link)
340 "All artifacts were successfully published: %s\n", output_str)
341 return TestCase.EX_OK
342 except KeyError as ex:
343 self.__logger.error("Please check env var: %s", str(ex))
344 return TestCase.EX_PUBLISH_ARTIFACTS_ERROR
345 except botocore.exceptions.NoCredentialsError:
347 "Please fill ~/.aws/credentials, ~/.boto or set "
348 "AWS_ACCESS_KEY_ID and AWS_SECRET_ACCESS_KEY in env")
349 return TestCase.EX_PUBLISH_ARTIFACTS_ERROR
350 except Exception: # pylint: disable=broad-except
351 self.__logger.exception("Cannot publish the artifacts")
352 return TestCase.EX_PUBLISH_ARTIFACTS_ERROR
355 """Clean the resources.
357 It can be overriden if resources must be deleted after
358 running the test case.