"""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)