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.
388 lines
18 KiB
Python
Executable File
388 lines
18 KiB
Python
Executable File
import argparse
|
|
from copy import deepcopy
|
|
import json
|
|
from pathlib import Path
|
|
import subprocess
|
|
import sys
|
|
|
|
import pytest
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "tools/devkit"))
|
|
from govoplan_devkit import issues
|
|
from govoplan_devkit.common import atomic_json, digest, state_root
|
|
from govoplan_devkit.workspace import load_project, source_fingerprint
|
|
|
|
|
|
@pytest.fixture
|
|
def args(tmp_path, monkeypatch):
|
|
monkeypatch.delenv("GITEA_TOKEN", raising=False)
|
|
root = tmp_path / "repo"
|
|
root.mkdir()
|
|
subprocess.run(["git", "init", "-q", str(root)], check=True)
|
|
subprocess.run(["git", "-C", str(root), "remote", "add", "origin", "https://gitea.invalid/team/repo.git"], check=True)
|
|
project = tmp_path / "project.json"
|
|
project.write_text(json.dumps({"schema_version": 1, "name": "Fixture", "repositories": [{"name": "repo", "path": "repo"}]}))
|
|
env_file = tmp_path / "private.env"
|
|
env_file.write_text("GITEA_TOKEN=fixture-private-token\nGITEA_OWNER=wrong-owner\n")
|
|
return argparse.Namespace(workspace_root=tmp_path, state_dir=tmp_path / "state", project=project,
|
|
root=root, issue=7, target_plan=None, remote="origin", env_file=env_file, evidence=None,
|
|
key="verification", note_summary=["Verified scoped changes."], next_steps=["Manual review remains."],
|
|
body_file=None, note_file=None, apply=False, retry_uncertain=False)
|
|
|
|
|
|
class FakeClient:
|
|
def __init__(self, target):
|
|
self.target = target
|
|
self.calls = []
|
|
self.comments = []
|
|
self.issue_id = 700
|
|
self.post_mode = "normal"
|
|
self.bad_binding = False
|
|
self.repeat_pages = False
|
|
self.bad_issue = False
|
|
self.fail = False
|
|
self.page_size = 2 # A server may enforce a cap smaller than the requested limit.
|
|
|
|
def close(self):
|
|
pass
|
|
|
|
def request_json(self, method, path, body=None, query=None):
|
|
self.calls.append((method, path, deepcopy(body), deepcopy(query)))
|
|
if self.fail:
|
|
raise RuntimeError("Remote echoed fixture-private-token")
|
|
if method == "GET" and path == self.target.path:
|
|
return {"id": self.issue_id, "number": self.target.issue, "html_url": self.target.url + ("wrong" if self.bad_issue else ""),
|
|
"state": "closed", "body": "- [ ] Preserve this checklist"}
|
|
if method == "GET" and path == self.target.path + "/comments":
|
|
page = 1 if self.repeat_pages else query["page"]
|
|
return deepcopy(self.comments[(page - 1) * self.page_size:page * self.page_size])
|
|
if method == "GET" and "/issues/comments/" in path:
|
|
comment = deepcopy(next(item for item in self.comments if str(item["id"]) == path.rsplit("/", 1)[1]))
|
|
comment["html_url"] = self.target.url + ("-other" if self.bad_binding else "") + "#issuecomment-" + str(comment["id"])
|
|
return comment
|
|
if method == "POST" and path == self.target.path + "/comments":
|
|
if self.post_mode == "timeout-before-commit":
|
|
raise TimeoutError("fixture-private-token")
|
|
comment = {"id": 1000 + len(self.comments), "body": body["body"]}
|
|
self.comments.append(comment)
|
|
if self.post_mode == "timeout-after-commit":
|
|
raise TimeoutError("fixture-private-token")
|
|
return deepcopy(comment)
|
|
raise AssertionError((method, path))
|
|
|
|
@property
|
|
def posts(self):
|
|
return [call for call in self.calls if call[0] == "POST"]
|
|
|
|
|
|
@pytest.fixture
|
|
def client(args, monkeypatch):
|
|
client = FakeClient(issues.resolve_target(args.root, args.issue, args.workspace_root))
|
|
monkeypatch.setattr(issues, "make_client", lambda target, token: client)
|
|
return client
|
|
|
|
|
|
def test_default_dry_run_is_offline_and_ignores_ambient_target_overrides(args, monkeypatch):
|
|
monkeypatch.setenv("GITEA_OWNER", "wrong")
|
|
monkeypatch.setenv("GITEA_REPO", "wrong")
|
|
monkeypatch.setenv("GITEA_URL", "https://wrong.invalid")
|
|
monkeypatch.setattr(issues, "make_client", lambda *_: pytest.fail("No network in preview"))
|
|
args.env_file = args.workspace_root / "does-not-exist.env"
|
|
result = issues.handle_note(args)
|
|
assert result["targets"][0]["url"] == "https://gitea.invalid/team/repo/issues/7"
|
|
assert result["targets"][0]["status"] == "would-post"
|
|
assert not args.state_dir.exists()
|
|
|
|
|
|
def test_scoped_checkpoint_evidence_compares_only_recorded_sources(args, monkeypatch):
|
|
from govoplan_devkit import runner
|
|
|
|
other = args.workspace_root / "other"
|
|
other.mkdir()
|
|
subprocess.run(["git", "init", "-q", str(other)], check=True)
|
|
declaration = json.loads(args.project.read_text())
|
|
declaration["repositories"].append({"name": "other", "path": "other"})
|
|
args.project.write_text(json.dumps(declaration))
|
|
monkeypatch.setattr(runner, "environment_fingerprint", lambda *_: "fixture-env")
|
|
run_args = argparse.Namespace(**{**vars(args), "jobs": 1, "profile": "quick", "dry_run": False})
|
|
result = runner.run_checks(run_args, [{"id": "scoped", "argv": [sys.executable, "-c", "print('ok')"],
|
|
"cwd": str(args.root), "inputs": {"repos": ["repo"]}}])
|
|
assert result["status"] == "passed"
|
|
evidence = issues.evidence_record(result["run_id"], args)
|
|
assert evidence["source_state"] == "matches-current"
|
|
assert evidence["source_scope"]["repos"] == ["repo"]
|
|
assert any("recorded repository input scope" in note for note in evidence["coverage_notes"])
|
|
(other / "unrelated.txt").write_text("Outside the recorded scope")
|
|
assert issues.evidence_record(result["run_id"], args)["source_state"] == "matches-current"
|
|
(args.root / "changed.txt").write_text("Inside the recorded scope")
|
|
assert issues.evidence_record(result["run_id"], args)["source_state"] == "historical-source-differs"
|
|
|
|
|
|
def test_passing_aggregate_cannot_hide_skipped_checks(args):
|
|
payload = receipt(args)
|
|
payload["stages"].append({"id": "skipped", "status": "skipped", "exit_code": None})
|
|
with pytest.raises(ValueError, match="inconsistent"):
|
|
issues.validate_receipt(payload, args.workspace_root)
|
|
|
|
|
|
def test_apply_is_append_only_read_back_and_idempotent(args, client):
|
|
args.apply = True
|
|
result = issues.handle_note(args)
|
|
assert result["targets"][0]["status"] == "posted-verified"
|
|
assert len(client.posts) == 1
|
|
again = issues.handle_note(args)
|
|
assert again["targets"][0]["status"] == "existing-verified"
|
|
assert len(client.posts) == 1
|
|
assert {call[0] for call in client.calls} == {"GET", "POST"}
|
|
assert all(call[1].endswith("/comments") for call in client.posts)
|
|
|
|
|
|
def test_complete_pagination_continues_past_short_pages(args, client):
|
|
body = issues.handle_note(args)["targets"][0]["body"]
|
|
client.comments = [{"id": n, "body": "unrelated"} for n in range(1, 6)] + [{"id": 6, "body": body}]
|
|
args.apply = True
|
|
result = issues.handle_note(args)
|
|
assert result["targets"][0]["status"] == "existing-verified"
|
|
assert not client.posts
|
|
assert max(call[3]["page"] for call in client.calls if call[3]) == 4
|
|
|
|
|
|
@pytest.mark.parametrize("collision", ["different-body", "duplicate", "repeating-pagination"])
|
|
def test_collisions_and_incomplete_pagination_refuse_post(args, client, collision):
|
|
body = issues.handle_note(args)["targets"][0]["body"]
|
|
client.comments = [{"id": 1, "body": body}]
|
|
if collision == "different-body":
|
|
client.comments[0]["body"] += "changed"
|
|
elif collision == "duplicate":
|
|
client.comments.append({"id": 2, "body": body})
|
|
else:
|
|
client.repeat_pages = True
|
|
args.apply = True
|
|
assert issues.handle_note(args)["_exit_code"] == 2
|
|
assert not client.posts
|
|
|
|
|
|
def test_timeout_after_commit_reconciles_without_replaying_post(args, client):
|
|
args.apply = True
|
|
client.post_mode = "timeout-after-commit"
|
|
result = issues.handle_note(args)
|
|
assert result["targets"][0]["status"] == "reconciled-verified"
|
|
assert len(client.posts) == 1
|
|
assert "fixture-private-token" not in json.dumps(result)
|
|
|
|
|
|
def test_uncertain_post_requires_explicit_retry_after_reconciliation(args, client):
|
|
args.apply = True
|
|
client.post_mode = "timeout-before-commit"
|
|
assert issues.handle_note(args)["targets"][0]["status"] == "uncertain"
|
|
client.post_mode = "normal"
|
|
assert issues.handle_note(args)["targets"][0]["status"] == "uncertain-retry-required"
|
|
assert len(client.posts) == 1
|
|
args.retry_uncertain = True
|
|
assert issues.handle_note(args)["targets"][0]["status"] == "posted-verified"
|
|
assert len(client.posts) == 2
|
|
journals = list(state_root(args.workspace_root, args.state_dir).glob("issue-notes/*.json"))
|
|
assert journals and all("fixture-private-token" not in path.read_text() for path in journals)
|
|
|
|
|
|
def test_uncertain_readback_is_not_reported_verified_and_later_reconciles(args, client):
|
|
args.apply = True
|
|
client.bad_binding = True
|
|
assert issues.handle_note(args)["targets"][0]["status"] == "uncertain"
|
|
client.bad_binding = False
|
|
assert issues.handle_note(args)["targets"][0]["status"] == "existing-verified"
|
|
assert len(client.posts) == 1
|
|
|
|
|
|
def test_journal_binds_immutable_issue_id(args, client):
|
|
args.apply = True
|
|
assert issues.handle_note(args)["targets"][0]["status"] == "posted-verified"
|
|
client.issue_id += 1
|
|
client.comments.clear()
|
|
args.retry_uncertain = True
|
|
assert issues.handle_note(args)["_exit_code"] == 2
|
|
assert len(client.posts) == 1
|
|
|
|
|
|
def _plan(args, rows):
|
|
args.root = args.issue = None
|
|
args.target_plan = args.workspace_root / "targets.json"
|
|
args.target_plan.write_text(json.dumps({"schema_version": 1, "targets": rows}))
|
|
|
|
|
|
@pytest.mark.parametrize("change", ["wrong-url", "duplicate", "outside", "mixed-origin", "credentials"])
|
|
def test_target_plans_require_exact_unique_workspace_bindings(args, change):
|
|
row = {"root": "repo", "issue": 7, "url": "https://gitea.invalid/team/repo/issues/7"}
|
|
if change == "wrong-url":
|
|
row["url"] = "https://gitea.invalid/team/other/issues/7"
|
|
elif change == "outside":
|
|
row["root"] = "../other"
|
|
elif change == "credentials":
|
|
subprocess.run(["git", "-C", str(args.root), "remote", "set-url", "origin", "https://user:private@gitea.invalid/team/repo.git"], check=True)
|
|
rows = [row, deepcopy(row)] if change == "duplicate" else [row]
|
|
if change == "mixed-origin":
|
|
other = args.workspace_root / "other"
|
|
other.mkdir()
|
|
subprocess.run(["git", "init", "-q", str(other)], check=True)
|
|
subprocess.run(["git", "-C", str(other), "remote", "add", "origin", "http://gitea.invalid/team/other.git"], check=True)
|
|
rows.append({"root": "other", "issue": 8, "url": "http://gitea.invalid/team/other/issues/8"})
|
|
_plan(args, rows)
|
|
with pytest.raises(ValueError):
|
|
issues.handle_note(args)
|
|
|
|
|
|
def test_all_targets_preflight_before_first_post_and_posts_are_serial(args, client, monkeypatch):
|
|
first = client.target.record()
|
|
second = {**first, "issue": 8, "url": first["url"].rsplit("/", 1)[0] + "/8"}
|
|
_plan(args, [first, second])
|
|
second_client = FakeClient(issues.resolve_target(Path(first["root"]), 8, args.workspace_root))
|
|
second_client.bad_issue = True
|
|
monkeypatch.setattr(issues, "make_client", lambda target, token: client if target.issue == 7 else second_client)
|
|
args.apply = True
|
|
assert issues.handle_note(args)["_exit_code"] == 2
|
|
assert not client.posts and not second_client.posts
|
|
second_client.bad_issue = False
|
|
result = issues.handle_note(args)
|
|
assert [row["status"] for row in result["targets"]] == ["posted-verified", "posted-verified"]
|
|
assert len(client.posts) == len(second_client.posts) == 1
|
|
|
|
|
|
def test_credentials_and_response_errors_are_not_returned(args, client):
|
|
args.apply = True
|
|
client.fail = True
|
|
result = issues.handle_note(args)
|
|
assert result["_exit_code"] == 2
|
|
assert "fixture-private-token" not in json.dumps(result)
|
|
args.note_summary = ["fixture-private-token"]
|
|
with pytest.raises(ValueError, match="credential"):
|
|
issues.handle_note(args)
|
|
|
|
|
|
def test_marker_injection_and_symlink_inputs_are_rejected(args):
|
|
args.note_summary = [issues.MARKER_PREFIX + "fake -->"]
|
|
with pytest.raises(ValueError, match="reserved"):
|
|
issues.handle_note(args)
|
|
args.note_summary = ["safe"]
|
|
args.body_file = args.workspace_root / "alias.md"
|
|
args.body_file.symlink_to(args.env_file)
|
|
with pytest.raises(ValueError, match="symlink"):
|
|
issues.handle_note(args)
|
|
|
|
|
|
@pytest.mark.parametrize("remote", ["https://gitea.invalid/team/repo.git?token=secret", "https://gitea.invalid/team/repo.git#other", "https://gitea.invalid:bad/team/repo.git"])
|
|
def test_ambiguous_git_remote_is_not_silently_reinterpreted(args, remote):
|
|
subprocess.run(["git", "-C", str(args.root), "remote", "set-url", "origin", remote], check=True)
|
|
with pytest.raises(ValueError):
|
|
issues.handle_note(args)
|
|
|
|
|
|
def test_structured_inputs_are_merged_as_data_not_executed(args):
|
|
args.note_file = args.workspace_root / "note.json"
|
|
args.note_file.write_text(json.dumps({"summary": ["Recorded earlier"], "next": ["Still pending"], "body": "$(never-execute)"}))
|
|
args.body_file = args.workspace_root / "body.md"
|
|
args.body_file.write_text("`never-run-this-either`")
|
|
body = issues.handle_note(args)["targets"][0]["body"]
|
|
assert "Recorded earlier" in body and "Verified scoped changes" in body
|
|
assert "$(never-execute)" in body and "`never-run-this-either`" in body
|
|
|
|
|
|
def receipt(args):
|
|
return {"schema_version": 1, "run_id": "fixture-run", "workspace_root": str(args.workspace_root),
|
|
"project_file": str(args.project), "source_fingerprint": source_fingerprint(load_project(args.workspace_root, args.project)),
|
|
"status": "passed", "snapshot_verified": True, "generated_at": "2026-09-08T12:00:00Z", "finished_at": "2026-09-08T12:00:01Z",
|
|
"stages": [{"id": "fixture", "status": "passed", "exit_code": 0, "duration_seconds": 1,
|
|
"log_path": "/private/log-not-opened", "argv": ["never-execute-this"]}]}
|
|
|
|
|
|
def test_external_receipts_are_unverified_metadata_with_source_comparison(args):
|
|
path = args.workspace_root / "external.json"
|
|
path.write_text(json.dumps(receipt(args)))
|
|
args.evidence = str(path)
|
|
result = issues.handle_note(args)
|
|
evidence = result["evidence"]
|
|
assert evidence["origin"] == "external-unverified"
|
|
assert evidence["source_state"] == "matches-current"
|
|
assert "argv" not in evidence["stages"][0]
|
|
(args.root / "dirty.txt").write_text("changed")
|
|
assert issues.handle_note(args)["evidence"]["source_state"] == "historical-source-differs"
|
|
|
|
|
|
@pytest.mark.parametrize("mutation", ["foreign", "bad-fingerprint", "false-pass", "duplicate-stage", "unknown-status", "noninteger-exit", "invalid-state-shape", "invalid-stage-state-shape", "bool-schema"])
|
|
def test_invalid_receipt_cannot_supply_evidence(args, mutation):
|
|
payload = receipt(args)
|
|
if mutation == "foreign":
|
|
payload["workspace_root"] = str(args.workspace_root.parent)
|
|
elif mutation == "bad-fingerprint":
|
|
payload["source_fingerprint"] = "claimed-green"
|
|
elif mutation == "false-pass":
|
|
payload["stages"][0]["exit_code"] = 1
|
|
elif mutation == "duplicate-stage":
|
|
payload["stages"] *= 2
|
|
elif mutation == "unknown-status":
|
|
payload["status"] = "complete-review"
|
|
elif mutation == "invalid-state-shape":
|
|
payload["status"] = {}
|
|
elif mutation == "invalid-stage-state-shape":
|
|
payload["stages"][0]["status"] = []
|
|
elif mutation == "bool-schema":
|
|
payload["schema_version"] = True
|
|
else:
|
|
payload["stages"][0]["exit_code"] = False
|
|
with pytest.raises(ValueError):
|
|
issues.validate_receipt(payload, args.workspace_root)
|
|
|
|
|
|
def test_local_receipt_integrity_is_checked_but_not_an_attestation(args):
|
|
payload = receipt(args)
|
|
payload["integrity_sha256"] = digest(payload)
|
|
path = state_root(args.workspace_root, args.state_dir) / "runs/fixture-run/receipt.json"
|
|
atomic_json(path, payload)
|
|
args.evidence = "fixture-run"
|
|
result = issues.handle_note(args)
|
|
assert result["evidence"]["origin"] == "local-integrity-checked"
|
|
assert "not an independent attestation" in result["evidence"]["attestation"]
|
|
payload["status"] = "failed"
|
|
atomic_json(path, payload)
|
|
with pytest.raises(ValueError, match="integrity"):
|
|
issues.handle_note(args)
|
|
|
|
|
|
def test_timed_out_stage_is_reportable_without_claiming_success(args):
|
|
payload = receipt(args)
|
|
payload["status"] = "failed"
|
|
payload["stages"][0].update(status="timed_out", exit_code=-15)
|
|
assert issues.validate_receipt(payload, args.workspace_root)["status"] == "failed"
|
|
|
|
|
|
def test_scoped_coverage_limits_are_preserved_in_evidence_and_rendered_note(args, monkeypatch):
|
|
payload = receipt(args)
|
|
monkeypatch.setenv("FIXTURE_SECRET", "fixture-secret-value")
|
|
payload["stages"][0]["coverage_notes"] = ["Compiler/chained suite not run.", "Manual <script>not executed</script> fixture-secret-value"]
|
|
path = args.workspace_root / "coverage.json"
|
|
path.write_text(json.dumps(payload))
|
|
args.evidence = str(path)
|
|
result = issues.handle_note(args)
|
|
evidence = result["evidence"]
|
|
assert evidence["coverage_notes"] == evidence["stages"][0]["coverage_notes"]
|
|
assert "Compiler/chained suite not run." in result["targets"][0]["body"]
|
|
assert "omitted checks ran" in result["targets"][0]["body"]
|
|
assert "<script>" not in result["targets"][0]["body"]
|
|
assert "fixture-secret-value" not in json.dumps(result)
|
|
assert any("coverage limitation" in line for line in result["summary"])
|
|
|
|
|
|
@pytest.mark.parametrize("notes", ["not-a-list", [None], [""], ["x" * 4097], ["limit"] * 2049, [issues.MARKER_PREFIX + "injection"]])
|
|
def test_malformed_coverage_metadata_is_rejected(args, notes):
|
|
payload = receipt(args)
|
|
payload["stages"][0]["coverage_notes"] = notes
|
|
with pytest.raises(ValueError):
|
|
issues.validate_receipt(payload, args.workspace_root)
|
|
|
|
|
|
@pytest.mark.parametrize("snapshot_verified", [None, False, "true", 1])
|
|
def test_external_pass_requires_verified_snapshot_flag(args, snapshot_verified):
|
|
payload = receipt(args)
|
|
payload["snapshot_verified"] = snapshot_verified
|
|
with pytest.raises(ValueError, match="verified source snapshot"):
|
|
issues.validate_receipt(payload, args.workspace_root)
|