Coverage for pytest_recap/models.py: 48%
169 statements
« prev ^ index » next coverage.py v7.8.0, created at 2025-06-14 23:19 -0600
1import logging
2from collections import Counter
3from dataclasses import asdict, dataclass, field
4from datetime import datetime, timedelta, timezone
5from enum import Enum
6from typing import Any, Dict, Iterable, List, Optional
8logger = logging.getLogger(__name__)
11class TestOutcome(Enum):
12 """Test outcome states.
14 Enum values:
15 PASSED: Test passed
16 FAILED: Test failed
17 SKIPPED: Test skipped
18 XFAILED: Expected failure
19 XPASSED: Unexpected pass
20 RERUN: Test was rerun
21 ERROR: Test errored
22 """
24 __test__ = False # Tell Pytest this is NOT a test class
26 PASSED = "PASSED" # Internal representation in UPPERCASE
27 FAILED = "FAILED"
28 SKIPPED = "SKIPPED"
29 XFAILED = "XFAILED"
30 XPASSED = "XPASSED"
31 RERUN = "RERUN"
32 ERROR = "ERROR"
34 @classmethod
35 def from_str(cls, outcome: Optional[str]) -> "TestOutcome":
36 """Convert string to TestOutcome, always uppercase internally.
38 Args:
39 outcome (Optional[str]): Outcome string.
41 Returns:
42 TestOutcome: Corresponding enum value.
44 """
45 if not outcome:
46 return cls.SKIPPED # Return a default enum value instead of None
47 try:
48 return cls[outcome.upper()]
49 except KeyError:
50 raise ValueError(f"Invalid test outcome: {outcome}")
52 def to_str(self) -> str:
53 """Convert TestOutcome to string, always lowercase externally.
55 Returns:
56 str: Lowercase outcome string.
58 """
59 return self.value.lower()
61 @classmethod
62 def to_list(cls) -> List[str]:
63 """Convert entire TestOutcome enum to a list of possible string values.
65 Returns:
66 List[str]: List of lowercase outcome strings.
68 """
69 return [outcome.value.lower() for outcome in cls]
71 def is_failed(self) -> bool:
72 """Check if the outcome represents a failure.
74 Returns:
75 bool: True if outcome is failure or error, else False.
77 """
78 return self in (self.FAILED, self.ERROR)
81@dataclass
82class TestResult:
83 """Represents a single test result for an individual test run.
85 Attributes:
86 nodeid (str): Unique identifier for the test node.
87 outcome (TestOutcome): Result outcome.
88 start_time (Optional[datetime]): Start time of the test.
89 stop_time (Optional[datetime]): Stop time of the test.
90 duration (Optional[float]): Duration in seconds.
91 caplog (str): Captured log output.
92 capstderr (str): Captured stderr output.
93 capstdout (str): Captured stdout output.
94 longreprtext (str): Long representation of failure, if any.
96 """
98 __test__ = False # Tell Pytest this is NOT a test class
100 nodeid: str
101 outcome: TestOutcome
102 start_time: Optional[datetime] = None
103 stop_time: Optional[datetime] = None
104 duration: Optional[float] = None
105 caplog: str = ""
106 capstderr: str = ""
107 capstdout: str = ""
108 longreprtext: str = ""
109 has_warning: bool = False
110 has_error: bool = False
112 def __post_init__(self):
113 """Validate and process initialization data.
115 Raises:
116 ValueError: If neither stop_time nor duration is provided.
118 """
119 # Only compute stop_time if both start_time and duration are present and stop_time is missing
120 if self.stop_time is None and self.start_time is not None and self.duration is not None:
121 self.stop_time = self.start_time + timedelta(seconds=self.duration)
122 # Only compute duration if both start_time and stop_time are present and duration is missing
123 elif self.duration is None and self.start_time is not None and self.stop_time is not None:
124 self.duration = (self.stop_time - self.start_time).total_seconds()
126 def to_dict(self) -> Dict:
127 """Convert test result to a dictionary for JSON serialization.
129 Returns:
130 dict: Dictionary representation of the test result.
132 """
133 # Handle both string and enum outcomes for backward compatibility
134 if not hasattr(self.outcome, "to_str"):
135 logger.warning(
136 "Non-enum (probably string outcome detected where TestOutcome enum expected. "
137 f"nodeid={self.nodeid}, outcome={self.outcome}, type={type(self.outcome)}. "
138 "For proper session context and query filtering, use TestOutcome enum: "
139 "outcome=TestOutcome.FAILED instead of outcome='failed'. "
140 "String outcomes are deprecated and will be removed in a future version."
141 )
142 outcome_str = str(self.outcome).lower()
143 else:
144 outcome_str = self.outcome.to_str()
146 return {
147 "nodeid": self.nodeid,
148 "outcome": outcome_str,
149 "start_time": self.start_time.isoformat() if self.start_time else None,
150 "stop_time": self.stop_time.isoformat() if self.stop_time else None,
151 "duration": self.duration,
152 "caplog": self.caplog,
153 "capstderr": self.capstderr,
154 "capstdout": self.capstdout,
155 "longreprtext": self.longreprtext,
156 }
158 @classmethod
159 def from_dict(cls, data: Dict) -> "TestResult":
160 """Create a TestResult from a dictionary."""
161 start_time = data.get("start_time")
162 if isinstance(start_time, str):
163 start_time = datetime.fromisoformat(start_time)
165 stop_time = data.get("stop_time")
166 if isinstance(stop_time, str):
167 stop_time = datetime.fromisoformat(stop_time)
169 return cls(
170 nodeid=data["nodeid"],
171 outcome=TestOutcome.from_str(data["outcome"]),
172 start_time=start_time,
173 stop_time=stop_time,
174 duration=data.get("duration"),
175 caplog=data.get("caplog", ""),
176 capstderr=data.get("capstderr", ""),
177 capstdout=data.get("capstdout", ""),
178 longreprtext=data.get("longreprtext", ""),
179 )
182@dataclass
183class RerunTestGroup:
184 """Groups test results for tests that were rerun, chronologically ordered with final result last.
186 Attributes:
187 nodeid (str): Test node ID.
188 tests (List[TestResult]): List of TestResult objects for each rerun.
189 """
191 __test__ = False
192 nodeid: str
193 tests: List["TestResult"] = field(default_factory=list)
195 @property
196 def final_outcome(self) -> Optional[str]:
197 """Compute the final outcome for the group based on test results.
199 Returns:
200 Optional[str]: The computed final outcome (e.g., "passed", "failed", "error"), or None if no tests.
201 """
202 if not self.tests:
203 return None
205 # Make sure only one test has an outcome that is not "rerun"
206 non_rerun_count = sum(test.outcome.value.lower() != "rerun" for test in self.tests)
207 assert non_rerun_count == 1, f"Expected at most one non-rerun test, got {non_rerun_count} instead"
209 # The final outcome is the outcome of the only test that is not a rerun
210 for test in self.tests:
211 if test.outcome.value.lower() != "rerun":
212 return test.outcome.value.lower()
213 return None
215 def add_test(self, result: "TestResult"):
216 """Add a test result and maintain chronological order."""
217 self.tests.append(result)
218 self.tests.sort(key=lambda t: t.start_time)
220 def to_dict(self) -> Dict:
221 d = {"nodeid": self.nodeid, "tests": [t.to_dict() for t in self.tests]}
222 return d
224 @classmethod
225 def from_dict(cls, data: Dict) -> "RerunTestGroup":
226 if not isinstance(data, dict):
227 raise ValueError(f"Invalid data for RerunTestGroup. Expected dict, got {type(data)}")
228 group = cls(
229 nodeid=data["nodeid"],
230 tests=[TestResult.from_dict(test_dict) for test_dict in data.get("tests", [])],
231 )
232 return group
235class TestSessionStats:
236 """Aggregates session-level statistics, including test outcomes and other events (e.g., warnings).
238 Attributes:
239 passed (int): Number of passed tests
240 failed (int): Number of failed tests
241 skipped (int): Number of skipped tests
242 xfailed (int): Number of unexpectedly failed tests
243 xpassed (int): Number of unexpectedly passed tests
244 error (int): Number of error tests
245 rerun (int): Number of rerun tests
246 warnings (int): Number of warnings encountered in this session
247 """
249 __test__ = False # Tell Pytest this is NOT a test class
251 def __init__(self, test_results: Iterable[Any], warnings_count: int = 0):
252 """
253 Args:
254 test_results (Iterable[TestResult]): List of TestResult objects.
255 warning_count (int): Number of warnings in the session.
256 """
257 # Aggregate test outcomes (e.g., passed, failed, etc.)
258 self.counter = Counter(
259 str(getattr(test_result, "outcome", test_result)).lower() for test_result in test_results
260 )
261 self.total = len(test_results)
262 # Add warnings as a separate count (always present, even if zero)
263 self.counter["warnings"] = warnings_count
265 def count(self, key: str) -> int:
266 """Return the count for a given outcome or event (case-insensitive string)."""
267 return self.counter.get(key.lower(), 0)
269 def as_dict(self) -> Dict[str, int]:
270 """Return all session-level event counts as a dict, with 'testoutcome.' prefix removed from keys. Always include 'warnings' if present in counter, even if zero."""
271 d = {
272 (k[len("testoutcome.") :] if k.startswith("testoutcome.") else k): v
273 for k, v in self.counter.items()
274 if v > 0
275 }
276 # Always include 'warnings' if present in counter, even if zero
277 if "warnings" in self.counter and "warnings" not in d:
278 d["warnings"] = 0
279 return d
281 def __str__(self) -> str:
282 """Return a string representation of the TestSessionStats object."""
283 return f"TestSessionStats(total={self.total}, {dict(self.counter)})"
286@dataclass
287class TestSession:
288 """Represents a test session recap with session-level metadata, results.
290 Attributes:
291 session_id (str): Unique session identifier.
292 session_start_time (datetime): Start time of the session.
293 session_stop_time (datetime): Stop time of the session.
294 system_under_test (dict): Information about the system under test (user-extensible).
295 session_tags (Dict[str, str]): Arbitrary tags for the session.
296 testing_system (Dict[str, Any]): Metadata about the testing system.
297 test_results (List[TestResult]): List of test results in the session.
298 rerun_test_groups (List[RerunTestGroup]): Groups of rerun tests.
299 session_stats (TestSessionStats): Test session statistics.
301 """
303 __test__ = False # Tell Pytest this is NOT a test class
305 def __init__(
306 self,
307 session_id: str,
308 session_start_time: datetime,
309 session_stop_time: datetime = None,
310 system_under_test: dict = None,
311 session_tags: dict = None,
312 testing_system: dict = None,
313 test_results: list = None,
314 rerun_test_groups: list = None,
315 warnings: Optional[List["RecapEvent"]] = None,
316 errors: Optional[List["RecapEvent"]] = None,
317 session_stats: TestSessionStats = None,
318 ):
319 self.session_id = session_id
320 self.session_start_time = session_start_time
321 self.session_stop_time = session_stop_time or datetime.now(timezone.utc)
322 self.system_under_test = system_under_test or {}
323 self.session_tags = session_tags or {}
324 self.testing_system = testing_system or {}
325 self.test_results = test_results or []
326 self.rerun_test_groups = rerun_test_groups or []
327 self.warnings = warnings or []
328 self.errors = errors or []
329 self.session_stats = session_stats or TestSessionStats(self.test_results, len(self.warnings))
331 def to_dict(self) -> Dict:
332 """Convert TestSession to a dictionary for JSON serialization.
334 Returns:
335 dict: Dictionary representation of the test session.
336 """
337 return {
338 "session_id": self.session_id,
339 "session_tags": self.session_tags or {},
340 "session_start_time": self.session_start_time.isoformat(),
341 "session_stop_time": self.session_stop_time.isoformat(),
342 "system_under_test": self.system_under_test or {},
343 "testing_system": self.testing_system or {},
344 "test_results": [test.to_dict() for test in self.test_results],
345 "rerun_test_groups": [
346 {"nodeid": group.nodeid, "tests": [t.to_dict() for t in group.tests]}
347 for group in self.rerun_test_groups
348 ],
349 "warnings": [w.to_dict() for w in self.warnings],
350 "errors": [e.to_dict() for e in self.errors],
351 "session_stats": self.session_stats.as_dict() if self.session_stats else {},
352 }
354 @classmethod
355 def from_dict(cls, d):
356 """Create a TestSession from a dictionary. Ensures warnings count is passed to TestSessionStats."""
357 if not isinstance(d, dict):
358 raise ValueError(f"Invalid data for TestSession. Expected dict, got {type(d)}")
359 session_start_time = d.get("session_start_time")
360 if isinstance(session_start_time, str):
361 session_start_time = datetime.fromisoformat(session_start_time)
362 session_stop_time = d.get("session_stop_time")
363 if isinstance(session_stop_time, str):
364 session_stop_time = datetime.fromisoformat(session_stop_time)
365 test_results = [TestResult.from_dict(test_result) for test_result in d.get("test_results", [])]
366 warnings = [RecapEvent(**w) if not isinstance(w, RecapEvent) else w for w in d.get("warnings", [])]
367 session_stats = TestSessionStats(test_results, warnings_count=len(warnings))
368 return cls(
369 session_id=d.get("session_id"),
370 session_start_time=session_start_time,
371 session_stop_time=session_stop_time,
372 system_under_test=d.get("system_under_test", {}),
373 session_tags=d.get("session_tags", {}),
374 testing_system=d.get("testing_system", {}),
375 test_results=test_results,
376 rerun_test_groups=[RerunTestGroup.from_dict(g) for g in d.get("rerun_test_groups", [])],
377 warnings=warnings,
378 errors=d.get("errors", []),
379 session_stats=session_stats,
380 )
382 def add_test_result(self, result: TestResult) -> None:
383 """Add a test result to this session.
385 Args:
386 result (TestResult): TestResult to add.
388 Raises:
389 ValueError: If result is not a TestResult instance.
390 """
391 if not isinstance(result, TestResult):
392 raise ValueError(
393 f"Invalid test result {result}; must be a TestResult object, nistead was type {type(result)}"
394 )
396 self.test_results.append(result)
398 def add_rerun_group(self, group: RerunTestGroup) -> None:
399 """Add a rerun test group to this session.
401 Args:
402 group (RerunTestGroup): RerunTestGroup to add.
404 Raises:
405 ValueError: If group is not a RerunTestGroup instance.
406 """
407 if not isinstance(group, RerunTestGroup):
408 raise ValueError(
409 f"Invalid rerun group {group}; must be a RerunTestGroup object, instead was type {type(group)}"
410 )
412 self.rerun_test_groups.append(group)
415class RecapEventType(str, Enum):
416 ERROR = "error"
417 WARNING = "warning"
420@dataclass
421class RecapEvent:
422 event_type: RecapEventType = RecapEventType.WARNING
423 nodeid: Optional[str] = None
424 when: Optional[str] = None
425 outcome: Optional[str] = None
426 message: Optional[str] = None
427 category: Optional[str] = None
428 filename: Optional[str] = None
429 lineno: Optional[int] = None
430 longrepr: Optional[Any] = None
431 sections: List[Any] = field(default_factory=list)
432 keywords: List[str] = field(default_factory=list)
433 location: Optional[Any] = None
435 def to_dict(self) -> dict:
436 """Convert the RecapEvent to a dictionary."""
437 return asdict(self)
439 def is_warning(self) -> bool:
440 """Return True if this event is classified as a warning."""
441 return self.event_type == RecapEventType.WARNING
443 def is_error(self) -> bool:
444 """Return True if this event is classified as an error."""
445 return self.event_type == RecapEventType.ERROR