Coverage for src/pytest_xray/helper.py: 87.5%
166 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-16 14:00 +0000
1import datetime as dt
2import enum
3import os
4import re
5from os import environ
6from typing import Any, Optional, Union
8from pytest_xray import constant
9from pytest_xray.constant import (
10 DATETIME_FORMAT,
11 ENV_XRAY_API_BASE_URL,
12 ENV_XRAY_API_KEY,
13 ENV_XRAY_API_PASSWORD,
14 ENV_XRAY_API_USER,
15 ENV_XRAY_API_VERIFY_SSL,
16 ENV_XRAY_CLIENT_ID,
17 ENV_XRAY_CLIENT_SECRET,
18)
19from pytest_xray.exceptions import XrayError
21DEFAULT_SUMMARY_DESCRIPTION: str = 'Execution of automated tests'
24class Status(str, enum.Enum):
25 TODO = 'TODO'
26 EXECUTING = 'EXECUTING'
27 PENDING = 'PENDING'
28 PASS = 'PASS'
29 FAIL = 'FAIL'
30 ABORTED = 'ABORTED'
31 BLOCKED = 'BLOCKED'
34# This is the hierarchy of the Status, from bottom to top.
35# When merging two statuses, the highest will be picked.
36# For example, a PASS and a FAIL will result in a FAIL,
37# A TODO and an ABORTED in an ABORTED, A TODO and a PASS in a TODO.
38STATUS_HIERARCHY: list[Status] = [
39 Status.PASS,
40 Status.TODO,
41 Status.EXECUTING,
42 Status.PENDING,
43 Status.FAIL,
44 Status.ABORTED,
45 Status.BLOCKED,
46]
48# Maps the Status from the internal Status enum to the string representations
49# requested by either the Cloud Jira, or the on-site Jira
50STATUS_STR_MAPPER_CLOUD: dict[Status, str] = {
51 Status.TODO: 'TODO',
52 Status.EXECUTING: 'EXECUTING',
53 Status.PENDING: 'PENDING',
54 Status.PASS: 'PASSED',
55 Status.FAIL: 'FAILED',
56 Status.ABORTED: 'ABORTED',
57 Status.BLOCKED: 'BLOCKED',
58}
60# On-site jira uses the enum strings directly
61STATUS_STR_MAPPER_JIRA: dict[Status, str] = {x: x.value for x in Status}
64class TestCase:
65 __test__ = False
67 def __init__(
68 self,
69 test_key: str,
70 status: Status,
71 comment: Optional[str] = None,
72 status_str_mapper: Optional[dict[Status, str]] = None,
73 evidences: Optional[list[dict[str, str]]] = None,
74 defects: Optional[list[str]] = None,
75 ) -> None:
76 self.test_key = test_key
77 self.status = status
78 self.comment = comment or ''
79 self.status_str_mapper = status_str_mapper or STATUS_STR_MAPPER_JIRA
80 # One report can map to multiple Test keys; each aggregate owns its lists.
81 self.evidences = list(evidences or [])
82 self.defects = list(defects or [])
84 def merge(self, other: 'TestCase') -> None:
85 """
86 Merges this test case with other, in order to obtain
87 a combined result. Comments will be just appended one after the other.
88 status will be merged according to a priority list.
89 Merge is only possible if the two tests have the same test_key
90 """
92 if self.test_key != other.test_key:
93 raise ValueError(f'Cannot merge test with different test keys: {self.test_key} {other.test_key}')
95 if self.comment == '':
96 if other.comment != '':
97 self.comment = other.comment
98 else:
99 if other.comment != '':
100 self.comment += '\n' + '-' * 80 + '\n'
101 self.comment += other.comment
103 self.status = _merge_status(self.status, other.status)
105 for defect in other.defects:
106 if defect not in self.defects:
107 self.defects.append(defect)
109 for evidence in other.evidences:
110 self.evidences.append(evidence)
112 def as_dict(self) -> dict[str, Any]:
113 data: dict[str, Any] = dict(
114 testKey=self.test_key,
115 status=self.status_str_mapper[self.status],
116 )
117 if self.comment != '':
118 data['comment'] = '{noformat:borderWidth=0px|bgColor=transparent}' + self.comment + '{noformat}'
119 if self.evidences:
120 data['evidences'] = self.evidences
121 if self.defects:
122 data['defects'] = self.defects
123 return data
126class TestExecution:
127 __test__ = False
129 def __init__(
130 self,
131 test_execution_key: Optional[str] = None,
132 test_plan_key: Optional[str] = None,
133 user: Optional[str] = None,
134 revision: Optional[str] = None,
135 tests: Optional[list[TestCase]] = None,
136 test_environments: Optional[list[str]] = None,
137 fix_version: Optional[str] = None,
138 summary: Optional[str] = None,
139 description: Optional[str] = None,
140 ):
141 self.test_execution_key = test_execution_key
142 self.test_plan_key: str = test_plan_key or ''
143 self.user = user or ''
144 self.revision = revision or _from_environ_or_none(constant.ENV_TEST_EXECUTION_REVISION)
145 self.start_date = dt.datetime.now(tz=dt.timezone.utc)
146 self.finish_date = dt.datetime.now(tz=dt.timezone.utc)
147 self.tests = tests or []
148 self.test_environments = test_environments or _from_environ(
149 constant.ENV_TEST_EXECUTION_TEST_ENVIRONMENTS, constant.ENV_MULTI_VALUE_SPLIT_PATTERN
150 )
151 self.fix_version = fix_version or _first_from_environ(constant.ENV_TEST_EXECUTION_FIX_VERSION)
152 self.summary = self._get_summery(summary)
153 self.description = description or _from_environ_or_none(constant.ENV_TEST_EXECUTION_DESC)
155 def _get_summery(self, summary: Union[str, None]) -> Union[str, None]:
156 summary = summary or _from_environ_or_none(constant.ENV_TEST_EXECUTION_SUMMARY)
157 if summary is not None:
158 return summary
159 # do not set `summary` if uploading to existing test execution
160 if self.test_execution_key is None:
161 return DEFAULT_SUMMARY_DESCRIPTION
162 return None
164 def append(self, test: Union[dict, TestCase]) -> None:
165 if not isinstance(test, TestCase): 165 ↛ 166line 165 didn't jump to line 166 because the condition on line 165 was never true
166 test = TestCase(**test)
167 self.tests.append(test)
169 def find_test_case(self, test_key: str) -> TestCase:
170 """
171 Searches a stored test case by identifier.
172 If not found, raises KeyError
173 """
174 # Linear search, but who cares really of performance here?
176 for test in self.tests:
177 if test.test_key == test_key:
178 return test
180 raise KeyError(test_key)
182 def as_dict(self) -> dict[str, Any]:
183 """Return test execution result as dictionary."""
184 tests = [test.as_dict() for test in self.tests]
185 info: dict[str, Any] = dict(
186 startDate=self.start_date.strftime(DATETIME_FORMAT),
187 finishDate=self.finish_date.strftime(DATETIME_FORMAT), # type: ignore
188 )
190 if self.fix_version:
191 info['version'] = self.fix_version
193 if self.test_environments and len(self.test_environments) > 0:
194 info['testEnvironments'] = self.test_environments
196 if self.summary is not None:
197 info['summary'] = self.summary
199 if self.description is not None:
200 info['description'] = self.description
202 if self.revision is not None: 202 ↛ 203line 202 didn't jump to line 203 because the condition on line 202 was never true
203 info['revision'] = self.revision
205 data: dict[str, Any] = dict(info=info, tests=tests)
206 if self.test_plan_key:
207 info['testPlanKey'] = self.test_plan_key
208 if self.test_execution_key is not None:
209 data['testExecutionKey'] = self.test_execution_key
210 return data
213def get_base_options() -> dict[str, Any]:
214 """Return authentication configuration from environment variables."""
215 options = {}
216 try:
217 base_url = environ[ENV_XRAY_API_BASE_URL]
218 except KeyError:
219 raise XrayError(f'pytest-jira-xray plugin requires environment variable: {ENV_XRAY_API_BASE_URL}') from None
221 verify = os.environ.get(ENV_XRAY_API_VERIFY_SSL, 'True')
223 if verify.upper() == 'TRUE': 223 ↛ 225line 223 didn't jump to line 225 because the condition on line 223 was always true
224 verify = True # type: ignore
225 elif verify.upper() == 'FALSE':
226 verify = False # type: ignore
227 else:
228 if not os.path.exists(verify):
229 raise XrayError(f'Cannot find certificate file "{verify}"')
231 options['VERIFY'] = verify
232 options['BASE_URL'] = base_url
233 return options
236def get_basic_auth() -> dict[str, Any]:
237 """Return basic authentication setup with username and password."""
238 options = get_base_options()
239 try:
240 user = environ[ENV_XRAY_API_USER]
241 password = environ[ENV_XRAY_API_PASSWORD]
242 except KeyError:
243 raise XrayError(
244 f'Basic authentication requires environment variables: {ENV_XRAY_API_USER}, {ENV_XRAY_API_PASSWORD}'
245 ) from None
247 options['USER'] = user
248 options['PASSWORD'] = password
249 return options
252def get_bearer_auth() -> dict[str, Any]:
253 """Return bearer authentication setup with Client ID and a Client Secret."""
254 options = get_base_options()
255 try:
256 client_id = environ[ENV_XRAY_CLIENT_ID]
257 client_secret = environ[ENV_XRAY_CLIENT_SECRET]
258 except KeyError:
259 raise XrayError(
260 f'Bearer authentication requires environment variables: {ENV_XRAY_CLIENT_ID}, {ENV_XRAY_CLIENT_SECRET}'
261 ) from None
263 options['CLIENT_ID'] = client_id
264 options['CLIENT_SECRET'] = client_secret
265 return options
268def get_api_key_auth() -> dict[str, Any]:
269 """Return personal access token authentication."""
270 options = get_base_options()
271 try:
272 api_key = environ[ENV_XRAY_API_KEY]
273 except KeyError:
274 raise XrayError(f'API Key authentication requires environment variable: {ENV_XRAY_API_KEY}') from None
276 options['API_KEY'] = api_key
277 return options
280def _from_environ_or_none(name: str) -> Optional[str]:
281 if name in environ: 281 ↛ 282line 281 didn't jump to line 282 because the condition on line 281 was never true
282 val = environ[name].strip()
283 if len(val) == 0:
284 return None
285 else:
286 return None
287 return val
290def _first_from_environ(name: str, separator: Optional[str] = None) -> Optional[str]:
291 return next(iter(_from_environ(name, separator)), None)
294def _from_environ(name: str, separator: Optional[str] = None) -> list[str]:
295 if name not in environ:
296 return []
298 param = environ[name]
300 if separator:
301 source = re.split(separator, param)
302 else:
303 source = [param]
305 # Return stripped non empty values
306 return list(filter(lambda x: len(x) > 0, map(lambda x: x.strip(), source)))
309def _merge_status(status_1: Status, status_2: Status) -> Status:
310 """Merges the status of two tests."""
312 return STATUS_HIERARCHY[max(STATUS_HIERARCHY.index(status_1), STATUS_HIERARCHY.index(status_2))]