settled-reach/tooling/test_jobs.py

#!/usr/bin/env python3
"""Units for the jobs service (T-1278), called DIRECTLY — no CLI round trip.

That is half the point of these tests. D-263 says a service must be
transport-agnostic so it can be called by a test, by another service, or by a
future second front end. If nothing ever exercises that, the layering is
unverified decoration — a claim in a decision record with no evidence behind it.
Every test here imports `service` and calls a function.

The other half is the two behaviours that are easy to get wrong and impossible
to notice when they are:

  1. A partial trailing line. The log is appended to by a live process, so a
     reader can arrive mid-write. Parsing greedily would either crash or, worse,
     silently discard the event and advance past it — losing exactly one line,
     the one being written when someone looked.
  2. Reconciliation. A job killed outright cannot record its ending, so its file
     says `running` forever. Trusting the file leaves a corpse looking busy and
     hangs anything waiting on it.

Run: python3 tooling/test_jobs.py
"""

import json
import sys
import tempfile
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parent.parent))

from tooling.domains.jobs import service  # noqa: E402
from tooling.domains.jobs.schemas import Job, Status  # noqa: E402


def _job(**overrides) -> Job:
    base = {
        "job": "20260101T000000-test",
        "command": "reach check client-version",
        "argv": ["check", "client-version"],
        "pid": 1,
        "started_at": "2026-01-01T00:00:00",
        "status": Status.RUNNING,
    }
    return Job(**{**base, **overrides})


def test_partial_trailing_line(failures: list[str]) -> None:
    """A half-written final line is left for the next read, not dropped."""
    with tempfile.TemporaryDirectory() as tmp:
        log = Path(tmp) / "j.jsonl"
        complete = json.dumps({"message": "one"}) + "\n"
        partial = '{"message": "tw'
        log.write_text(complete + partial, encoding="utf-8")

        events, offset = service.read_events_from(log, 0)
        if len(events) != 1:
            failures.append(f"partial line: expected 1 complete event, got {len(events)}")
        if offset != len(complete):
            failures.append(
                f"partial line: offset {offset} should stop at {len(complete)}, "
                "before the incomplete line — otherwise that event is lost forever"
            )

        # Now the writer finishes the line. The event must appear.
        log.write_text(complete + json.dumps({"message": "two"}) + "\n", encoding="utf-8")
        events, _ = service.read_events_from(log, offset)
        if [e.get("message") for e in events] != ["two"]:
            failures.append(f"partial line: resumed read lost the completed event: {events}")


def test_offset_resume_is_stable(failures: list[str]) -> None:
    """Reading from the end returns nothing and does not move the offset."""
    with tempfile.TemporaryDirectory() as tmp:
        log = Path(tmp) / "j.jsonl"
        log.write_text(json.dumps({"message": "one"}) + "\n", encoding="utf-8")
        _, first = service.read_events_from(log, 0)
        events, second = service.read_events_from(log, first)
        if events or first != second:
            failures.append(
                f"offset resume: re-reading returned {len(events)} events and moved "
                f"{first}->{second}; a follow loop would replay forever"
            )


def test_dead_pid_is_reconciled(failures: list[str]) -> None:
    """running + a pid that is gone == died, never running."""
    # PID 1 exists; a very high pid almost certainly does not.
    alive = service._reconcile(_job(pid=1))
    if alive.status is not Status.RUNNING:
        failures.append("reconcile: a live pid was reported as not running")

    dead = service._reconcile(_job(pid=4_000_000))
    if dead.status is not Status.DIED:
        failures.append(
            f"reconcile: a dead pid stayed {dead.status.value} — a corpse that looks "
            "busy hangs every caller waiting on it"
        )


def test_died_never_relays_success(failures: list[str]) -> None:
    """A job that died has no exit code of its own, and must not borrow 0."""
    died = _job(status=Status.DIED)
    if died.effective_exit_code == 0:
        failures.append(
            "died job relayed exit 0 — that is the exit-0 trap: a killed job "
            "reported as success to whatever gated on it"
        )
    finished = _job(status=Status.FAILED, exit_code=2)
    if finished.effective_exit_code != 2:
        failures.append("failed job did not relay its own exit code")


def main() -> int:
    failures: list[str] = []
    test_partial_trailing_line(failures)
    test_offset_resume_is_stable(failures)
    test_dead_pid_is_reconciled(failures)
    test_died_never_relays_success(failures)

    if failures:
        print("test_jobs: FAIL", file=sys.stderr)
        for failure in failures:
            print(f"  - {failure}", file=sys.stderr)
        return 1
    print("test_jobs: OK — offsets resume cleanly, dead pids reconcile, died never relays 0")
    return 0


if __name__ == "__main__":
    sys.exit(main())