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
25 from six.moves import urllib
27 from xtesting.utils import decorators
28 from xtesting.utils import env
30 __author__ = "Cedric Ollivier <cedric.ollivier@orange.com>"
33 @six.add_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 = "/var/lib/xtesting/results"
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.res_dir = "{}/{}".format(self.dir_results, self.case_name)
74 assert self.project_name
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',
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__()
92 def get_duration(self):
93 """Return the duration of the test case.
96 duration if start_time and stop_time are set
102 assert self.start_time
103 assert self.stop_time
104 if self.stop_time < self.start_time:
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")
112 def is_successful(self):
113 """Interpret the result of the test case.
115 It allows getting the result of TestCase. It completes run()
116 which only returns the execution status.
118 It can be overriden if checking result is not suitable.
121 TestCase.EX_OK if result is 'PASS'.
122 TestCase.EX_TESTCASE_SKIPPED if test case is skipped.
123 TestCase.EX_TESTCASE_FAILED otherwise.
127 return TestCase.EX_TESTCASE_SKIPPED
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
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
147 def check_requirements(self): # pylint: disable=no-self-use
148 """Check the requirements of the test case.
150 It can be overriden on purpose.
152 self.is_skipped = False
155 def run(self, **kwargs):
156 """Run the test case.
158 It allows running TestCase and getting its execution
161 The subclasses must override the default implementation which
164 The new implementation must set the following attributes to
165 push the results to DB:
172 kwargs: Arbitrary keyword arguments.
175 @decorators.can_dump_request_to_file
176 def push_to_db(self):
177 """Push the results of the test case to the DB.
179 It allows publishing the results and checking the status.
181 It could be overriden if the common implementation is not
184 The following attributes must be set before pushing the results to DB:
192 The next vars must be set in env:
201 TestCase.EX_OK if results were pushed to DB.
202 TestCase.EX_PUSH_TO_DB_ERROR otherwise.
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')
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"
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'))
241 "The results were successfully pushed to DB: \n\n%s\n",
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
255 def publish_artifacts(self): # pylint: disable=too-many-locals
256 """Push the artifacts to the S3 repository.
258 It allows publishing the artifacts.
260 It could be overriden if the common implementation is not
263 The credentials must be configured before publishing the artifacts:
265 * fill ~/.aws/credentials or ~/.boto,
266 * set AWS_ACCESS_KEY_ID and AWS_SECRET_ACCESS_KEY in env.
268 The next vars must be set in env:
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).
275 TestCase.EX_OK if artifacts were published to repository.
276 TestCase.EX_PUBLISH_ARTIFACTS_ERROR otherwise.
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
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)
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("/")
298 self.details["links"] = []
299 for root, _, files in os.walk(self.dir_results):
300 for pub_file in files:
301 # pylint: disable=no-member
302 b3resource.Bucket(bucket_name).upload_file(
303 os.path.join(root, pub_file),
304 os.path.join(path, os.path.relpath(
305 os.path.join(root, pub_file),
306 start=self.dir_results)))
307 dst_http_url = os.environ["HTTP_DST_URL"]
308 link = os.path.join(dst_http_url, os.path.relpath(
309 os.path.join(root, pub_file),
310 start=self.dir_results))
311 output_str += "\n{}".format(link)
312 self.details["links"].append(link)
314 "All artifacts were successfully published: %s\n", output_str)
315 return TestCase.EX_OK
316 except KeyError as ex:
317 self.__logger.error("Please check env var: %s", str(ex))
318 return TestCase.EX_PUBLISH_ARTIFACTS_ERROR
319 except botocore.exceptions.NoCredentialsError:
321 "Please fill ~/.aws/credentials, ~/.boto or set "
322 "AWS_ACCESS_KEY_ID and AWS_SECRET_ACCESS_KEY in env")
323 return TestCase.EX_PUBLISH_ARTIFACTS_ERROR
324 except Exception: # pylint: disable=broad-except
325 self.__logger.exception("Cannot publish the artifacts")
326 return TestCase.EX_PUBLISH_ARTIFACTS_ERROR
329 """Clean the resources.
331 It can be overriden if resources must be deleted after
332 running the test case.