Skip to content

report

report

Publish matrix progress as an atomic JSON snapshot, independent of terminal output.

SlotReport(state='pending', reason='', observed='?', network_hint=None, duration=0, skips=None) dataclass

A slot's current phase and eventual outcome; duration includes its build.

state = 'pending' class-attribute instance-attribute

reason = '' class-attribute instance-attribute

observed = '?' class-attribute instance-attribute

network_hint = None class-attribute instance-attribute

duration = 0 class-attribute instance-attribute

skips = None class-attribute instance-attribute

MatrixReport(path, slots)

Write snapshots after each transition so readers can follow an unfinished run.

A missing destination disables file output. Readers see either the old snapshot or the new one, never half a JSON document.

Publish the initial pending slots before execution starts.

Source code in src/terok_util/matrix/report.py 
34
35
36
37
38
39
40
41
42
43
44
def __init__(self, path: Path | None, slots: list[str]) -> None:
    """Publish the initial pending slots before execution starts."""
    self.path = path
    self.slots = {name: SlotReport() for name in slots}
    self.state = "running"
    self.exit_code: int | None = None
    self.error = ""
    self._started = time.monotonic()
    self._slot_started: dict[str, float] = {}
    self._lock = threading.Lock()
    self._write()

path = path instance-attribute

slots = {name: SlotReport() for name in slots} instance-attribute

state = 'running' instance-attribute

exit_code = None instance-attribute

error = '' instance-attribute

update(name, state, *, reason='', observed='?', network_hint=None, skips=None)

Record a build, test, or verdict from either serial or parallel execution.

Source code in src/terok_util/matrix/report.py 
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
def update(
    self,
    name: str,
    state: str,
    *,
    reason: str = "",
    observed: str = "?",
    network_hint: str | None = None,
    skips: dict[str, dict[str, int]] | None = None,
) -> None:
    """Record a build, test, or verdict from either serial or parallel execution."""
    with self._lock:
        now = time.monotonic()
        started = self._slot_started.setdefault(name, now)
        self.slots[name] = SlotReport(
            state, reason, observed, network_hint, now - started, skips
        )
        self._write()

finish(exit_code, error='')

Close the report, retaining completed verdicts after errors or interruption.

Source code in src/terok_util/matrix/report.py 
65
66
67
68
69
70
71
72
73
74
75
76
def finish(self, exit_code: int, error: str = "") -> None:
    """Close the report, retaining completed verdicts after errors or interruption."""
    with self._lock:
        self.exit_code = exit_code
        self.error = error
        self.state = {0: "passed", 1: "failed", 130: "cancelled"}.get(exit_code, "error")
        for slot in self.slots.values():
            if slot.state in {"pending", "building", "built", "testing"}:
                if exit_code:
                    slot.state = "cancelled" if exit_code == 130 else "error"
                    slot.reason = error or "matrix ended before this slot completed"
        self._write()