Coverage for src/pytest_xray/case_results.py: 97.3%

1"""Transport-independent, append-only execution observations (Apache-2.0).""" 

2 

3from dataclasses import asdict, dataclass, field 

4from typing import Any, Optional 

5 

6 

7@dataclass 

8class PhaseResult: 

9 phase: str 

10 outcome: str 

11 pytest_outcome: str 

12 duration: float 

13 

14 

15def aggregate(phases: list[PhaseResult]) -> str: 

16 """Conservative summary; never erase an earlier failed or retried phase.""" 

17 outcomes = {phase.outcome for phase in phases} 

18 if outcomes & {'failed', 'error', 'xpass_strict', 'rerun'}: 

19 return 'failed' 

20 if outcomes - {'passed', 'skipped', 'xfail', 'xpass'}: 20 ↛ 21line 20 didn't jump to line 21 because the condition on line 20 was never true

21 return 'incomplete' 

22 if 'skipped' in outcomes or 'xfail' in outcomes: 

23 return 'skipped' 

24 if any(phase.phase == 'call' and phase.outcome in {'passed', 'xpass'} for phase in phases): 

25 return 'passed' 

26 return 'incomplete' 

27 

28 

29@dataclass 

30class AttemptResult: 

31 number: int 

32 worker_id: str 

33 parameter_id: Optional[str] 

34 parameter_values: Optional[dict[str, Any]] = None 

35 phases: list[PhaseResult] = field(default_factory=list) 

36 

37 def as_dict(self) -> dict[str, Any]: 

38 data = asdict(self) 

39 if self.parameter_values is None: 

40 del data['parameter_values'] 

41 data['complete'] = any(phase.phase == 'teardown' for phase in self.phases) 

42 status = aggregate(self.phases) 

43 data['status'] = status if data['complete'] or status == 'failed' else 'incomplete' 

44 return data 

45 

46 

47@dataclass 

48class CaseResult: 

49 nodeid: str 

50 attempts: list[AttemptResult] = field(default_factory=list) 

51 

52 def as_dict(self) -> dict[str, Any]: 

53 statuses = {attempt.as_dict()['status'] for attempt in self.attempts} 

54 # An incomplete observation cannot make a finished failure disappear. 

55 status = next( 

56 (value for value in ('failed', 'incomplete', 'skipped', 'passed') if value in statuses), 'incomplete' 

57 ) 

58 return {'nodeid': self.nodeid, 'status': status, 'attempts': [attempt.as_dict() for attempt in self.attempts]} 

59 

60 

61class CaseResults: 

62 """Keep every phase, grouping attempts by node ID and reporting worker.""" 

63 

64 def __init__(self) -> None: 

65 self.cases: dict[str, CaseResult] = {} 

66 self._active: dict[tuple[str, str], AttemptResult] = {} 

67 

68 def record( 

69 self, 

70 nodeid: str, 

71 worker_id: str, 

72 parameter_id: Optional[str], 

73 phase: PhaseResult, 

74 parameter_values: Optional[dict[str, Any]] = None, 

75 ) -> None: 

76 case = self.cases.setdefault(nodeid, CaseResult(nodeid)) 

77 key = (nodeid, worker_id) 

78 if phase.phase == 'setup' or key not in self._active: 

79 attempt = AttemptResult(len(case.attempts) + 1, worker_id, parameter_id, parameter_values) 

80 case.attempts.append(attempt) 

81 self._active[key] = attempt 

82 attempt = self._active[key] 

83 attempt.phases.append(phase) 

84 if phase.phase == 'teardown': 

85 del self._active[key] 

86 

87 def as_dict(self, exit_status: int) -> dict[str, Any]: 

88 return { 

89 'schema_version': 1, 

90 'pytest_exit_status': exit_status, 

91 'cases': [self.cases[nodeid].as_dict() for nodeid in sorted(self.cases)], 

92 }