Coverage for src/secchi/diagnostics.py: 91%

1"""Readable, structured diagnostics shared by CLI and dashboard surfaces.""" 

2 

3from __future__ import annotations 

4 

5from collections.abc import Callable 

6from dataclasses import dataclass 

7from datetime import UTC, datetime 

8from enum import StrEnum 

9from pathlib import Path 

10from threading import Lock 

11 

12 

13class DiagnosticStatus(StrEnum): 

14 SUCCESS = "SUCCESS" 

15 WARN = "WARN" 

16 FAILURE = "FAILURE" 

17 

18 

19@dataclass(frozen=True) 

20class DiagnosticEvent: 

21 """One user-facing process or HTTP diagnostic event.""" 

22 

23 timestamp: datetime 

24 status: DiagnosticStatus 

25 source: str 

26 message: str 

27 url: str = "" 

28 status_code: int | None = None 

29 

30 def format(self) -> str: 

31 time = self.timestamp.astimezone(UTC).strftime("%H:%M:%S") 

32 suffix = f" -> {self.status_code}" if self.status_code is not None else "" 

33 target = f" {self.url}" if self.url else "" 

34 return f"{time} {self.status.value:<7} [{self.source}] {self.message}{target}{suffix}" 

35 

36 

37class DiagnosticLog: 

38 """Bounded session log with optional human-readable file output.""" 

39 

40 def __init__( 

41 self, 

42 *, 

43 path: Path | None = None, 

44 max_events: int = 1000, 

45 clock: Callable[[], datetime] | None = None, 

46 ) -> None: 

47 self.path = path 

48 self.max_events = max_events 

49 self.clock = clock or (lambda: datetime.now(UTC)) 

50 self._events: list[DiagnosticEvent] = [] 

51 self._lock = Lock() 

52 if path is not None: 

53 path.parent.mkdir(parents=True, exist_ok=True) 

54 path.write_text("") 

55 

56 def record( 

57 self, 

58 status: DiagnosticStatus, 

59 source: str, 

60 message: str, 

61 *, 

62 url: str = "", 

63 status_code: int | None = None, 

64 ) -> DiagnosticEvent: 

65 event = DiagnosticEvent( 

66 timestamp=self.clock(), 

67 status=status, 

68 source=source, 

69 message=message, 

70 url=url, 

71 status_code=status_code, 

72 ) 

73 with self._lock: 

74 self._events.append(event) 

75 del self._events[: max(0, len(self._events) - self.max_events)] 

76 if self.path is not None: 

77 with self.path.open("a") as output: 

78 output.write(event.format() + "\n") 

79 return event 

80 

81 def snapshot(self) -> list[DiagnosticEvent]: 

82 with self._lock: 

83 return list(self._events) 

84 

85 def has_status(self, *statuses: DiagnosticStatus) -> bool: 

86 wanted = set(statuses) 

87 return any(event.status in wanted for event in self.snapshot()) 

88 

89 

90def diagnostic_for_http_error(exc: Exception) -> str: 

91 """Return a concise, stable message without a traceback.""" 

92 if hasattr(exc, "response"): 

93 response = exc.response 

94 reason = response.reason_phrase or "HTTP error" 

95 return f"HTTP {response.status_code} {reason}" 

96 message = str(exc).strip() 

97 return message or exc.__class__.__name__