zemion
2ffdb23f69
Verified with the coordinated workspace changes by devkit full run 2026-09-08T225814-186389-0000-3e3ed7cd (all seven phases passed). This shared UI pass does not mark the individual module reviews complete.
320 lines
12 KiB
Python
Executable File
320 lines
12 KiB
Python
Executable File
"""Live/provisional monitoring and discovery never substitute for verified evidence."""
|
|
|
|
from argparse import Namespace
|
|
import argparse
|
|
import json
|
|
from pathlib import Path
|
|
import sys
|
|
import threading
|
|
import time
|
|
from unittest.mock import patch
|
|
|
|
import pytest
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "tools/devkit"))
|
|
sys.path.insert(0, str(Path(__file__).resolve().parent))
|
|
from test_devkit_runner import example as example, run, stage
|
|
from govoplan_devkit import catalog, cli, monitoring, runner
|
|
from govoplan_devkit.checkpoints import Checkpoints
|
|
from govoplan_devkit.common import atomic_json, read_json, state_root
|
|
from govoplan_devkit.process import OutputSnapshot
|
|
|
|
|
|
def command(args, *argv):
|
|
parser = argparse.ArgumentParser()
|
|
runner.register(parser.add_subparsers(dest="command", required=True))
|
|
selected = parser.parse_args(argv, namespace=Namespace(**vars(args)))
|
|
return selected.handler(selected)
|
|
|
|
|
|
def test_preparing_receipt_and_run_id_exist_before_source_fingerprinting(example):
|
|
args, repo = example
|
|
events = []
|
|
args.on_progress = events.append
|
|
original = Checkpoints.source
|
|
|
|
def fingerprint(self, *values, **kwargs):
|
|
assert events and events[0]["phase"] == "preparing"
|
|
first = events[0]
|
|
assert Path(first["receipt_path"]).is_file()
|
|
record = runner.read_receipt(
|
|
args.workspace_root, args.state_dir, first["run_id"]
|
|
)
|
|
if record["phase"] == "preparing":
|
|
assert record["status"] == "running"
|
|
assert record["snapshot_verified"] is False
|
|
assert record["source_fingerprint"] is None
|
|
return original(self, *values, **kwargs)
|
|
|
|
with patch.object(Checkpoints, "source", fingerprint):
|
|
result = run(args, [stage(repo)])
|
|
assert result["status"] == "passed"
|
|
assert events[-1]["phase"] == "finished"
|
|
assert events[-1]["run_id"] == events[0]["run_id"] == result["run_id"]
|
|
|
|
|
|
def test_preflight_failure_persists_discoverable_nonpassing_receipt(example):
|
|
args, repo = example
|
|
events = []
|
|
args.on_progress = events.append
|
|
with patch.object(
|
|
Checkpoints,
|
|
"source",
|
|
side_effect=ValueError("fixture fingerprint unavailable"),
|
|
):
|
|
with pytest.raises(ValueError, match="fixture fingerprint unavailable"):
|
|
run(args, [stage(repo)])
|
|
assert events[0]["phase"] == "preparing"
|
|
record = runner.read_receipt(
|
|
args.workspace_root, args.state_dir, events[0]["run_id"]
|
|
)
|
|
assert record["status"] == "failed"
|
|
assert record["phase"] == "finished"
|
|
assert record["snapshot_verified"] is False
|
|
assert all(item["status"] != "passed" for item in record["stages"])
|
|
assert "fixture fingerprint unavailable" in record["error"]
|
|
assert monitoring.latest_run(args)["status"] == "failed"
|
|
|
|
|
|
def test_run_history_is_read_only_and_cursor_pagination_has_no_duplicates(example):
|
|
args, repo = example
|
|
created = [run(args, [stage(repo)])["run_id"] for _ in range(3)]
|
|
base = state_root(args.workspace_root, args.state_dir)
|
|
before = {
|
|
path: (path.stat().st_mtime_ns, path.read_bytes())
|
|
for path in base.rglob("*")
|
|
if path.is_file()
|
|
}
|
|
page = monitoring.list_runs(Namespace(**vars(args), limit=2, before=None))
|
|
assert [item["run_id"] for item in page["runs"]] == sorted(created, reverse=True)[
|
|
:2
|
|
]
|
|
assert page["next_cursor"]
|
|
next_page = monitoring.list_runs(
|
|
Namespace(**vars(args), limit=2, before=page["next_cursor"])
|
|
)
|
|
assert [item["run_id"] for item in next_page["runs"]] == sorted(
|
|
created, reverse=True
|
|
)[2:]
|
|
assert next_page["next_cursor"] is None
|
|
assert monitoring.latest_run(args)["run_id"] == max(created)
|
|
assert before == {
|
|
path: (path.stat().st_mtime_ns, path.read_bytes())
|
|
for path in base.rglob("*")
|
|
if path.is_file()
|
|
}
|
|
|
|
|
|
def test_empty_run_history_does_not_create_a_state_directory(example):
|
|
args, _ = example
|
|
assert not args.state_dir.exists()
|
|
result = monitoring.list_runs(args)
|
|
assert result["runs"] == []
|
|
assert monitoring.latest_run(args)["status"] == "not_found"
|
|
assert not args.state_dir.exists()
|
|
|
|
|
|
def test_invalid_newest_run_is_visible_and_never_replaced_with_older_pass(example):
|
|
args, repo = example
|
|
older = run(args, [stage(repo)])["run_id"]
|
|
invalid = "zz-invalid-newest"
|
|
path = (
|
|
state_root(args.workspace_root, args.state_dir)
|
|
/ "runs"
|
|
/ invalid
|
|
/ "receipt.json"
|
|
)
|
|
atomic_json(path, {"malformed": True})
|
|
rows = monitoring.list_runs(args)
|
|
assert rows["runs"][0]["run_id"] == invalid
|
|
assert rows["runs"][0]["status"] == "invalid"
|
|
assert any(row["run_id"] == older for row in rows["runs"])
|
|
assert rows["_exit_code"] == 1
|
|
with pytest.raises(ValueError, match="Latest run is invalid"):
|
|
monitoring.latest_run(args)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"changes", [{"limit": 0}, {"limit": 101}, {"limit": True}, {"before": "../escape"}]
|
|
)
|
|
def test_history_rejects_invalid_bounds_and_cursor(example, changes):
|
|
args, _ = example
|
|
with pytest.raises(ValueError):
|
|
monitoring.list_runs(Namespace(**{**vars(args), **changes}))
|
|
|
|
|
|
def test_live_logs_are_available_before_completion_but_final_only_refuses_them(example):
|
|
args, repo = example
|
|
events, results, errors = [], [], []
|
|
args.on_progress = events.append
|
|
|
|
def execute():
|
|
try:
|
|
results.append(
|
|
run(
|
|
args,
|
|
[
|
|
stage(
|
|
repo,
|
|
code="import time; print('live ready',flush=True); time.sleep(1.5); print('finished')",
|
|
)
|
|
],
|
|
)
|
|
)
|
|
except BaseException as exc:
|
|
errors.append(exc)
|
|
|
|
worker = threading.Thread(target=execute)
|
|
worker.start()
|
|
try:
|
|
deadline = time.monotonic() + 4
|
|
live = None
|
|
while time.monotonic() < deadline:
|
|
if events:
|
|
live = command(args, "logs", events[0]["run_id"], "--stage", "one")
|
|
if "live ready" in live["excerpt"]:
|
|
break
|
|
time.sleep(0.02)
|
|
assert live is not None and "live ready" in live["excerpt"]
|
|
assert live["provisional"] is True
|
|
assert live["log_verified"] is False
|
|
assert live["snapshot_verified"] is False
|
|
assert live["run_status"] == "running"
|
|
with pytest.raises(ValueError, match="No finalized stage log"):
|
|
command(args, "logs", events[0]["run_id"], "--stage", "one", "--final-only")
|
|
finally:
|
|
worker.join(timeout=5)
|
|
assert not worker.is_alive() and not errors
|
|
final = command(
|
|
args, "logs", results[0]["run_id"], "--stage", "one", "--final-only"
|
|
)
|
|
assert final["provisional"] is False and final["log_verified"] is True
|
|
assert final["snapshot_verified"] is True
|
|
assert "finished" in final["excerpt"]
|
|
|
|
|
|
@pytest.mark.parametrize("quiet", [False, True])
|
|
def test_cli_keeps_json_stdout_clean_and_progress_on_stderr_or_quiet(
|
|
tmp_path, capsys, quiet
|
|
):
|
|
event = {
|
|
"event": "check_progress",
|
|
"run_id": "fixture",
|
|
"phase": "preparing",
|
|
"status": "running",
|
|
"counts": {"pending": 1},
|
|
"total_stages": 1,
|
|
"elapsed_seconds": 0,
|
|
"active_stages": [],
|
|
"receipt_path": str(tmp_path / "receipt.json"),
|
|
}
|
|
|
|
def check(args, _stages):
|
|
args.on_progress(event)
|
|
return {"status": "passed", "summary": ["fixture complete"]}
|
|
|
|
with (
|
|
patch.object(catalog, "build_stages", return_value=[]),
|
|
patch.object(runner, "run_checks", side_effect=check),
|
|
):
|
|
assert (
|
|
cli.main(
|
|
[
|
|
"check",
|
|
"--workspace-root",
|
|
str(tmp_path),
|
|
"--json",
|
|
*(["--quiet"] if quiet else []),
|
|
]
|
|
)
|
|
== 0
|
|
)
|
|
output = capsys.readouterr()
|
|
assert json.loads(output.out)["status"] == "passed"
|
|
if quiet:
|
|
assert output.err == ""
|
|
else:
|
|
assert json.loads(output.err)["event"] == "check_progress"
|
|
|
|
|
|
def snapshot(data, *, split=0, omitted=0, final=False):
|
|
return OutputSnapshot(data, b"", bool(omitted), omitted, 0, split, 0, final)
|
|
|
|
|
|
def test_provisional_output_withholds_incomplete_secret_line(monkeypatch):
|
|
monkeypatch.setenv("FIXTURE_SECRET", "credential-known-to-redaction")
|
|
data = b"public line\nAPI_KEY=credential-known-to-"
|
|
live = monitoring.capture_text(snapshot(data), provisional=True)
|
|
assert live == "public line\n"
|
|
final = monitoring.capture_text(
|
|
snapshot(b"public line\nAPI_KEY=credential-known-to-redaction", final=True)
|
|
)
|
|
assert "credential-known" not in final
|
|
assert "[redacted]" in final
|
|
|
|
|
|
def test_truncated_boundaries_cannot_expose_cut_authorization_or_secret_fragments():
|
|
head = b"safe head\nAuthorization: Bearer partial-secret-head"
|
|
tail = b"partial-secret-tail\nsafe final line\n"
|
|
for provisional in (False, True):
|
|
text = monitoring.capture_text(
|
|
snapshot(head + tail, split=len(head), omitted=90), provisional=provisional
|
|
)
|
|
assert "safe head" in text and "safe final line" in text
|
|
assert "partial-secret" not in text
|
|
assert "90 bytes omitted" in text
|
|
|
|
|
|
def test_final_failure_tail_survives_retention_and_stays_redacted(example, monkeypatch):
|
|
args, repo = example
|
|
monkeypatch.setenv("FIXTURE_SECRET", "sensitive-final-credential")
|
|
with patch.object(runner, "MAX_LOG_BYTES", 512):
|
|
result = run(
|
|
args,
|
|
[
|
|
stage(
|
|
repo,
|
|
code="import os; print('begin'); print('x'*20000); print(os.environ['FIXTURE_SECRET']); print('FINAL IMPORTANT ERROR'); raise SystemExit(9)",
|
|
)
|
|
],
|
|
)
|
|
final = command(
|
|
args, "logs", result["run_id"], "--stage", "one", "--final-only"
|
|
)
|
|
assert result["status"] == "failed"
|
|
assert final["log_verified"] is True
|
|
assert "FINAL IMPORTANT ERROR" in final["excerpt"]
|
|
assert "sensitive-final-credential" not in final["excerpt"]
|
|
assert "[redacted]" in final["excerpt"]
|
|
assert result["stages"][0]["omitted_output_bytes"] > 0
|
|
|
|
|
|
@pytest.mark.parametrize("maximum", [1, 2, 3, 7, 75, 76, 77, 80, 81, 82, 255])
|
|
@pytest.mark.parametrize("tail_only", [False, True])
|
|
def test_tiny_display_bounds_preserve_valid_utf8_and_byte_limit(maximum, tail_only):
|
|
value = "Ä😊 " * 200 + "FINAL"
|
|
text = monitoring.bounded_display(value, maximum, tail_only=tail_only)
|
|
assert len(text.encode("utf-8", errors="strict")) <= maximum
|
|
assert text.endswith("L")
|
|
if maximum >= 5 and (tail_only or maximum >= 255):
|
|
assert text.endswith("FINAL")
|
|
|
|
|
|
def test_live_record_is_bounded_provisional_and_rejects_mismatched_identity(example):
|
|
args, _ = example
|
|
run_id, stage_id = "fixture-run", "fixture-stage"
|
|
path = (
|
|
state_root(args.workspace_root, args.state_dir)
|
|
/ "runs"
|
|
/ run_id
|
|
/ (stage_id + ".log")
|
|
)
|
|
monitoring.write_live(path, stage_id, snapshot(b"line\n"), time.monotonic())
|
|
live = monitoring.read_live(args.workspace_root, args.state_dir, run_id, stage_id)
|
|
assert live["provisional"] is True and live["excerpt"] == "line\n"
|
|
record = read_json(path.with_suffix(".live.json"))
|
|
record["run_id"] = "different-run"
|
|
atomic_json(path.with_suffix(".live.json"), record)
|
|
with pytest.raises(ValueError, match="Invalid provisional log"):
|
|
monitoring.read_live(args.workspace_root, args.state_dir, run_id, stage_id)
|