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.
467 lines
15 KiB
Python
Executable File
467 lines
15 KiB
Python
Executable File
"""Real fixture processes and Git worktrees; never invoke product/remote mutations."""
|
|
|
|
from argparse import Namespace
|
|
import argparse
|
|
import json
|
|
from pathlib import Path
|
|
import subprocess
|
|
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"))
|
|
from govoplan_devkit import runner
|
|
from govoplan_devkit.checkpoints import Checkpoints
|
|
from govoplan_devkit.common import (
|
|
atomic_json,
|
|
digest,
|
|
read_json,
|
|
resource_lock,
|
|
state_root,
|
|
)
|
|
from govoplan_devkit.workspace import load_project, source_fingerprint
|
|
|
|
|
|
@pytest.fixture
|
|
def example(tmp_path, monkeypatch):
|
|
monkeypatch.setenv("XDG_STATE_HOME", str(tmp_path / "xdg-state"))
|
|
workspace = tmp_path / "workspace"
|
|
repo = workspace / "example"
|
|
repo.mkdir(parents=True)
|
|
subprocess.run(["git", "init", "-q", str(repo)], check=True)
|
|
(repo / "source.txt").write_text("first\n")
|
|
subprocess.run(["git", "-C", str(repo), "add", "source.txt"], check=True)
|
|
subprocess.run(
|
|
[
|
|
"git",
|
|
"-C",
|
|
str(repo),
|
|
"-c",
|
|
"user.name=Fixture",
|
|
"-c",
|
|
"user.email=fixture@example.invalid",
|
|
"commit",
|
|
"-qm",
|
|
"fixture",
|
|
],
|
|
check=True,
|
|
)
|
|
project = tmp_path / "project.json"
|
|
project.write_text(
|
|
json.dumps(
|
|
{
|
|
"schema_version": 1,
|
|
"name": "Example",
|
|
"repositories": [{"name": "example", "path": "example"}],
|
|
"checks": [],
|
|
"profiles": {},
|
|
}
|
|
)
|
|
)
|
|
args = Namespace(
|
|
workspace_root=workspace,
|
|
project=project,
|
|
state_dir=tmp_path / "state",
|
|
dry_run=False,
|
|
jobs=2,
|
|
profile="quick",
|
|
resume=None,
|
|
)
|
|
return args, repo
|
|
|
|
|
|
def stage(repo, name="one", code="print('ok')", **extra):
|
|
return {
|
|
"id": name,
|
|
"title": name,
|
|
"argv": [sys.executable, "-c", code],
|
|
"cwd": str(repo),
|
|
"timeout_seconds": 10,
|
|
**extra,
|
|
}
|
|
|
|
|
|
def run(args, stages):
|
|
# Environment inspection is tested separately; keep fixture executions cheap/deterministic.
|
|
with patch.object(runner, "environment_fingerprint", return_value="fixture-env"):
|
|
return runner.run_checks(args, stages)
|
|
|
|
|
|
def test_success_receipt_and_compact_summary(example):
|
|
args, repo = example
|
|
result = run(args, [stage(repo)])
|
|
assert result["status"] == "passed"
|
|
receipt = runner.read_receipt(args.workspace_root, args.state_dir, result["run_id"])
|
|
assert receipt["stages"][0]["exit_code"] == 0
|
|
assert Path(receipt["stages"][0]["log_path"]).read_text() == "ok\n"
|
|
assert runner.summarize(receipt)["counts"] == {"passed": 1}
|
|
assert Path(result["receipt_path"]).stat().st_mode & 0o777 == 0o600
|
|
|
|
|
|
def test_native_environment_binds_selected_checkout(tmp_path, monkeypatch):
|
|
from govoplan_devkit.environment import execution_environment
|
|
|
|
monkeypatch.setenv("GOVOPLAN_WORKSPACE_ROOT", "/another/workspace")
|
|
monkeypatch.setenv("GOVOPLAN_CORE_ROOT", "/another/core")
|
|
monkeypatch.setenv("GOVOPLAN_CORE_SOURCE_ROOT", "/another/source")
|
|
project = load_project(tmp_path)
|
|
env = execution_environment(
|
|
tmp_path, project, {"python": sys.executable, "node": "node", "npm": "npm"}
|
|
)
|
|
assert env["GOVOPLAN_WORKSPACE_ROOT"] == str(tmp_path)
|
|
assert env["GOVOPLAN_CORE_ROOT"] == str(tmp_path / "govoplan-core")
|
|
assert env["GOVOPLAN_CORE_SOURCE_ROOT"] == str(tmp_path / "govoplan-core")
|
|
|
|
|
|
def test_portable_environment_does_not_inject_govoplan_scope(example, monkeypatch):
|
|
from govoplan_devkit.environment import execution_environment
|
|
|
|
args, _ = example
|
|
for key in (
|
|
"GOVOPLAN_WORKSPACE_ROOT",
|
|
"GOVOPLAN_CORE_ROOT",
|
|
"GOVOPLAN_CORE_SOURCE_ROOT",
|
|
):
|
|
monkeypatch.delenv(key, raising=False)
|
|
env = execution_environment(
|
|
args.workspace_root,
|
|
load_project(args.workspace_root, args.project),
|
|
{"python": sys.executable, "node": "node", "npm": "npm"},
|
|
)
|
|
assert "GOVOPLAN_WORKSPACE_ROOT" not in env
|
|
|
|
|
|
def test_failure_skips_dependents_but_runs_independent_check(example):
|
|
args, repo = example
|
|
result = run(
|
|
args,
|
|
[
|
|
stage(repo, "bad", "raise SystemExit(4)"),
|
|
stage(repo, "dependent", deps=["bad"]),
|
|
stage(repo, "independent"),
|
|
],
|
|
)
|
|
states = {item["id"]: item["status"] for item in result["stages"]}
|
|
assert states == {"bad": "failed", "dependent": "skipped", "independent": "passed"}
|
|
assert result["_exit_code"] == 1
|
|
|
|
|
|
def test_timeout_is_not_a_pass(example):
|
|
args, repo = example
|
|
result = run(
|
|
args, [stage(repo, code="import time; time.sleep(10)", timeout_seconds=0.1)]
|
|
)
|
|
assert result["stages"][0]["status"] == "timed_out"
|
|
|
|
|
|
def test_dry_run_does_not_create_state(example):
|
|
args, repo = example
|
|
args.dry_run = True
|
|
result = run(args, [stage(repo, code="raise SystemExit(9)")])
|
|
assert result["status"] == "planned"
|
|
assert not args.state_dir.exists()
|
|
|
|
|
|
def test_empty_plan_is_not_claimed_as_verified(example):
|
|
args, _ = example
|
|
assert run(args, [])["status"] == "not_run"
|
|
|
|
|
|
def test_plan_validation_rejects_cycles_unknowns_duplicate_ids_escape(example):
|
|
args, repo = example
|
|
invalid = [
|
|
[stage(repo, deps=["unknown"])],
|
|
[stage(repo, "a", deps=["b"]), stage(repo, "b", deps=["a"])],
|
|
[stage(repo), stage(repo)],
|
|
[stage(repo, "../escape")],
|
|
[stage(repo.parent.parent)],
|
|
]
|
|
for plan in invalid:
|
|
with pytest.raises(ValueError):
|
|
runner.validate_stages(plan, args.workspace_root, {})
|
|
|
|
|
|
def test_source_identity_includes_staged_unstaged_and_untracked_bytes(example):
|
|
args, repo = example
|
|
project = load_project(args.workspace_root, args.project)
|
|
first = source_fingerprint(project)
|
|
(repo / "source.txt").write_text("second\n")
|
|
second = source_fingerprint(project)
|
|
assert second != first
|
|
subprocess.run(["git", "-C", str(repo), "add", "source.txt"], check=True)
|
|
third = source_fingerprint(project)
|
|
assert third != second
|
|
(repo / "extra.txt").write_text("extra")
|
|
assert source_fingerprint(project) != third
|
|
|
|
|
|
def test_source_mutation_during_run_invalidates_receipt(example):
|
|
args, repo = example
|
|
result = run(
|
|
args,
|
|
[
|
|
stage(
|
|
repo,
|
|
code="from pathlib import Path; Path('source.txt').write_text('changed')",
|
|
)
|
|
],
|
|
)
|
|
assert result["stages"][0]["status"] == "stale"
|
|
assert result["stages"][0]["checkpoint_verified"] is False
|
|
assert result["status"] == "stale"
|
|
|
|
|
|
def test_resume_reuses_only_identical_plan_and_inputs(example):
|
|
args, repo = example
|
|
plan = [stage(repo)]
|
|
first = run(args, plan)
|
|
args.resume = first["run_id"]
|
|
second = run(args, plan)
|
|
assert second["stages"][0]["reused_from"] == first["run_id"]
|
|
changed_plan = run(args, [stage(repo, code="print('different')")])
|
|
assert changed_plan["status"] == "passed"
|
|
assert "reused_from" not in changed_plan["stages"][0]
|
|
assert Path(changed_plan["stages"][0]["log_path"]).read_text() == "different\n"
|
|
(repo / "source.txt").write_text("changed")
|
|
changed_source = run(args, plan)
|
|
assert changed_source["status"] == "passed"
|
|
assert "reused_from" not in changed_source["stages"][0]
|
|
assert changed_source["stages"][0]["cache_key"] != first["stages"][0]["cache_key"]
|
|
|
|
|
|
def test_receipt_integrity_and_foreign_id_are_rejected(example):
|
|
args, repo = example
|
|
result = run(args, [stage(repo)])
|
|
path = Path(result["receipt_path"])
|
|
record = read_json(path)
|
|
record["status"] = "fabricated"
|
|
atomic_json(path, record)
|
|
with pytest.raises(ValueError, match="integrity"):
|
|
runner.read_receipt(args.workspace_root, args.state_dir, result["run_id"])
|
|
with pytest.raises(ValueError):
|
|
runner.read_receipt(args.workspace_root, args.state_dir, "../other")
|
|
|
|
|
|
def test_shared_resources_serialize_stages(example):
|
|
args, repo = example
|
|
active = 0
|
|
maximum = 0
|
|
lock = threading.Lock()
|
|
original = runner._execute_stage_command
|
|
|
|
def execute(*args, **kwargs):
|
|
nonlocal active, maximum
|
|
with lock:
|
|
active += 1
|
|
maximum = max(maximum, active)
|
|
try:
|
|
time.sleep(0.04)
|
|
return original(*args, **kwargs)
|
|
finally:
|
|
with lock:
|
|
active -= 1
|
|
|
|
with patch.object(runner, "_execute_stage_command", side_effect=execute):
|
|
result = run(
|
|
args,
|
|
[
|
|
stage(repo, "a", resources=["shared"]),
|
|
stage(repo, "b", resources=["shared"]),
|
|
],
|
|
)
|
|
assert result["status"] == "passed"
|
|
assert maximum == 1
|
|
|
|
|
|
def test_cross_process_resource_conflict_is_explicit(example):
|
|
args, repo = example
|
|
locks = state_root(args.workspace_root) / "resource-locks"
|
|
with resource_lock(locks, "shared"):
|
|
result = run(args, [stage(repo, resources=["shared"])])
|
|
assert result["stages"][0]["status"] == "blocked"
|
|
|
|
|
|
def test_output_is_bounded_and_known_secrets_redacted(example, monkeypatch):
|
|
args, repo = example
|
|
monkeypatch.setenv("FIXTURE_SECRET", "private-fixture-credential")
|
|
with patch.object(runner, "MAX_LOG_BYTES", 1024):
|
|
result = run(
|
|
args,
|
|
[
|
|
stage(
|
|
repo,
|
|
code="import os; print(os.environ['FIXTURE_SECRET']); print('x'*10000)",
|
|
)
|
|
],
|
|
)
|
|
record = result["stages"][0]
|
|
output = Path(record["log_path"]).read_text()
|
|
assert record["output_truncated"] is True
|
|
assert "private-fixture-credential" not in output
|
|
assert "[redacted]" in output
|
|
assert len(output) < 2000
|
|
|
|
|
|
def test_symlinked_state_is_rejected(example, tmp_path):
|
|
args, repo = example
|
|
target = tmp_path / "target"
|
|
target.mkdir()
|
|
args.state_dir.symlink_to(target, target_is_directory=True)
|
|
with pytest.raises(ValueError, match="symlink"):
|
|
run(args, [stage(repo)])
|
|
|
|
|
|
@pytest.mark.parametrize("flag", ["--assume-unchanged", "--skip-worktree"])
|
|
def test_hidden_tracked_edits_invalidate_source_identity(example, flag):
|
|
args, repo = example
|
|
subprocess.run(
|
|
["git", "-C", str(repo), "update-index", flag, "source.txt"], check=True
|
|
)
|
|
project = load_project(args.workspace_root, args.project)
|
|
initial = source_fingerprint(project)
|
|
(repo / "source.txt").write_text("hidden edit\n")
|
|
assert source_fingerprint(project) != initial
|
|
|
|
|
|
def test_failed_final_snapshot_never_persists_a_passing_receipt(example):
|
|
args, repo = example
|
|
events = []
|
|
args.on_progress = events.append
|
|
original = Checkpoints.source
|
|
|
|
def fingerprint(self, *values, **kwargs):
|
|
if events and events[-1]["phase"] == "finalizing":
|
|
raise ValueError("unreadable source")
|
|
return original(self, *values, **kwargs)
|
|
|
|
with patch.object(Checkpoints, "source", fingerprint):
|
|
result = run(args, [stage(repo)])
|
|
assert result["status"] == "stale"
|
|
assert result["snapshot_verified"] is False
|
|
assert result["_exit_code"] == 1
|
|
assert "unreadable source" in result["invalidated_reason"]
|
|
assert (
|
|
runner.read_receipt(args.workspace_root, args.state_dir, result["run_id"])[
|
|
"status"
|
|
]
|
|
== "stale"
|
|
)
|
|
|
|
|
|
def test_resume_rejects_a_modified_cached_log(example):
|
|
args, repo = example
|
|
plan = [stage(repo)]
|
|
first = run(args, plan)
|
|
Path(first["stages"][0]["log_path"]).write_text("altered")
|
|
args.resume = first["run_id"]
|
|
result = run(args, plan)
|
|
assert result["status"] == "failed"
|
|
assert result["stages"][0]["status"] == "failed"
|
|
assert "log integrity" in result["stages"][0]["error"]
|
|
assert "reused_from" not in result["stages"][0]
|
|
|
|
|
|
def test_receipt_argument_redaction_applies_to_unknown_separate_values(example):
|
|
args, repo = example
|
|
plan = stage(repo)
|
|
plan["argv"].extend(["--password", "fixture-not-from-environment"])
|
|
result = run(args, [plan])
|
|
assert (
|
|
"fixture-not-from-environment" not in Path(result["receipt_path"]).read_text()
|
|
)
|
|
|
|
|
|
def recovery_parser():
|
|
parser = argparse.ArgumentParser()
|
|
runner.register(parser.add_subparsers())
|
|
return parser
|
|
|
|
|
|
def test_recovery_requires_manual_survivor_confirmation(example):
|
|
args, repo = example
|
|
result = run(args, [stage(repo)])
|
|
record = read_json(Path(result["receipt_path"]))
|
|
record["status"] = "running"
|
|
record["snapshot_verified"] = False
|
|
atomic_json(Path(result["receipt_path"]), runner._seal(record))
|
|
parsed = recovery_parser().parse_args(
|
|
["recover", result["run_id"], "--apply"], namespace=args
|
|
)
|
|
with pytest.raises(ValueError, match="manual verification"):
|
|
parsed.handler(parsed)
|
|
|
|
|
|
def test_recovery_does_not_overwrite_completion_between_reads(example):
|
|
args, repo = example
|
|
result = run(args, [stage(repo)])
|
|
completed = runner.read_receipt(
|
|
args.workspace_root, args.state_dir, result["run_id"]
|
|
)
|
|
initial = {**completed, "status": "running"}
|
|
parsed = recovery_parser().parse_args(
|
|
["recover", result["run_id"], "--apply", "--confirm-processes-stopped"],
|
|
namespace=args,
|
|
)
|
|
with (
|
|
patch.object(runner, "read_receipt", side_effect=[initial, completed]),
|
|
patch.object(runner, "atomic_json") as write,
|
|
):
|
|
assert parsed.handler(parsed)["status"] == "unchanged"
|
|
write.assert_not_called()
|
|
|
|
|
|
def test_active_owner_lock_blocks_recovery(example):
|
|
args, repo = example
|
|
result = run(args, [stage(repo)])
|
|
record = read_json(Path(result["receipt_path"]))
|
|
record.update(status="running", snapshot_verified=False)
|
|
atomic_json(Path(result["receipt_path"]), runner._seal(record))
|
|
parsed = recovery_parser().parse_args(
|
|
["recover", result["run_id"], "--apply", "--confirm-processes-stopped"],
|
|
namespace=args,
|
|
)
|
|
with resource_lock(
|
|
state_root(args.workspace_root, args.state_dir) / "locks",
|
|
"run:" + result["run_id"],
|
|
):
|
|
with pytest.raises(RuntimeError, match="busy"):
|
|
parsed.handler(parsed)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"mutation",
|
|
[
|
|
"unknown_status",
|
|
"unverified_pass",
|
|
"duplicate_stage",
|
|
"nonboolean_verification",
|
|
"nonpassing_stage",
|
|
],
|
|
)
|
|
def test_even_correctly_hashed_receipts_must_have_consistent_states(example, mutation):
|
|
args, repo = example
|
|
result = run(args, [stage(repo)])
|
|
path = Path(result["receipt_path"])
|
|
record = read_json(path)
|
|
if mutation == "unknown_status":
|
|
record["status"] = "wonderful"
|
|
elif mutation == "unverified_pass":
|
|
record["snapshot_verified"] = False
|
|
elif mutation == "duplicate_stage":
|
|
record["stages"].append(record["stages"][0])
|
|
elif mutation == "nonboolean_verification":
|
|
record["snapshot_verified"] = "true"
|
|
else:
|
|
record["stages"][0]["status"] = "failed"
|
|
# Deliberately construct an externally corrupted but correctly hashed record;
|
|
# the runner's writer now rejects inconsistent checkpoint state before sealing.
|
|
record["integrity_sha256"] = digest(
|
|
{key: value for key, value in record.items() if key != "integrity_sha256"}
|
|
)
|
|
atomic_json(path, record)
|
|
with pytest.raises(ValueError):
|
|
runner.read_receipt(args.workspace_root, args.state_dir, result["run_id"])
|