Coverage for src/pytest_xray/case_report.py: 98.9%

1"""Opt-in local case report; has no Xray publisher or network transport.""" 

2 

3import json 

4import math 

5import os 

6import tempfile 

7import warnings 

8from pathlib import Path 

9 

10import pytest 

11 

12from pytest_xray.case_results import CaseResults, PhaseResult 

13 

14 

15def scalar_values(params): 

16 """Do not traverse containers or invoke user serialization/repr methods.""" 

17 return { 

18 name: value 

19 if type(value) in (str, int, bool, type(None)) or (type(value) is float and math.isfinite(value)) 

20 else '<unsupported value: use a descriptive pytest id>' 

21 for name, value in params.items() 

22 } 

23 

24 

25def phase_outcome(report): 

26 if report.outcome == 'rerun': 

27 return 'rerun' 

28 if hasattr(report, 'wasxfail'): 

29 if report.skipped: 

30 return 'xfail' 

31 if report.passed: 31 ↛ 33line 31 didn't jump to line 33 because the condition on line 31 was always true

32 return 'xpass' 

33 if report.failed: 

34 if report.when != 'call': 

35 return 'error' 

36 if report.longreprtext.startswith('[XPASS(strict)]'): 

37 return 'xpass_strict' 

38 return 'failed' 

39 return report.outcome 

40 

41 

42class CaseReportPlugin: 

43 def __init__(self, config, path, include_values=False, redact_identities=False): 

44 self.config = config 

45 self.path = Path(path).expanduser().resolve() 

46 self.include_values = include_values 

47 self.redact_identities = redact_identities 

48 self._identities = {} 

49 self.results = CaseResults() 

50 

51 @pytest.hookimpl(hookwrapper=True, tryfirst=True) 

52 def pytest_runtest_makereport(self, item, call): 

53 outcome = yield 

54 report = outcome.get_result() 

55 callspec = getattr(item, 'callspec', None) 

56 report.xray_case_metadata = { 

57 'parameter_id': callspec.id if callspec else None, 

58 'outcome': phase_outcome(report), 

59 } 

60 if self.include_values and callspec: 

61 report.xray_case_metadata['parameter_values'] = scalar_values(callspec.params) 

62 

63 def pytest_runtest_logreport(self, report): 

64 if hasattr(self.config, 'workerinput'): 

65 return 

66 metadata = getattr(report, 'xray_case_metadata', {}) 

67 nodeid = report.nodeid 

68 parameter_id = metadata.get('parameter_id') 

69 if self.redact_identities: 

70 nodeid = self._identities.setdefault(nodeid, f'case-{len(self._identities) + 1}') 

71 parameter_id = f'{nodeid}-parameters' if parameter_id is not None else None 

72 self.results.record( 

73 nodeid=nodeid, 

74 worker_id=getattr(report, 'worker_id', 'local'), 

75 parameter_id=parameter_id, 

76 parameter_values=metadata.get('parameter_values'), 

77 phase=PhaseResult( 

78 report.when, metadata.get('outcome', phase_outcome(report)), report.outcome, report.duration 

79 ), 

80 ) 

81 

82 @pytest.hookimpl(trylast=True) 

83 def pytest_sessionfinish(self, session, exitstatus): 

84 if hasattr(self.config, 'workerinput'): 

85 return 

86 temporary = None 

87 try: 

88 content = json.dumps(self.results.as_dict(int(session.exitstatus)), indent=2, ensure_ascii=True, allow_nan=False) 

89 with tempfile.NamedTemporaryFile(mode='w', encoding='utf-8', dir=self.path.parent, delete=False) as stream: 

90 temporary = Path(stream.name) 

91 stream.write(content + '\n') 

92 os.replace(temporary, self.path) 

93 except (OSError, ValueError) as exc: 

94 session.exitstatus = pytest.ExitCode.INTERNAL_ERROR 

95 warnings.warn(pytest.PytestWarning(f'Could not write offline case report {self.path}: {exc}'), stacklevel=2) 

96 finally: 

97 if temporary is not None: 

98 temporary.unlink(missing_ok=True)