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__)
62 def __init__(self, **kwargs):
64 self.project_name = os.environ.get(
65 'PROJECT_NAME', kwargs.get('project_name', 'xtesting'))
66 self.case_name = kwargs.get('case_name', '')
67 self.criteria = kwargs.get('criteria', 100)
71 self.is_skipped = False
72 self.output_log_name = os.path.basename(constants.LOG_PATH)
73 self.output_debug_log_name = os.path.basename(constants.DEBUG_LOG_PATH)
74 self.res_dir = os.path.join(self.dir_results, self.case_name)
78 assert self.project_name
83 result = 'PASS' if (self.is_successful(
84 ) == TestCase.EX_OK) else 'FAIL'
85 msg = prettytable.PrettyTable(
86 header_style='upper', padding_width=5,
87 field_names=['test case', 'project', 'duration',
89 msg.add_row([self.case_name, self.project_name,
90 self.get_duration(), result])
91 return msg.get_string()
92 except AssertionError:
93 self.__logger.error("We cannot print invalid objects")
94 return super().__str__()
96 def get_duration(self):
97 """Return the duration of the test case.
100 duration if start_time and stop_time are set
106 assert self.start_time
107 assert self.stop_time
108 if self.stop_time < self.start_time:
111 f"{str(int(self.stop_time - self.start_time) // 60).zfill(2)}:"
112 f"{str(int(self.stop_time - self.start_time) % 60).zfill(2)}")
114 except Exception: # pylint: disable=broad-except
115 self.__logger.error("Please run test before getting the duration")
118 def is_successful(self):
119 """Interpret the result of the test case.
121 It allows getting the result of TestCase. It completes run()
122 which only returns the execution status.
124 It can be overriden if checking result is not suitable.
127 TestCase.EX_OK if result is 'PASS'.
128 TestCase.EX_TESTCASE_SKIPPED if test case is skipped.
129 TestCase.EX_TESTCASE_FAILED otherwise.
133 return TestCase.EX_TESTCASE_SKIPPED
135 assert self.result is not None
136 if (not isinstance(self.result, str) and
137 not isinstance(self.criteria, str)):
138 if self.result >= self.criteria:
139 return TestCase.EX_OK
141 # Backward compatibility
142 # It must be removed as soon as TestCase subclasses
143 # stop setting result = 'PASS' or 'FAIL'.
144 # In this case criteria is unread.
145 self.__logger.warning(
146 "Please update result which must be an int!")
147 if self.result == 'PASS':
148 return TestCase.EX_OK
149 except AssertionError:
150 self.__logger.error("Please run test before checking the results")
151 return TestCase.EX_TESTCASE_FAILED
153 def check_requirements(self):
154 """Check the requirements of the test case.
156 It can be overriden on purpose.
158 self.is_skipped = False
161 def run(self, **kwargs):
162 """Run the test case.
164 It allows running TestCase and getting its execution
167 The subclasses must override the default implementation which
170 The new implementation must set the following attributes to
171 push the results to DB:
178 kwargs: Arbitrary keyword arguments.
181 @decorators.can_dump_request_to_file
182 def push_to_db(self):
183 """Push the results of the test case to the DB.
185 It allows publishing the results and checking the status.
187 It could be overriden if the common implementation is not
190 The following attributes must be set before pushing the results to DB:
198 The next vars must be set in env:
207 TestCase.EX_OK if results were pushed to DB.
208 TestCase.EX_PUSH_TO_DB_ERROR otherwise.
212 return TestCase.EX_PUSH_TO_DB_ERROR
213 assert self.project_name
214 assert self.case_name
215 assert self.start_time
216 assert self.stop_time
217 url = env.get('TEST_DB_URL')
218 data = {"project_name": self.project_name,
219 "case_name": self.case_name,
220 "details": self.details}
221 data["installer"] = env.get('INSTALLER_TYPE')
222 data["scenario"] = env.get('DEPLOY_SCENARIO')
223 data["pod_name"] = env.get('NODE_NAME')
224 data["build_tag"] = env.get('BUILD_TAG')
225 data["criteria"] = 'PASS' if self.is_successful(
226 ) == TestCase.EX_OK else 'FAIL'
227 data["start_date"] = datetime.fromtimestamp(
228 self.start_time).strftime('%Y-%m-%d %H:%M:%S')
229 data["stop_date"] = datetime.fromtimestamp(
230 self.stop_time).strftime('%Y-%m-%d %H:%M:%S')
232 data["version"] = re.search(
233 TestCase._job_name_rule,
234 env.get('BUILD_TAG')).group(2)
235 except Exception: # pylint: disable=broad-except
236 data["version"] = "unknown"
238 url, data=json.dumps(data, sort_keys=True),
239 headers=self.headers,
241 req.raise_for_status()
242 if urlparse(url).scheme != "file":
243 # href must be postprocessed as OPNFV testapi is misconfigured
244 # (localhost is returned)
245 uid = re.sub(r'^.*/api/v1/results/*', '', req.json()["href"])
246 netloc = env.get('TEST_DB_EXT_URL') if env.get(
247 'TEST_DB_EXT_URL') else env.get('TEST_DB_URL')
249 "The results were successfully pushed to DB: \n\n%s\n",
250 os.path.join(netloc, uid))
251 except AssertionError:
252 self.__logger.exception(
253 "Please run test before publishing the results")
254 return TestCase.EX_PUSH_TO_DB_ERROR
255 except requests.exceptions.HTTPError:
256 self.__logger.exception("The HTTP request raises issues")
257 return TestCase.EX_PUSH_TO_DB_ERROR
258 except Exception: # pylint: disable=broad-except
259 self.__logger.exception("The results cannot be pushed to DB")
260 return TestCase.EX_PUSH_TO_DB_ERROR
261 return TestCase.EX_OK
263 def publish_artifacts(self): # pylint: disable=too-many-locals
264 """Push the artifacts to the S3 repository.
266 It allows publishing the artifacts.
268 It could be overriden if the common implementation is not
271 The credentials must be configured before publishing the artifacts:
273 * fill ~/.aws/credentials or ~/.boto,
274 * set AWS_ACCESS_KEY_ID and AWS_SECRET_ACCESS_KEY in env.
276 The next vars must be set in env:
278 * S3_ENDPOINT_URL (http://127.0.0.1:9000),
279 * S3_DST_URL (s3://xtesting/prefix),
280 * HTTP_DST_URL (http://127.0.0.1/prefix).
283 TestCase.EX_OK if artifacts were published to repository.
284 TestCase.EX_PUBLISH_ARTIFACTS_ERROR otherwise.
287 b3resource = boto3.resource(
288 's3', endpoint_url=os.environ["S3_ENDPOINT_URL"])
289 dst_s3_url = os.environ["S3_DST_URL"]
290 multipart_threshold = 5 * 1024 ** 5 if "google" in os.environ[
291 "S3_ENDPOINT_URL"] else 8 * 1024 * 1024
292 config = TransferConfig(multipart_threshold=multipart_threshold)
293 bucket_name = urlparse(dst_s3_url).netloc
295 b3resource.meta.client.head_bucket(Bucket=bucket_name)
296 except botocore.exceptions.ClientError as exc:
297 error_code = exc.response['Error']['Code']
298 if error_code == '404':
299 # pylint: disable=no-member
300 b3resource.create_bucket(Bucket=bucket_name)
303 except Exception as exc: # pylint: disable=broad-except
305 path = urlparse(dst_s3_url).path.strip("/")
306 dst_http_url = os.environ["HTTP_DST_URL"]
308 # protects if test cases return details as None
309 self.details = self.details or {}
310 self.details["links"] = []
311 for log_file in [self.output_log_name, self.output_debug_log_name]:
312 if os.path.exists(os.path.join(self.dir_results, log_file)):
313 abs_file = os.path.join(self.dir_results, log_file)
314 mime_type = mimetypes.guess_type(abs_file)
316 "Publishing %s %s", abs_file, mime_type)
317 # pylint: disable=no-member
318 b3resource.Bucket(bucket_name).upload_file(
319 abs_file, os.path.join(path, log_file), Config=config,
320 ExtraArgs={'ContentType': mime_type[
321 0] or 'application/octet-stream'})
322 link = os.path.join(dst_http_url, log_file)
323 output_str += f"\n{link}"
324 self.details["links"].append(link)
325 for root, _, files in os.walk(self.res_dir):
326 for pub_file in files:
327 abs_file = os.path.join(root, pub_file)
328 mime_type = mimetypes.guess_type(abs_file)
330 "Publishing %s %s", abs_file, mime_type)
331 # pylint: disable=no-member
332 b3resource.Bucket(bucket_name).upload_file(
334 os.path.join(path, os.path.relpath(
335 os.path.join(root, pub_file),
336 start=self.dir_results)),
338 ExtraArgs={'ContentType': mime_type[
339 0] or 'application/octet-stream'})
340 link = os.path.join(dst_http_url, os.path.relpath(
341 os.path.join(root, pub_file),
342 start=self.dir_results))
343 output_str += f"\n{link}"
344 self.details["links"].append(link)
346 "All artifacts were successfully published: %s\n", output_str)
347 return TestCase.EX_OK
348 except KeyError as ex:
349 self.__logger.error("Please check env var: %s", str(ex))
350 return TestCase.EX_PUBLISH_ARTIFACTS_ERROR
351 except botocore.exceptions.NoCredentialsError:
353 "Please fill ~/.aws/credentials, ~/.boto or set "
354 "AWS_ACCESS_KEY_ID and AWS_SECRET_ACCESS_KEY in env")
355 return TestCase.EX_PUBLISH_ARTIFACTS_ERROR
356 except Exception: # pylint: disable=broad-except
357 self.__logger.exception("Cannot publish the artifacts")
358 return TestCase.EX_PUBLISH_ARTIFACTS_ERROR
361 """Clean the resources.
363 It can be overriden if resources must be deleted after
364 running the test case.