Coverage for src/pytest_xray/xray_plugin.py: 94.3%
192 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-16 14:00 +0000
1# Modified in the parameter-details prototype; see NOTICE.
2import datetime as dt
3import json
4import os
5import re
6from pathlib import Path
7from typing import Any, Optional, Protocol, Union
9import pytest
10from _pytest.config import Config, ExitCode
11from _pytest.mark import Mark
12from _pytest.nodes import Item
13from _pytest.reports import TestReport
14from _pytest.terminal import TerminalReporter
16from pytest_xray.case_report import scalar_values
17from pytest_xray.comment_limits import bound_comments
18from pytest_xray.constant import (
19 JIRA_CLOUD,
20 XRAY_ADD_CAPTURES,
21 XRAY_ALLOW_DUPLICATE_IDS,
22 XRAY_EXECUTION_ID,
23 XRAY_MARKER_NAME,
24 XRAY_TEST_PLAN_ID,
25 XRAYPATH,
26)
27from pytest_xray.exceptions import XrayError
28from pytest_xray.file_publisher import FilePublisher
29from pytest_xray.helper import (
30 STATUS_STR_MAPPER_CLOUD,
31 STATUS_STR_MAPPER_JIRA,
32 Status,
33 TestCase,
34 TestExecution,
35)
38class Publisher(Protocol):
39 def publish(self, data: dict[str, Any]) -> str: ...
42class XrayPlugin:
43 """Collects results from pytest and exports to Jira Xray server."""
45 def __init__(self, config: pytest.Config, publisher: Publisher) -> None:
46 self.truncated_comments = 0
47 self.collection_failed = False
48 self.worker_failed = False
49 self.config = config
50 self.parameter_details = config.getoption('--xray-parameter-details', default=False)
51 self.publisher = publisher
52 self.test_execution_id: str = self.config.getoption(XRAY_EXECUTION_ID)
53 self.test_plan_id: str = self.config.getoption(XRAY_TEST_PLAN_ID)
54 self.is_cloud_server: str = self.config.getoption(JIRA_CLOUD)
55 self.allow_duplicate_ids: bool = self.config.getoption(XRAY_ALLOW_DUPLICATE_IDS)
56 logfile = self.config.getoption(XRAYPATH)
57 self.add_captures: bool = self.config.getoption(XRAY_ADD_CAPTURES)
58 self.logfile: Optional[str] = self._get_normalize_logfile(logfile) if logfile else None
59 self.issue_id: Union[str, None] = None # issue id returned by XRAY server
60 self.exception: Union[Exception, None] = None # keeps an exception if raised by XrayPublisher
61 self.test_execution: TestExecution = TestExecution(
62 test_execution_key=self.test_execution_id, test_plan_key=self.test_plan_id
63 )
64 self.status_str_mapper: dict[Status, str] = STATUS_STR_MAPPER_JIRA
65 if self.is_cloud_server:
66 self.status_str_mapper = STATUS_STR_MAPPER_CLOUD
68 @staticmethod
69 def _get_normalize_logfile(logfile: str) -> str:
70 logfile = os.path.expanduser(os.path.expandvars(logfile))
71 logfile = os.path.normpath(os.path.abspath(logfile))
72 return logfile
74 def _get_test_keys(self, item: Item) -> list[str]:
75 """Return JIRA ids associated with test item"""
76 test_keys: list[str] = []
77 marker = self._get_xray_marker(item)
79 if not marker:
80 return test_keys
82 if len(marker.args) == 0: 82 ↛ 83line 82 didn't jump to line 83 because the condition on line 82 was never true
83 raise XrayError(
84 'pytest.mark.xray needs at least one argument, '
85 f'the test {item.nodeid} does not seem to be decorated in proper way.'
86 )
87 if isinstance(marker.args[0], str):
88 test_keys = list(marker.args)
89 elif isinstance(marker.args[0], list): 89 ↛ 92line 89 didn't jump to line 92 because the condition on line 89 was always true
90 test_keys = marker.args[0]
91 else:
92 raise XrayError(f'xray marker can only accept strings or lists but got {type(marker.args[0])}')
93 if not test_keys or any(not isinstance(key, str) or not key.strip() for key in test_keys): 93 ↛ 94line 93 didn't jump to line 94 because the condition on line 93 was never true
94 raise XrayError('xray marker requires nonempty string test keys')
95 # Repeating one key in one marker must not duplicate comments or evidence.
96 return list(dict.fromkeys(test_keys))
98 def _get_defects(self, item: Item) -> list[str]:
99 """Return JIRA ids for defects associated with test item"""
100 marker = self._get_xray_marker(item)
102 if not marker:
103 return []
105 defects = marker.kwargs.get('defects', [])
107 if not isinstance(defects, list) or any(not isinstance(value, str) or not value.strip() for value in defects):
108 raise XrayError('xray marker defects must be a list of nonempty strings (an empty list is allowed)')
109 return defects
111 def _verify_jira_ids_for_items(self, items: list[Item]) -> None:
112 """Verify duplicated jira ids."""
113 jira_ids: list[str] = []
114 duplicated_jira_ids: list[str] = []
116 for item in items:
117 test_keys = self._get_test_keys(item)
118 if not test_keys:
119 continue
121 for test_key in test_keys:
122 if test_key in jira_ids:
123 duplicated_jira_ids.append(test_key)
124 else:
125 jira_ids.append(test_key)
127 if duplicated_jira_ids and not self.allow_duplicate_ids:
128 raise XrayError(f'Duplicated test case ids: {duplicated_jira_ids}')
130 @staticmethod
131 def _get_xray_marker(item: Item) -> Optional[Mark]:
132 return item.get_closest_marker(XRAY_MARKER_NAME)
134 def pytest_sessionstart(self, session):
135 self.test_execution.start_date = dt.datetime.now(tz=dt.timezone.utc)
137 @pytest.hookimpl(tryfirst=True, hookwrapper=True)
138 def pytest_runtest_makereport(self, item, call):
139 outcome = yield
140 report = outcome.get_result()
141 if self.parameter_details and hasattr(item, 'callspec'):
142 # Snapshot only scalar values; never call arbitrary user __repr__.
143 report.xray_parameters = scalar_values(item.callspec.params)
145 if not hasattr(report, 'test_keys'): 145 ↛ 147line 145 didn't jump to line 147 because the condition on line 145 was always true
146 report.test_keys = {}
147 test_keys = self._get_test_keys(item)
148 if test_keys and item.nodeid not in report.test_keys:
149 report.test_keys[item.nodeid] = test_keys
151 if not hasattr(report, 'defects'): 151 ↛ 153line 151 didn't jump to line 153 because the condition on line 151 was always true
152 report.defects = {}
153 defects = self._get_defects(item)
154 if defects and item.nodeid not in report.defects:
155 report.defects[item.nodeid] = defects
157 def pytest_runtest_logreport(self, report: TestReport):
158 status = self._get_status_from_report(report)
159 if status is None:
160 return
162 test_keys = getattr(report, 'test_keys', {}).get(report.nodeid)
163 if test_keys is None:
164 return
166 defects = getattr(report, 'defects', {}).get(report.nodeid)
167 evidences = getattr(report, 'evidences', [])
169 comment = report.longreprtext
170 if self.add_captures:
171 if comment != '': 171 ↛ 172line 171 didn't jump to line 172 because the condition on line 171 was never true
172 comment += '\n'
173 if report.capstdout: 173 ↛ 175line 173 didn't jump to line 175 because the condition on line 173 was always true
174 comment += f'{"-" * 29} Captured stdout call {"-" * 29}\n{report.capstdout}'
175 if report.capstderr: 175 ↛ 177line 175 didn't jump to line 177 because the condition on line 175 was always true
176 comment += f'{"-" * 29} Captured stderr call {"-" * 29}\n{report.capstderr}'
177 if report.caplog: 177 ↛ 182line 177 didn't jump to line 182 because the condition on line 177 was always true
178 # Remove the ncurses escape chars used to color log level
179 logt = re.subn('\x1b.*?m', '', report.caplog)
180 comment += f'{"-" * 30} Captured log call {"-" * 31}\n{logt[0]}'
182 if self.parameter_details and hasattr(report, 'xray_parameters'):
183 details = json.dumps(
184 {
185 'nodeid': report.nodeid,
186 'parameters': report.xray_parameters,
187 'phase': report.when,
188 'pytest_outcome': report.outcome,
189 'xray_status': self.status_str_mapper[status],
190 },
191 ensure_ascii=True,
192 sort_keys=True,
193 )
194 comment = f'pytest parameter case: {details}\n{comment}'
196 for test_key in test_keys:
197 new_test_case = TestCase(
198 test_key=test_key,
199 status=status,
200 comment=comment,
201 status_str_mapper=self.status_str_mapper,
202 evidences=evidences,
203 defects=defects,
204 )
205 try:
206 test_case = self.test_execution.find_test_case(test_key)
207 except KeyError:
208 self.test_execution.append(new_test_case)
209 else:
210 test_case.merge(new_test_case)
212 def _get_status_from_report(self, report) -> Optional[Status]:
213 if report.outcome == 'rerun':
214 return Status.FAIL
215 if report.failed:
216 if report.when != 'call':
217 return Status.FAIL
218 if hasattr(report, 'wasxfail'): 218 ↛ 219line 218 didn't jump to line 219 because the condition on line 218 was never true
219 return Status.PASS
220 return Status.FAIL
221 if report.skipped:
222 if hasattr(report, 'wasxfail'):
223 return Status.FAIL
224 return Status.ABORTED
225 if report.passed and report.when == 'call':
226 return Status.PASS
228 return None
230 def pytest_collection_modifyitems(self, config: Config, items: list[Item]) -> None:
231 # Validate before any fixture runs; do not include marker values in errors.
232 for item in items:
233 try:
234 self._get_defects(item)
235 except XrayError as exc:
236 # A collect report reaches the xdist controller; UsageError here
237 # can let workers announce runnable items before exiting.
238 items.clear()
239 config.hook.pytest_collectreport(
240 report=pytest.CollectReport(
241 nodeid='xray defects configuration',
242 outcome='failed',
243 longrepr=exc.message,
244 result=[],
245 )
246 )
247 return
248 self._verify_jira_ids_for_items(items)
250 def pytest_collectreport(self, report):
251 if report.failed:
252 self.collection_failed = True
254 @pytest.hookimpl(optionalhook=True)
255 def pytest_testnodedown(self, node, error):
256 if error:
257 self.worker_failed = True
259 def pytest_sessionfinish(self, session: pytest.Session) -> None:
260 if hasattr(self.config, 'workerinput'): # skipping on xdist
261 return
262 self.test_execution.finish_date = dt.datetime.now(tz=dt.timezone.utc)
263 results = self.test_execution.as_dict()
264 session.config.pluginmanager.hook.pytest_xray_results(results=results, session=session)
265 try:
266 results, self.truncated_comments = bound_comments(
267 results, self.config.getoption('--xray-comment-limit'), self.config.getoption('--xray-full-report')
268 )
269 if not isinstance(self.publisher, FilePublisher) and (
270 self.worker_failed
271 or self.collection_failed
272 or session.shouldfail
273 or session.shouldstop
274 or session.exitstatus not in (ExitCode.OK, ExitCode.TESTS_FAILED)
275 or not results['tests']
276 ):
277 raise XrayError('Upload withheld: session incomplete, worker crashed, or no marked results')
278 self.issue_id = self.publisher.publish(results)
279 except XrayError as exc:
280 self.exception = exc
281 if session.exitstatus == ExitCode.OK:
282 session.exitstatus = ExitCode.INTERNAL_ERROR
284 def pytest_terminal_summary(self, terminalreporter: TerminalReporter, exitstatus: ExitCode, config: Config) -> None:
285 if self.truncated_comments:
286 terminalreporter.write_line(
287 f'XRAY: truncated {self.truncated_comments} comment(s) to '
288 f'{self.config.getoption("--xray-comment-limit")} UTF-8 bytes; complete payload saved to '
289 f'{self.config.getoption("--xray-full-report")}',
290 yellow=True,
291 )
292 if self.exception:
293 terminalreporter.ensure_newline()
294 terminalreporter.section('Jira XRAY', sep='-', red=True, bold=True)
295 terminalreporter.write_line('Could not publish results to Jira XRAY!')
296 if self.exception.message: # type: ignore[attr-defined] 296 ↛ exitline 296 didn't return from function 'pytest_terminal_summary' because the condition on line 296 was always true
297 terminalreporter.write_line(self.exception.message) # type: ignore[attr-defined]
298 else:
299 if self.issue_id and self.logfile:
300 terminalreporter.write_sep(
301 '-', f'Generated XRAY execution report file: {Path(self.logfile).absolute()}'
302 )
303 elif self.issue_id:
304 terminalreporter.write_sep('-', f'Uploaded results to JIRA XRAY. Test Execution Id: {self.issue_id}')