from __future__ import annotations from concurrent.futures import ThreadPoolExecutor from itertools import count import json import os from pathlib import Path import re import shutil import subprocess import sys import tempfile import threading import unittest from unittest.mock import patch from fastapi.testclient import TestClient from filelock import FileLock META_ROOT = Path(__file__).resolve().parents[1] RELEASE_ROOT = META_ROOT / "tools" / "release" if str(RELEASE_ROOT) not in sys.path: sys.path.insert(0, str(RELEASE_ROOT)) from govoplan_release.release_run import ( # noqa: E402 MAX_EVENTS, MAX_RECORD_BYTES, ReleaseRunConflict, ReleaseRunCorrupt, ReleaseRunNotFound, ReleaseRunStore, _record_digest, ) from govoplan_release.candidate_artifact import ( # noqa: E402 CandidateArtifactReceipt, candidate_output_path, harden_private_candidate_tree, issue_candidate_receipt, ) from server.app import ( # noqa: E402 create_app, default_release_run_root, release_workspace_fingerprint, ) WORKSPACE_FINGERPRINT = "a" * 64 CREATE_REQUESTS = count(1) class ReleaseRunStoreTests(unittest.TestCase): def setUp(self) -> None: self.temporary = tempfile.TemporaryDirectory() self.root = Path(self.temporary.name) / "release-runs" self.store = ReleaseRunStore( self.root, expected_workspace_fingerprint=WORKSPACE_FINGERPRINT, ) self.runtime_trust = patch( "server.app.require_trusted_release_runtime" ) self.runtime_binding = patch( "server.app.verify_release_runtime_binding" ) self.source_bindings = patch( "server.app.bind_plan_source_states", side_effect=bind_test_plan_source_states, ) self.runtime_trust.start() self.runtime_binding.start() self.source_bindings.start() def tearDown(self) -> None: self.source_bindings.stop() self.runtime_binding.stop() self.runtime_trust.stop() self.temporary.cleanup() def test_create_persists_private_immutable_plan_and_ordered_steps(self) -> None: run = create_run( self.store, input_snapshot=run_input(), plan_snapshot=release_plan() ) reopened = ReleaseRunStore( self.root, expected_workspace_fingerprint=WORKSPACE_FINGERPRINT, ).get(run["run_id"]) record_path = self.root / f"{run['run_id']}.json" self.assertEqual("govoplan.release-run", reopened["schema"]) self.assertEqual(4, reopened["schema_version"]) self.assertEqual( {"govoplan-core": "1.2.3", "govoplan-files": "1.2.4"}, reopened["immutable"]["input"]["repo_versions"], ) self.assertEqual("planned", reopened["state"]["status"]) self.assertTrue(reopened["state"]["steps"][0]["available"]) self.assertFalse(reopened["state"]["steps"][1]["available"]) self.assertEqual("core:preflight", reopened["recommended_next"]["step_id"]) self.assertEqual(0o700, os.stat(self.root).st_mode & 0o777) self.assertEqual(0o600, os.stat(record_path).st_mode & 0o777) self.assertNotIn("processed_requests", reopened["state"]) self.assertNotIn("attempt_fingerprint", reopened["state"]["steps"][0]) def test_catalog_receipt_is_bounded_persisted_and_idempotent(self) -> None: run_id = create_run( self.store, plan_snapshot=mutating_first_plan(), request_id="receipt-create-request-0001", )["run_id"] attempt_id = "receipt-attempt-request-0001" self.store.start_step(run_id, "core:tag", attempt_id=attempt_id) receipt = { "kind": "catalog_candidate", "candidate_id": "candidate-" + "a" * 32, "catalog_sha256": "b" * 64, } finished = self.store.finish_step( run_id, "core:tag", attempt_id=attempt_id, succeeded=True, result_code="catalog_candidate_ready", result_receipt=receipt, ) replayed = self.store.finish_step( run_id, "core:tag", attempt_id=attempt_id, succeeded=True, result_code="catalog_candidate_ready", result_receipt=receipt, ) self.assertEqual(receipt, finished["state"]["steps"][0]["result_receipt"]) self.assertEqual(finished, replayed) with self.assertRaisesRegex(ReleaseRunConflict, "different recorded outcome"): self.store.finish_step( run_id, "core:tag", attempt_id=attempt_id, succeeded=True, result_code="catalog_candidate_ready", result_receipt=receipt | {"catalog_sha256": "c" * 64}, ) def test_catalog_publication_receipt_binds_candidate_origin_commit_and_tag( self, ) -> None: run_id = create_run( self.store, plan_snapshot=mutating_first_plan(), request_id="publication-receipt-create-0001", )["run_id"] attempt_id = "publication-receipt-attempt-0001" self.store.start_step(run_id, "core:tag", attempt_id=attempt_id) receipt = { "kind": "catalog_publication", "candidate_id": "candidate-" + "a" * 32, "catalog_sha256": "b" * 64, "keyring_sha256": "c" * 64, "publication_commit_sha": "d" * 40, "publication_tag_object_sha": "e" * 40, "publication_tag_commit_sha": "d" * 40, "branch": "main", "tag_name": "catalog-stable-1", "remote": "origin", "remote_sha256": "f" * 64, } finished = self.store.finish_step( run_id, "core:tag", attempt_id=attempt_id, succeeded=True, result_code="catalog_published", result_receipt=receipt, ) self.assertEqual(receipt, finished["state"]["steps"][0]["result_receipt"]) with self.assertRaisesRegex( ReleaseRunConflict, "publication receipt identity" ): self.store.finish_step( run_id, "core:tag", attempt_id=attempt_id, succeeded=True, result_code="catalog_published", result_receipt=receipt | {"publication_tag_commit_sha": "0" * 40}, ) def test_schema_three_record_is_verified_then_migrated_privately(self) -> None: run_id = create_run( self.store, request_id="legacy-schema-create-request-0001", )["run_id"] path = self.root / f"{run_id}.json" payload = json.loads(path.read_text(encoding="utf-8")) payload["schema_version"] = 3 for step in payload["state"]["steps"]: step.pop("result_receipt") payload["record_digest"] = _record_digest(payload) path.write_text(json.dumps(payload), encoding="utf-8") path.chmod(0o600) migrated = self.store.get(run_id) persisted = json.loads(path.read_text(encoding="utf-8")) self.assertEqual(4, migrated["schema_version"]) self.assertEqual(4, persisted["schema_version"]) self.assertTrue( all(step["result_receipt"] is None for step in persisted["state"]["steps"]) ) def test_zero_step_plan_is_rejected(self) -> None: plan = release_plan() plan["dry_run_steps"] = [] with self.assertRaisesRegex(ReleaseRunConflict, "step collection"): create_run(self.store, input_snapshot=run_input(), plan_snapshot=plan) def test_create_request_is_durable_idempotent_and_input_bound(self) -> None: request_id = "durable-create-request-0001" created = create_run(self.store, request_id=request_id) changed_plan = release_plan() changed_plan["dry_run_steps"][0]["title"] = "Dashboard drift" # type: ignore[index] replayed = create_run( self.store, request_id=request_id, plan_snapshot=changed_plan, ) found = self.store.find_created_run( request_id=request_id, input_snapshot=run_input(), ) self.assertEqual(created["run_id"], replayed["run_id"]) self.assertEqual(created["run_id"], found["run_id"] if found else None) self.assertEqual( "Inspect Core", replayed["immutable"]["plan"]["dry_run_steps"][0]["title"], ) self.assertEqual(1, len(list(self.root.glob("*.json")))) changed_input = run_input() changed_input["public_catalog"] = True with self.assertRaisesRegex(ReleaseRunConflict, "different inputs"): create_run( self.store, request_id=request_id, input_snapshot=changed_input, ) def test_attempt_transitions_are_idempotent_and_strictly_ordered(self) -> None: run_id = create_run( self.store, input_snapshot=run_input(), plan_snapshot=release_plan() )["run_id"] with self.assertRaisesRegex(ReleaseRunConflict, "Complete core:preflight"): self.store.start_step(run_id, "core:tag", attempt_id="attempt-tag-0001") started = self.store.start_step( run_id, "core:preflight", attempt_id="attempt-preflight-0001" ) duplicate_start = self.store.start_step( run_id, "core:preflight", attempt_id="attempt-preflight-0001" ) self.assertEqual(1, started["state"]["steps"][0]["attempt_count"]) self.assertEqual( len(started["state"]["events"]), len(duplicate_start["state"]["events"]), ) finished = self.store.finish_step( run_id, "core:preflight", attempt_id="attempt-preflight-0001", succeeded=True, result_code="preflight_valid", ) duplicate_finish = self.store.finish_step( run_id, "core:preflight", attempt_id="attempt-preflight-0001", succeeded=True, result_code="preflight_valid", ) self.assertEqual("succeeded", finished["state"]["steps"][0]["state"]) self.assertEqual( len(finished["state"]["events"]), len(duplicate_finish["state"]["events"]), ) self.assertTrue(finished["state"]["steps"][1]["available"]) def test_failed_attempt_and_delayed_start_replays_are_idempotent(self) -> None: run_id = create_run( self.store, input_snapshot=run_input(), plan_snapshot=release_plan() )["run_id"] attempt_id = "failed-attempt-replay-0001" self.store.start_step(run_id, "core:preflight", attempt_id=attempt_id) failed = self.store.finish_step( run_id, "core:preflight", attempt_id=attempt_id, succeeded=False, result_code="preflight_failed", ) duplicate = self.store.start_step( run_id, "core:preflight", attempt_id=attempt_id ) self.assertEqual("failed", duplicate["state"]["steps"][0]["state"]) self.assertEqual( len(failed["state"]["events"]), len(duplicate["state"]["events"]) ) retried = self.store.retry_step( run_id, "core:preflight", request_id="failed-retry-request-0001", ) delayed = self.store.start_step( run_id, "core:preflight", attempt_id=attempt_id ) self.assertEqual("pending", delayed["state"]["steps"][0]["state"]) self.assertEqual( len(retried["state"]["events"]), len(delayed["state"]["events"]) ) def test_resume_requires_a_running_attempt_and_only_records_an_effect(self) -> None: run_id = create_run( self.store, input_snapshot=run_input(), plan_snapshot=release_plan() )["run_id"] original_request = "resume-original-request-0001" with self.assertRaisesRegex(ReleaseRunConflict, "only be resumed"): self.store.resume(run_id, request_id=original_request) before_start = self.store.get(run_id) self.store.start_step( run_id, "core:preflight", attempt_id="resume-delayed-attempt-0001", ) resumed = self.store.resume(run_id, request_id=original_request) duplicate = self.store.resume(run_id, request_id=original_request) self.assertEqual("interrupted", duplicate["state"]["steps"][0]["state"]) self.assertEqual( len(resumed["state"]["events"]), len(duplicate["state"]["events"]) ) raw = json.loads( (self.root / f"{run_id}.json").read_text(encoding="utf-8") ) self.assertEqual(1, len(raw["state"]["processed_requests"])) self.assertLess(len(before_start["state"]["events"]), len(resumed["state"]["events"])) def test_attempt_start_reserves_capacity_for_ambiguous_effect_recovery(self) -> None: saturated_run_id = create_run( self.store, input_snapshot=run_input(), plan_snapshot=release_plan() )["run_id"] with patch("govoplan_release.release_run.MAX_COMMANDS", 1): with self.assertRaisesRegex(ReleaseRunConflict, "recovery capacity"): self.store.start_step( saturated_run_id, "core:preflight", attempt_id="capacity-prestart-attempt-0001", ) untouched = self.store.get(saturated_run_id) self.assertEqual("pending", untouched["state"]["steps"][0]["state"]) self.assertEqual(0, untouched["state"]["steps"][0]["attempt_count"]) terminal_run_id = create_run( self.store, input_snapshot=run_input(), plan_snapshot=mutating_first_plan(), )["run_id"] with patch("govoplan_release.release_run.MAX_COMMANDS", 2): self.store.start_step( terminal_run_id, "core:tag", attempt_id="capacity-terminal-attempt" ) self.store.resume( terminal_run_id, request_id="capacity-terminal-resume-0001" ) with self.assertRaisesRegex(ReleaseRunConflict, "safety limit"): self.store.reconcile_step( terminal_run_id, "core:tag", request_id="capacity-unresolved-request-0001", outcome="unresolved", confirmation="RECONCILE", ) completed = self.store.reconcile_step( terminal_run_id, "core:tag", request_id="capacity-terminal-request-0001", outcome="effect_succeeded", confirmation="RECONCILE", ) self.assertEqual("completed", completed["state"]["status"]) def test_attempt_start_reserves_serialized_recovery_capacity(self) -> None: rejected_plan = mutating_first_plan() rejected_plan["notes"] = ["x" * (MAX_RECORD_BYTES - 2_500)] rejected_run_id = create_run( self.store, request_id="byte-capacity-rejected-create-0001", plan_snapshot=rejected_plan, )["run_id"] rejected_path = self.root / f"{rejected_run_id}.json" self.assertGreater(rejected_path.stat().st_size, MAX_RECORD_BYTES - 1_000) with self.assertRaisesRegex( ReleaseRunConflict, "serialized recovery capacity" ): self.store.start_step( rejected_run_id, "core:tag", attempt_id="byte-capacity-rejected-attempt-0001", ) untouched = self.store.get(rejected_run_id) self.assertEqual("pending", untouched["state"]["steps"][0]["state"]) self.assertEqual(0, untouched["state"]["steps"][0]["attempt_count"]) accepted_plan = mutating_first_plan() accepted_plan["notes"] = ["x" * (MAX_RECORD_BYTES - 4_300)] accepted_run_id = create_run( self.store, request_id="byte-capacity-accepted-create-0001", plan_snapshot=accepted_plan, )["run_id"] accepted_path = self.root / f"{accepted_run_id}.json" self.store.start_step( accepted_run_id, "core:tag", attempt_id="byte-capacity-accepted-attempt-0001", ) self.store.resume( accepted_run_id, request_id="byte-capacity-accepted-resume-0001", ) with self.assertRaisesRegex( ReleaseRunConflict, "reserved for a terminal reconciliation" ): self.store.reconcile_step( accepted_run_id, "core:tag", request_id="byte-capacity-unresolved-request-0001", outcome="unresolved", confirmation="RECONCILE", ) completed = self.store.reconcile_step( accepted_run_id, "core:tag", request_id="byte-capacity-terminal-request-0001", outcome="effect_succeeded", confirmation="RECONCILE", ) self.assertEqual("completed", completed["state"]["status"]) self.assertLessEqual(accepted_path.stat().st_size, MAX_RECORD_BYTES) def test_read_only_start_reserves_serialized_resume_and_retry_capacity( self, ) -> None: plan = mutating_first_plan() plan["dry_run_steps"][0]["mutating"] = False # type: ignore[index] plan["notes"] = ["x" * (MAX_RECORD_BYTES - 5_500)] run_id = create_run( self.store, request_id="byte-capacity-read-only-create-0001", plan_snapshot=plan, )["run_id"] self.store.start_step( run_id, "core:tag", attempt_id="byte-capacity-read-only-attempt-0001", ) self.store.resume( run_id, request_id="byte-capacity-read-only-resume-0001", ) retried = self.store.retry_step( run_id, "core:tag", request_id="byte-capacity-read-only-retry-0001", ) self.assertEqual("pending", retried["state"]["steps"][0]["state"]) def test_unresolved_reconciliation_is_bounded_and_preserves_terminal_slot(self) -> None: run_id = create_run( self.store, input_snapshot=run_input(), plan_snapshot=mutating_first_plan(), )["run_id"] with patch("govoplan_release.release_run.MAX_COMMANDS", 3): self.store.start_step( run_id, "core:tag", attempt_id="capacity-unresolved-attempt" ) self.store.resume( run_id, request_id="capacity-unresolved-resume-0001" ) unresolved = self.store.reconcile_step( run_id, "core:tag", request_id="capacity-unresolved-request-0001", outcome="unresolved", confirmation="RECONCILE", ) duplicate = self.store.reconcile_step( run_id, "core:tag", request_id="capacity-unresolved-request-0001", outcome="unresolved", confirmation="RECONCILE", ) self.assertEqual( len(unresolved["state"]["events"]), len(duplicate["state"]["events"]), ) with self.assertRaisesRegex(ReleaseRunConflict, "already recorded"): self.store.reconcile_step( run_id, "core:tag", request_id="capacity-unresolved-request-0002", outcome="unresolved", confirmation="RECONCILE", ) completed = self.store.reconcile_step( run_id, "core:tag", request_id="capacity-final-request-0001", outcome="effect_succeeded", confirmation="RECONCILE", ) self.assertEqual("completed", completed["state"]["status"]) def test_reopen_and_resume_preserve_ambiguous_mutating_effect(self) -> None: run_id = create_run( self.store, input_snapshot=run_input(), plan_snapshot=mutating_first_plan() )["run_id"] self.store.start_step(run_id, "core:tag", attempt_id="attempt-tag-0001") reopened_store = ReleaseRunStore( self.root, expected_workspace_fingerprint=WORKSPACE_FINGERPRINT, ) before_resume = reopened_store.get(run_id) self.assertEqual("running", before_resume["state"]["steps"][0]["state"]) resumed = reopened_store.resume(run_id, request_id="resume-request-0001") duplicate = reopened_store.resume(run_id, request_id="resume-request-0001") self.assertEqual("interrupted", resumed["state"]["steps"][0]["state"]) self.assertEqual("reconcile_step", resumed["recommended_next"]["id"]) self.assertTrue(resumed["recommended_next"]["available"]) self.assertEqual( len(resumed["state"]["events"]), len(duplicate["state"]["events"]) ) with self.assertRaisesRegex(ReleaseRunConflict, "observed effect"): reopened_store.retry_step( run_id, "core:tag", request_id="retry-request-0001", ) with self.assertRaisesRegex(ReleaseRunConflict, "Type RECONCILE"): reopened_store.reconcile_step( run_id, "core:tag", request_id="reconcile-request-0001", outcome="effect_absent", confirmation="", ) retried = reopened_store.reconcile_step( run_id, "core:tag", request_id="reconcile-request-0002", outcome="effect_absent", confirmation="RECONCILE", ) duplicate_retry = reopened_store.reconcile_step( run_id, "core:tag", request_id="reconcile-request-0002", outcome="effect_absent", confirmation="RECONCILE", ) self.assertEqual("pending", retried["state"]["steps"][0]["state"]) self.assertEqual( len(retried["state"]["events"]), len(duplicate_retry["state"]["events"]), ) def test_reconciliation_can_record_unresolved_or_verified_success(self) -> None: run_id = create_run( self.store, input_snapshot=run_input(), plan_snapshot=mutating_first_plan() )["run_id"] self.store.start_step(run_id, "core:tag", attempt_id="reconcile-attempt-0001") self.store.resume(run_id, request_id="reconcile-resume-request-0001") unresolved = self.store.reconcile_step( run_id, "core:tag", request_id="reconcile-outcome-request-0001", outcome="unresolved", confirmation="RECONCILE", ) self.assertEqual("interrupted", unresolved["state"]["steps"][0]["state"]) self.assertEqual( "unresolved", unresolved["state"]["events"][-1]["result_code"] ) with self.assertRaisesRegex(ReleaseRunConflict, "different run-state action"): self.store.reconcile_step( run_id, "core:tag", request_id="reconcile-outcome-request-0001", outcome="effect_succeeded", confirmation="RECONCILE", ) succeeded = self.store.reconcile_step( run_id, "core:tag", request_id="reconcile-outcome-request-0002", outcome="effect_succeeded", confirmation="RECONCILE", ) self.assertEqual("succeeded", succeeded["state"]["steps"][0]["state"]) self.assertEqual( "reconciled_effect_succeeded", succeeded["state"]["steps"][0]["result_code"], ) self.assertEqual("completed", succeeded["state"]["status"]) def test_tampered_record_fails_closed_and_is_not_rewritten(self) -> None: run_id = create_run( self.store, input_snapshot=run_input(), plan_snapshot=release_plan() )["run_id"] path = self.root / f"{run_id}.json" payload = json.loads(path.read_text(encoding="utf-8")) payload["immutable"]["input"]["channel"] = "testing" path.write_text(json.dumps(payload), encoding="utf-8") tampered = path.read_bytes() with self.assertRaises(ReleaseRunCorrupt): self.store.get(run_id) self.assertEqual(tampered, path.read_bytes()) self.assertEqual("unavailable", self.store.list()[0]["status"]) def test_broad_record_mode_and_symlinked_root_fail_closed(self) -> None: run_id = create_run( self.store, input_snapshot=run_input(), plan_snapshot=release_plan() )["run_id"] path = self.root / f"{run_id}.json" path.chmod(0o640) with self.assertRaisesRegex(ReleaseRunCorrupt, "broader than 0600"): self.store.get(run_id) self.assertEqual(0o640, os.stat(path).st_mode & 0o777) target = Path(self.temporary.name) / "target" target.mkdir() linked = Path(self.temporary.name) / "linked" linked.symlink_to(target, target_is_directory=True) with self.assertRaisesRegex(ReleaseRunCorrupt, "symbolic-link"): ReleaseRunStore( linked, expected_workspace_fingerprint=WORKSPACE_FINGERPRINT, ).list() def test_symlinked_lock_and_lock_io_errors_fail_closed(self) -> None: self.store.list() lock_path = self.root / ".release-runs.lock" lock_path.unlink() target = self.root / "lock-target" target.write_text("do not follow", encoding="utf-8") lock_path.symlink_to(target) with self.assertRaisesRegex(ReleaseRunCorrupt, "lock path"): self.store.list() lock_path.unlink() with patch( "govoplan_release.release_run.FileLock.acquire", side_effect=PermissionError("denied"), ): with self.assertRaisesRegex(ReleaseRunCorrupt, "acquired safely"): self.store.list() original_release = FileLock.release def fail_operator_release( lock: FileLock, *args: object, **kwargs: object ) -> None: if kwargs.get("force") is True: original_release(lock, *args, **kwargs) # type: ignore[arg-type] return raise PermissionError("denied") with patch.object( FileLock, "release", autospec=True, side_effect=fail_operator_release, ): with self.assertRaisesRegex(ReleaseRunCorrupt, "released safely"): self.store.list() def test_atomic_write_oserror_is_normalized_and_leaves_no_record(self) -> None: with patch( "govoplan_release.release_run.os.replace", side_effect=PermissionError("denied"), ): with self.assertRaisesRegex(ReleaseRunCorrupt, "persisted safely"): create_run( self.store, request_id="atomic-write-create-request-0001", ) self.assertEqual([], list(self.root.glob("*.json"))) def test_state_specific_fields_and_boolean_counts_fail_closed(self) -> None: run_id = create_run( self.store, input_snapshot=run_input(), plan_snapshot=release_plan() )["run_id"] path = self.root / f"{run_id}.json" payload = json.loads(path.read_text(encoding="utf-8")) payload["state"]["steps"][0]["attempt_count"] = True payload["state"]["steps"][0]["started_at"] = "2026-07-22T00:00:00Z" path.write_text(json.dumps(payload), encoding="utf-8") path.chmod(0o600) with self.assertRaises(ReleaseRunCorrupt): self.store.get(run_id) def test_valid_looking_mutable_state_tampering_fails_record_checksum(self) -> None: run_id = create_run( self.store, input_snapshot=run_input(), plan_snapshot=mutating_first_plan() )["run_id"] self.store.start_step(run_id, "core:tag", attempt_id="checksum-attempt-0001") self.store.finish_step( run_id, "core:tag", attempt_id="checksum-attempt-0001", succeeded=True, result_code="tag_created", ) path = self.root / f"{run_id}.json" payload = json.loads(path.read_text(encoding="utf-8")) payload["state"]["steps"][0]["result_code"] = "tag_published" path.write_text(json.dumps(payload), encoding="utf-8") path.chmod(0o600) tampered = path.read_bytes() with self.assertRaises(ReleaseRunCorrupt): self.store.get(run_id) self.assertEqual(tampered, path.read_bytes()) def test_impossible_step_chronology_fails_even_with_recomputed_checksum(self) -> None: run_id = create_run( self.store, input_snapshot=run_input(), plan_snapshot=release_plan() )["run_id"] self.store.start_step( run_id, "core:preflight", attempt_id="chronology-attempt-0001" ) self.store.finish_step( run_id, "core:preflight", attempt_id="chronology-attempt-0001", succeeded=True, result_code="preflight_valid", ) path = self.root / f"{run_id}.json" payload = json.loads(path.read_text(encoding="utf-8")) payload["state"]["steps"][0]["started_at"] = "2026-07-22T20:00:00Z" payload["state"]["steps"][0]["finished_at"] = "2026-07-22T19:00:00Z" payload["record_digest"] = _record_digest(payload) path.write_text(json.dumps(payload), encoding="utf-8") path.chmod(0o600) with self.assertRaises(ReleaseRunCorrupt): self.store.get(run_id) def test_record_and_retained_event_timestamps_are_strictly_ordered(self) -> None: run_id = create_run( self.store, input_snapshot=run_input(), plan_snapshot=release_plan() )["run_id"] self.store.start_step( run_id, "core:preflight", attempt_id="timestamp-attempt-0001" ) self.store.resume(run_id, request_id="timestamp-resume-request-0001") path = self.root / f"{run_id}.json" payload = json.loads(path.read_text(encoding="utf-8")) payload["created_at"] = "2026-07-22T00:00:00Z" payload["updated_at"] = "2026-07-22T00:00:03Z" payload["state"]["events"][0]["at"] = "2026-07-22T00:00:02Z" payload["state"]["events"][1]["at"] = "2026-07-22T00:00:01Z" payload["record_digest"] = _record_digest(payload) path.write_text(json.dumps(payload), encoding="utf-8") path.chmod(0o600) with self.assertRaises(ReleaseRunCorrupt): self.store.get(run_id) def test_attempt_count_must_match_lifetime_attempt_history(self) -> None: run_id = create_run( self.store, input_snapshot=run_input(), plan_snapshot=release_plan() )["run_id"] self.store.start_step( run_id, "core:preflight", attempt_id="counted-attempt-0001" ) path = self.root / f"{run_id}.json" payload = json.loads(path.read_text(encoding="utf-8")) payload["state"]["steps"][0]["attempt_count"] = 2 payload["record_digest"] = _record_digest(payload) path.write_text(json.dumps(payload), encoding="utf-8") path.chmod(0o600) with self.assertRaises(ReleaseRunCorrupt): self.store.get(run_id) def test_workspace_fingerprint_is_enforced_for_every_store_operation(self) -> None: run_id = create_run( self.store, input_snapshot=run_input(), plan_snapshot=release_plan() )["run_id"] foreign = ReleaseRunStore( self.root, expected_workspace_fingerprint="b" * 64, ) self.assertEqual([], foreign.list()) with self.assertRaises(ReleaseRunNotFound): foreign.get(run_id) with self.assertRaises(ReleaseRunNotFound): foreign.resume(run_id, request_id="foreign-resume-request-0001") with self.assertRaisesRegex(ReleaseRunConflict, "different workspace"): create_run( foreign, input_snapshot=run_input(), plan_snapshot=release_plan(), ) def test_list_limit_is_applied_after_foreign_workspace_records_are_skipped( self, ) -> None: with patch( "govoplan_release.release_run._timestamp", return_value="2026-07-22T00:00:00Z", ): local_run = create_run( self.store, input_snapshot=run_input(), plan_snapshot=release_plan() ) foreign_store = ReleaseRunStore( self.root, expected_workspace_fingerprint="b" * 64, ) foreign_input = run_input() foreign_input["workspace_fingerprint"] = "b" * 64 with patch( "govoplan_release.release_run._timestamp", return_value="2026-07-22T00:00:01Z", ): create_run( foreign_store, input_snapshot=foreign_input, plan_snapshot=release_plan() ) listed = self.store.list(limit=1) self.assertEqual([local_run["run_id"]], [item["run_id"] for item in listed]) def test_list_orders_verified_runs_by_creation_timestamp_before_unavailable( self, ) -> None: with patch( "govoplan_release.release_run._timestamp", return_value="2026-07-22T00:00:00Z", ): older = create_run( self.store, request_id="timestamp-order-request-older-0001", ) with patch( "govoplan_release.release_run._timestamp", return_value="2026-07-22T01:00:00Z", ): newer = create_run( self.store, request_id="timestamp-order-request-newer-0001", ) with patch( "govoplan_release.release_run._timestamp", return_value="2026-07-22T02:00:00Z", ): self.store.start_step( older["run_id"], "core:preflight", attempt_id="timestamp-order-attempt-0001", ) corrupt = create_run( self.store, request_id="timestamp-order-request-corrupt-0001", ) corrupt_path = self.root / f"{corrupt['run_id']}.json" corrupt_path.write_text("{}", encoding="utf-8") corrupt_path.chmod(0o600) listed = self.store.list(limit=3) self.assertEqual( [newer["run_id"], older["run_id"], corrupt["run_id"]], [item["run_id"] for item in listed], ) self.assertEqual("unavailable", listed[-1]["status"]) self.assertEqual(newer["run_id"], self.store.list(limit=1)[0]["run_id"]) def test_cursor_pagination_is_stable_and_keeps_unavailable_records_visible( self, ) -> None: created: list[dict[str, object]] = [] for index in range(3): with patch( "govoplan_release.release_run._timestamp", return_value=f"2026-07-22T0{index}:00:00Z", ): created.append( create_run( self.store, request_id=f"cursor-page-create-request-{index:04d}", ) ) corrupt_path = self.root / f"{created[0]['run_id']}.json" corrupt_path.write_text("{}", encoding="utf-8") corrupt_path.chmod(0o600) first = self.store.list_page(limit=1) second = self.store.list_page(limit=1, cursor=first["next_cursor"]) third = self.store.list_page(limit=1, cursor=second["next_cursor"]) self.assertEqual(created[2]["run_id"], first["runs"][0]["run_id"]) self.assertEqual(created[1]["run_id"], second["runs"][0]["run_id"]) self.assertEqual("unavailable", third["runs"][0]["status"]) self.assertIsNone(third["next_cursor"]) with self.assertRaisesRegex(ReleaseRunConflict, "cursor is invalid"): self.store.list_page(limit=1, cursor=f"{first['next_cursor']}tampered") def test_cursor_does_not_skip_or_repeat_when_an_older_run_is_updated( self, ) -> None: created: list[dict[str, object]] = [] for index in range(3): with patch( "govoplan_release.release_run._timestamp", return_value=f"2026-07-22T0{index}:00:00Z", ): created.append( create_run( self.store, request_id=f"cursor-update-create-request-{index:04d}", ) ) first = self.store.list_page(limit=1) with patch( "govoplan_release.release_run._timestamp", return_value="2026-07-22T12:00:00Z", ): self.store.start_step( created[0]["run_id"], "core:preflight", attempt_id="cursor-update-attempt-request-0001", ) second = self.store.list_page(limit=1, cursor=first["next_cursor"]) third = self.store.list_page(limit=1, cursor=second["next_cursor"]) self.assertEqual( [item["run_id"] for item in reversed(created)], [ first["runs"][0]["run_id"], second["runs"][0]["run_id"], third["runs"][0]["run_id"], ], ) def test_retention_prunes_only_oldest_completed_records_and_fails_closed( self, ) -> None: with patch("govoplan_release.release_run.MAX_RUN_RECORDS", 3): completed: list[dict[str, object]] = [] for index in range(2): with patch( "govoplan_release.release_run._timestamp", return_value=f"2026-07-22T0{index}:00:00Z", ): run = create_run( self.store, request_id=f"retention-completed-create-{index:04d}", plan_snapshot=mutating_first_plan(), ) attempt_id = f"retention-completed-attempt-{index:04d}" self.store.start_step( run["run_id"], "core:tag", attempt_id=attempt_id ) self.store.finish_step( run["run_id"], "core:tag", attempt_id=attempt_id, succeeded=True, result_code="tag_created", ) completed.append(run) active = create_run( self.store, request_id="retention-active-create-0001", ) replacement = create_run( self.store, request_id="retention-replacement-create-0001", ) self.assertFalse( (self.root / f"{completed[0]['run_id']}.json").exists() ) self.assertTrue((self.root / f"{completed[1]['run_id']}.json").exists()) self.assertTrue((self.root / f"{active['run_id']}.json").exists()) self.assertTrue((self.root / f"{replacement['run_id']}.json").exists()) corrupt_path = self.root / f"{completed[1]['run_id']}.json" corrupt_path.write_text("{}", encoding="utf-8") corrupt_path.chmod(0o600) with self.assertRaisesRegex(ReleaseRunConflict, "retention limit"): create_run( self.store, request_id="retention-refused-create-0001", ) self.assertEqual(3, len(list(self.root.glob("*.json")))) self.assertIn( completed[1]["run_id"], { item["run_id"] for item in self.store.list_page(limit=3)["runs"] if item["status"] == "unavailable" }, ) def test_list_enumeration_oserror_is_normalized(self) -> None: self.store.list() with patch.object( Path, "iterdir", side_effect=PermissionError("denied"), ): with self.assertRaisesRegex(ReleaseRunCorrupt, "enumerated safely"): self.store.list() def test_concurrent_claims_have_one_winner_and_no_lost_update(self) -> None: run_id = create_run( self.store, input_snapshot=run_input(), plan_snapshot=release_plan() )["run_id"] barrier = threading.Barrier(2) def claim(attempt_id: str) -> str: contender = ReleaseRunStore( self.root, expected_workspace_fingerprint=WORKSPACE_FINGERPRINT, ) barrier.wait(timeout=5) try: contender.start_step(run_id, "core:preflight", attempt_id=attempt_id) except ReleaseRunConflict: return "conflict" return "started" with ThreadPoolExecutor(max_workers=2) as executor: results = list( executor.map( claim, ("concurrent-attempt-0001", "concurrent-attempt-0002") ) ) reopened = self.store.get(run_id) self.assertEqual(["conflict", "started"], sorted(results)) self.assertEqual(1, reopened["state"]["steps"][0]["attempt_count"]) self.assertEqual( 1, sum( event["type"] == "step_started" for event in reopened["state"]["events"] ), ) def test_event_history_is_bounded_and_command_history_is_not_evicted(self) -> None: run_id = create_run( self.store, input_snapshot=run_input(), plan_snapshot=release_plan() )["run_id"] for index in range(90): attempt_id = f"bounded-attempt-{index:04d}" self.store.start_step(run_id, "core:preflight", attempt_id=attempt_id) self.store.finish_step( run_id, "core:preflight", attempt_id=attempt_id, succeeded=False, result_code="preflight_failed", ) self.store.retry_step( run_id, "core:preflight", request_id=f"bounded-retry-{index:04d}", ) view = self.store.get(run_id) raw = json.loads((self.root / f"{run_id}.json").read_text(encoding="utf-8")) self.assertEqual(MAX_EVENTS, len(view["state"]["events"])) self.assertEqual(90, len(raw["state"]["processed_requests"])) self.assertEqual(90, len(raw["state"]["processed_attempts"])) self.assertNotIn("bounded-retry-0089", json.dumps(raw)) for event in view["state"]["events"]: self.assertLessEqual( set(event), {"sequence", "at", "type", "step_id", "result_code"} ) class ReleaseRunApiTests(unittest.TestCase): def setUp(self) -> None: self.runtime_trust = patch("server.app.require_trusted_release_runtime") self.runtime_binding = patch("server.app.verify_release_runtime_binding") self.source_bindings = patch( "server.app.bind_plan_source_states", side_effect=bind_test_plan_source_states, ) self.runtime_trust.start() self.runtime_binding.start() self.source_bindings.start() def tearDown(self) -> None: self.source_bindings.stop() self.runtime_binding.stop() self.runtime_trust.stop() def test_api_exposes_and_validates_cursor_pagination(self) -> None: with tempfile.TemporaryDirectory() as temp_dir: app = create_app( workspace_root=Path(temp_dir), run_state_root=Path(temp_dir) / "runs", token="token", ) headers = {"X-Release-Console-Token": "token"} with ( patch("server.app.build_dashboard", return_value=object()), patch( "server.app.build_selective_release_plan", return_value=release_plan(), ), TestClient(app) as client, ): for index in range(2): response = client.post( "/api/release-runs", headers=headers, json=api_run_input( request_id=f"api-page-create-request-{index:04d}" ), ) self.assertEqual(200, response.status_code) first = client.get( "/api/release-runs?limit=1", headers=headers ).json() second = client.get( "/api/release-runs", headers=headers, params={"limit": 1, "cursor": first["next_cursor"]}, ).json() invalid = client.get( "/api/release-runs?cursor=not-a-cursor", headers=headers ) self.assertEqual(1, len(first["runs"])) self.assertEqual(1, len(second["runs"])) self.assertNotEqual( first["runs"][0]["run_id"], second["runs"][0]["run_id"] ) self.assertIsNone(second["next_cursor"]) self.assertEqual(409, invalid.status_code) def test_executor_claims_before_effect_and_replays_lost_success_response( self, ) -> None: with tempfile.TemporaryDirectory() as temp_dir: app = create_app( workspace_root=Path(temp_dir), run_state_root=Path(temp_dir) / "runs", candidate_root=Path(temp_dir) / "candidates", token="token", ) headers = {"X-Release-Console-Token": "token"} with ( patch("server.app.build_dashboard", return_value=object()), patch( "server.app.build_selective_release_plan", return_value=release_plan(), ), patch( "server.app.bind_plan_source_states", side_effect=bind_test_plan_source_states, ), patch( "server.app.verify_repository_preflight_binding", return_value=repository_receipt(), ), patch( "server.app.execute_repository_step", return_value=({"status": "inspected"}, repository_receipt()), ) as executor, TestClient(app) as client, ): created = client.post( "/api/release-runs", headers=headers, json=api_run_input(request_id="api-executor-create-0001"), ).json() run_id = created["run_id"] self.assertTrue(created["state"]["steps"][0]["executor"]["available"]) self.assertEqual( "preflight", created["state"]["steps"][0]["executor"]["kind"] ) payload = {"request_id": "api-executor-attempt-0001"} executed = client.post( f"/api/release-runs/{run_id}/steps/core:preflight/execute", headers=headers, json=payload, ) replayed = client.post( f"/api/release-runs/{run_id}/steps/core:preflight/execute", headers=headers, json=payload, ) self.assertEqual(200, executed.status_code) self.assertEqual("succeeded", executed.json()["state"]["steps"][0]["state"]) self.assertEqual("replayed", replayed.json()["execution_result"]["status"]) executor.assert_called_once() def test_lost_finish_write_interrupts_without_reexecuting_effect(self) -> None: with tempfile.TemporaryDirectory() as temp_dir: app = create_app( workspace_root=Path(temp_dir), run_state_root=Path(temp_dir) / "runs", token="token", ) headers = {"X-Release-Console-Token": "token"} with ( patch("server.app.build_dashboard", return_value=object()), patch( "server.app.build_selective_release_plan", return_value=release_plan(), ), patch( "server.app.bind_plan_source_states", side_effect=bind_test_plan_source_states, ), patch( "server.app.verify_repository_preflight_binding", return_value=repository_receipt(), ), patch( "server.app.execute_repository_step", return_value=({"status": "inspected"}, repository_receipt()), ) as executor, TestClient(app) as client, ): run_id = client.post( "/api/release-runs", headers=headers, json=api_run_input(request_id="api-lost-finish-create-0001"), ).json()["run_id"] with patch.object( app.state.release_runs, "finish_step", side_effect=ReleaseRunCorrupt("simulated durable write failure"), ): first = client.post( f"/api/release-runs/{run_id}/steps/core:preflight/execute", headers=headers, json={"request_id": "api-lost-finish-attempt-0001"}, ) replay = client.post( f"/api/release-runs/{run_id}/steps/core:preflight/execute", headers=headers, json={"request_id": "api-lost-finish-attempt-0001"}, ) loaded = client.get( f"/api/release-runs/{run_id}", headers=headers ).json() self.assertEqual(409, first.status_code) self.assertEqual(409, replay.status_code) self.assertEqual("interrupted", loaded["state"]["steps"][0]["state"]) executor.assert_called_once() def test_executor_exception_is_interrupted_and_never_retried_implicitly(self) -> None: with tempfile.TemporaryDirectory() as temp_dir: app = create_app( workspace_root=Path(temp_dir), run_state_root=Path(temp_dir) / "runs", token="token", ) headers = {"X-Release-Console-Token": "token"} with ( patch("server.app.build_dashboard", return_value=object()), patch( "server.app.build_selective_release_plan", return_value=mutating_first_plan(), ), patch( "server.app.bind_plan_source_states", side_effect=bind_test_plan_source_states, ), patch("server.app.verify_repository_step_precondition"), patch( "server.app.execute_repository_step", side_effect=RuntimeError("connection lost after push"), ) as executor, TestClient(app) as client, ): run_id = client.post( "/api/release-runs", headers=headers, json=api_run_input(request_id="api-ambiguous-create-0001"), ).json()["run_id"] payload = { "request_id": "api-ambiguous-attempt-0001", "confirm": "TAG", } first = client.post( f"/api/release-runs/{run_id}/steps/core:tag/execute", headers=headers, json=payload, ) second = client.post( f"/api/release-runs/{run_id}/steps/core:tag/execute", headers=headers, json=payload, ) loaded = client.get( f"/api/release-runs/{run_id}", headers=headers ).json() self.assertEqual(409, first.status_code) self.assertEqual(409, second.status_code) self.assertEqual("interrupted", loaded["state"]["steps"][0]["state"]) self.assertEqual( "executor_outcome_unknown", loaded["state"]["steps"][0]["result_code"], ) executor.assert_called_once() def test_catalog_execution_persists_and_consumes_only_issued_receipt(self) -> None: with tempfile.TemporaryDirectory() as temp_dir: candidate_root = Path(temp_dir) / "candidates" app = create_app( workspace_root=Path(temp_dir), run_state_root=Path(temp_dir) / "runs", candidate_root=candidate_root, token="token", ) headers = {"X-Release-Console-Token": "token"} def generated(**kwargs: object) -> tuple[dict[str, object], CandidateArtifactReceipt]: candidate_id = str(kwargs["candidate_id"]) candidate = candidate_output_path(candidate_root, candidate_id) channels = candidate / "channels" channels.mkdir(parents=True) (channels / "stable.json").write_text( json.dumps({"channel": "stable", "signatures": [{}]}), encoding="utf-8", ) harden_private_candidate_tree(candidate) return ( {"status": "ready"}, issue_candidate_receipt( root=candidate_root, candidate_id=candidate_id, channel="stable", ), ) with ( patch("server.app.build_dashboard", return_value=object()), patch( "server.app.build_selective_release_plan", return_value=catalog_release_plan(), ), patch( "server.app.bind_plan_source_states", side_effect=bind_test_plan_source_states, ), patch( "server.app.verify_catalog_publication_precondition", return_value=repository_receipt( repo="addideas-govoplan-website", target_tag="" ), ), patch("server.app.default_signing_keys", return_value=("key=/private",)), patch( "server.app.generate_catalog_candidate", side_effect=generated ) as generator, patch( "server.app.publish_received_candidate", side_effect=published_candidate_result, ) as publisher, TestClient(app) as client, ): run_id = client.post( "/api/release-runs", headers=headers, json=api_run_input(request_id="api-candidate-create-0001"), ).json()["run_id"] generated_response = client.post( f"/api/release-runs/{run_id}/steps/catalog:selective-generator/execute", headers=headers, json={ "request_id": "api-candidate-generate-0001", "confirm": "GENERATE", }, ) with patch("server.app.default_signing_keys", return_value=()): replayed_generation = client.post( f"/api/release-runs/{run_id}/steps/catalog:selective-generator/execute", headers=headers, json={ "request_id": "api-candidate-generate-0001", "confirm": "GENERATE", }, ) published = client.post( f"/api/release-runs/{run_id}/steps/catalog:validate-sign-publish/execute", headers=headers, json={ "request_id": "api-candidate-publish-0001", "confirm": "PUSH", }, ) receipt = generated_response.json()["state"]["steps"][0]["result_receipt"] self.assertRegex(receipt["candidate_id"], r"^candidate-[0-9a-f]{32}$") self.assertEqual(64, len(receipt["catalog_sha256"])) self.assertEqual(200, replayed_generation.status_code) self.assertEqual( "replayed", replayed_generation.json()["execution_result"]["status"], ) generator.assert_called_once() self.assertEqual(200, published.status_code) self.assertEqual("completed", published.json()["state"]["status"]) publisher.assert_called_once_with( candidate_path=candidate_root / receipt["candidate_id"], candidate_receipt=receipt, channel="stable", workspace_root=Path(temp_dir), remote="origin", expected_website_receipt=repository_receipt( repo="addideas-govoplan-website", target_tag="" ), ) def test_legacy_catalog_endpoint_is_preview_only(self) -> None: with tempfile.TemporaryDirectory() as temp_dir: app = create_app(workspace_root=Path(temp_dir), token="token") with TestClient(app) as client: response = client.post( "/api/catalog-candidates/publish", headers={"X-Release-Console-Token": "token"}, json={ "candidate_dir": "/tmp/untrusted", "apply": True, "push": True, "confirm": "PUSH", }, ) self.assertEqual(409, response.status_code) self.assertIn("durable release run", response.json()["detail"]) def test_release_channels_and_durable_remote_are_strictly_validated(self) -> None: with tempfile.TemporaryDirectory() as temp_dir: app = create_app(workspace_root=Path(temp_dir), token="token") headers = {"X-Release-Console-Token": "token"} with ( patch("server.app.build_dashboard") as dashboard, TestClient(app) as client, ): query = client.get( "/api/dashboard", headers=headers, params={"channel": "../stable"} ) created = client.post( "/api/release-runs", headers=headers, json=api_run_input( request_id="api-invalid-channel-create-0001", channel="/absolute", ), ) execute = client.post( "/api/release-runs/not-a-run/steps/core:tag/execute", headers=headers, json={ "request_id": "api-invalid-remote-attempt-0001", "confirm": "TAG", "remote": "upstream", }, ) self.assertEqual(422, query.status_code) self.assertEqual(422, created.status_code) self.assertEqual(422, execute.status_code) dashboard.assert_not_called() def test_token_guarded_create_list_read_resume_and_retry(self) -> None: with tempfile.TemporaryDirectory() as temp_dir: root = Path(temp_dir) / "runs" app = create_app( workspace_root=Path(temp_dir), run_state_root=root, token="release-token", ) headers = {"X-Release-Console-Token": "release-token"} with ( patch("server.app.build_dashboard", return_value=object()), patch( "server.app.build_selective_release_plan", return_value=release_plan(), ), TestClient(app) as client, ): unauthorized = client.get("/api/release-runs") created_response = client.post( "/api/release-runs", headers=headers, json=api_run_input(request_id="api-create-request-0001"), ) self.assertEqual(200, created_response.status_code) created = created_response.json() run_id = created["run_id"] listed = client.get("/api/release-runs", headers=headers).json() loaded = client.get( f"/api/release-runs/{run_id}", headers=headers ).json() app.state.release_runs.start_step( run_id, "core:preflight", attempt_id="api-attempt-0001", ) resumed = client.post( f"/api/release-runs/{run_id}/resume", headers=headers, json={"request_id": "api-resume-0001"}, ) retried = client.post( f"/api/release-runs/{run_id}/steps/core:preflight/retry", headers=headers, json={"request_id": "api-retry-0001"}, ) self.assertEqual(401, unauthorized.status_code) self.assertEqual(run_id, listed["runs"][0]["run_id"]) self.assertEqual(run_id, loaded["run_id"]) self.assertEqual( "interrupted", resumed.json()["state"]["steps"][0]["state"] ) self.assertEqual(200, retried.status_code) self.assertEqual("pending", retried.json()["state"]["steps"][0]["state"]) def test_create_requires_request_id_and_replays_before_plan_rebuild(self) -> None: with tempfile.TemporaryDirectory() as temp_dir: app = create_app( workspace_root=Path(temp_dir), run_state_root=Path(temp_dir) / "runs", token="token", ) headers = {"X-Release-Console-Token": "token"} dashboard = patch("server.app.build_dashboard", return_value=object()) planner = patch( "server.app.build_selective_release_plan", return_value=release_plan(), ) with dashboard as build_dashboard, planner as build_plan, TestClient( app ) as client: missing = client.post( "/api/release-runs", headers=headers, json={ key: value for key, value in api_run_input().items() if key != "request_id" }, ) payload = api_run_input(request_id="api-idempotent-create-0001") created = client.post( "/api/release-runs", headers=headers, json=payload ) self.assertEqual(200, created.status_code) build_dashboard.reset_mock() build_plan.reset_mock() replayed = client.post( "/api/release-runs", headers=headers, json=payload ) mismatched = client.post( "/api/release-runs", headers=headers, json=payload | {"public_catalog": True}, ) self.assertEqual(422, missing.status_code) self.assertEqual(200, replayed.status_code) self.assertEqual(created.json()["run_id"], replayed.json()["run_id"]) self.assertEqual(409, mismatched.status_code) self.assertIn("different inputs", mismatched.json()["detail"]) build_dashboard.assert_not_called() build_plan.assert_not_called() self.assertEqual( 1, len(list(app.state.release_runs.root.glob("*.json"))) ) def test_create_rejects_unknown_or_unresolved_repository(self) -> None: with tempfile.TemporaryDirectory() as temp_dir: root = Path(temp_dir) / "runs" app = create_app( workspace_root=Path(temp_dir), run_state_root=root, token="token" ) with ( patch("server.app.build_dashboard", return_value=object()), patch( "server.app.build_selective_release_plan", return_value=release_plan(), ), TestClient(app) as client, ): response = client.post( "/api/release-runs", headers={"X-Release-Console-Token": "token"}, json={ "request_id": "api-unknown-create-request-0001", "channel": "stable", "repo_versions": {"govoplan-unknown": "1.0.0"}, "public_catalog": False, }, ) self.assertEqual(409, response.status_code) self.assertIn("did not resolve exactly", response.json()["detail"]) self.assertEqual([], list(root.glob("*.json"))) def test_api_rejects_tampered_record_without_rewriting_it(self) -> None: with tempfile.TemporaryDirectory() as temp_dir: root = Path(temp_dir) / "runs" app = create_app( workspace_root=Path(temp_dir), run_state_root=root, token="token" ) run_id = create_run( app.state.release_runs, request_id="api-tamper-create-request-0001", input_snapshot=run_input( workspace_fingerprint=app.state.workspace_fingerprint ), plan_snapshot=release_plan(), )["run_id"] path = app.state.release_runs.root / f"{run_id}.json" path.write_text("{}", encoding="utf-8") malformed = path.read_bytes() with TestClient(app) as client: response = client.get( f"/api/release-runs/{run_id}", headers={"X-Release-Console-Token": "token"}, ) self.assertEqual(409, response.status_code) self.assertEqual(malformed, path.read_bytes()) def test_api_maps_unavailable_run_storage_without_server_exception(self) -> None: with tempfile.TemporaryDirectory() as temp_dir: parent = Path(temp_dir) target = parent / "target" target.mkdir() linked_root = parent / "linked-runs" linked_root.symlink_to(target, target_is_directory=True) app = create_app( workspace_root=parent, run_state_root=linked_root, token="token", ) with TestClient(app) as client: response = client.get( "/api/release-runs", headers={"X-Release-Console-Token": "token"}, ) self.assertEqual(409, response.status_code) self.assertIn("symbolic-link", response.json()["detail"]) def test_api_requires_confirmation_and_records_reconciliation_outcome(self) -> None: with tempfile.TemporaryDirectory() as temp_dir: root = Path(temp_dir) / "runs" app = create_app( workspace_root=Path(temp_dir), run_state_root=root, token="token" ) headers = {"X-Release-Console-Token": "token"} with ( patch("server.app.build_dashboard", return_value=object()), patch( "server.app.build_selective_release_plan", return_value=mutating_first_plan(), ), patch( "server.app.reconciled_repository_receipt", return_value=repository_receipt(), ), TestClient(app) as client, ): created = client.post( "/api/release-runs", headers=headers, json=api_run_input(request_id="api-reconcile-create-0001"), ).json() run_id = created["run_id"] app.state.release_runs.start_step( run_id, "core:tag", attempt_id="api-reconcile-attempt-0001" ) client.post( f"/api/release-runs/{run_id}/resume", headers=headers, json={"request_id": "api-reconcile-resume-0001"}, ) missing_confirmation = client.post( f"/api/release-runs/{run_id}/steps/core:tag/reconcile", headers=headers, json={ "request_id": "api-reconcile-request-0001", "outcome": "effect_succeeded", }, ) reconciled = client.post( f"/api/release-runs/{run_id}/steps/core:tag/reconcile", headers=headers, json={ "request_id": "api-reconcile-request-0002", "outcome": "effect_succeeded", "confirm": "RECONCILE", }, ) self.assertEqual(409, missing_confirmation.status_code) self.assertEqual(200, reconciled.status_code) self.assertEqual("completed", reconciled.json()["state"]["status"]) self.assertEqual( "effect_succeeded", reconciled.json()["state"]["events"][-1]["result_code"], ) def test_default_storage_and_immutable_input_are_workspace_bound(self) -> None: with tempfile.TemporaryDirectory() as temp_dir: parent = Path(temp_dir) state_home = parent / "operator-state" first = parent / "first" second = parent / "second" first.mkdir() second.mkdir() with patch.dict( os.environ, {"XDG_STATE_HOME": str(state_home)}, clear=False ): first_root = default_release_run_root(first) second_root = default_release_run_root(second) default_app = create_app(workspace_root=first, token="token") with TestClient(default_app) as client: default_list = client.get( "/api/release-runs", headers={"X-Release-Console-Token": "token"}, ) self.assertEqual(200, default_list.status_code) self.assertEqual(0o700, first_root.stat().st_mode & 0o777) self.assertNotEqual(first_root, second_root) self.assertEqual("release-runs", first_root.name) self.assertEqual(state_home, first_root.parents[3]) self.assertEqual( f"workspace-{release_workspace_fingerprint(first)}", first_root.parent.name, ) self.assertFalse(first_root.is_relative_to(first)) local_workspace = parent / "local" local_meta = local_workspace / "govoplan" local_meta.mkdir(parents=True) (local_meta / "repositories.json").write_text("{}", encoding="utf-8") with patch.dict( os.environ, {"XDG_STATE_HOME": str(state_home)}, clear=False ): self.assertTrue( default_release_run_root(local_workspace).is_relative_to(state_home) ) explicit_root = parent / "explicit" app = create_app( workspace_root=first, run_state_root=explicit_root, token="token", ) with ( patch("server.app.build_dashboard", return_value=object()), patch( "server.app.build_selective_release_plan", return_value=release_plan(), ), TestClient(app) as client, ): response = client.post( "/api/release-runs", headers={"X-Release-Console-Token": "token"}, json=api_run_input(request_id="api-explicit-create-0001"), ) self.assertEqual(200, response.status_code) self.assertEqual( release_workspace_fingerprint(first), response.json()["immutable"]["input"]["workspace_fingerprint"], ) self.assertEqual( explicit_root.resolve() / f"workspace-{release_workspace_fingerprint(first)}" / "release-runs", app.state.release_runs.root, ) def test_shared_explicit_root_does_not_cross_workspace_boundary(self) -> None: with tempfile.TemporaryDirectory() as temp_dir: parent = Path(temp_dir) root = parent / "shared-runs" first = parent / "first" second = parent / "second" first.mkdir() second.mkdir() first_app = create_app( workspace_root=first, run_state_root=root, token="token" ) headers = {"X-Release-Console-Token": "token"} with ( patch("server.app.build_dashboard", return_value=object()), patch( "server.app.build_selective_release_plan", return_value=release_plan(), ), TestClient(first_app) as client, ): created = client.post( "/api/release-runs", headers=headers, json=api_run_input(request_id="api-shared-create-0001"), ).json() first_record = ( first_app.state.release_runs.root / f"{created['run_id']}.json" ) first_record.write_text("{}", encoding="utf-8") second_app = create_app( workspace_root=second, run_state_root=root, token="token" ) with TestClient(second_app) as client: listed = client.get("/api/release-runs", headers=headers) loaded = client.get( f"/api/release-runs/{created['run_id']}", headers=headers ) self.assertEqual([], listed.json()["runs"]) self.assertEqual(404, loaded.status_code) self.assertNotEqual( first_app.state.release_runs.root, second_app.state.release_runs.root, ) def test_ui_and_runbook_state_tracking_boundary_are_explicit(self) -> None: webui = (RELEASE_ROOT / "webui" / "index.html").read_text(encoding="utf-8") runbook = (META_ROOT / "docs" / "RELEASE_CONSOLE.md").read_text( encoding="utf-8" ) self.assertIn("Durable Run State", webui) self.assertIn("Durable execution.", webui) self.assertIn("Create Run from Selection", webui) self.assertIn("Prepare Retry", webui) self.assertIn("step.retry_available", webui) self.assertIn("Record Reconciliation", webui) self.assertIn("effect_succeeded", webui) self.assertNotIn( 'step.id === "catalog:validate-sign-publish" ? "disabled"', webui, ) self.assertIn("renderReleaseRunStoreUnavailable", webui) self.assertIn('const runsRequest = api("/api/release-runs")', webui) self.assertIn("pendingReleaseRunPayload", webui) self.assertIn("recoverPendingReleaseRun", webui) self.assertIn("recoverPendingRunCommands", webui) self.assertIn("inFlightRunCommandId", webui) self.assertIn("Run saved; list refresh unavailable", webui) self.assertIn("The run record is execution evidence only", runbook) self.assertIn("isolated checkout", runbook) self.assertIn("effect_absent", runbook) self.assertIn("$XDG_STATE_HOME", runbook) self.assertIn("same-directory", runbook) self.assertIn("caller-generated `request_id`", runbook) self.assertIn("reserves the two command-ledger", runbook) def test_entire_release_console_script_is_valid_javascript(self) -> None: node = shutil.which("node") if node is None: self.skipTest("Node.js is required for the Release Console UI check") webui = (RELEASE_ROOT / "webui" / "index.html").read_text(encoding="utf-8") match = re.search(r"", webui, re.DOTALL) self.assertIsNotNone(match, "release console script element is missing") result = subprocess.run( # noqa: S603 [node, "--check", "-"], input=match.group("script") if match is not None else "", check=False, capture_output=True, text=True, ) self.assertEqual(0, result.returncode, result.stderr) def test_ui_reuses_uncertain_ids_and_separates_saved_run_list_failure( self, ) -> None: node = shutil.which("node") if node is None: self.skipTest("Node.js is required for the Release Console UI check") webui = (RELEASE_ROOT / "webui" / "index.html").read_text(encoding="utf-8") function_names = ( "pendingReleaseRunPayload", "readPendingReleaseRun", "clearPendingReleaseRun", "recoverPendingReleaseRun", "submitReleaseRunCreate", "refreshReleaseRunListAfterCreate", "readInFlightRunCommands", "pendingRunCommandRequest", "recoverPendingRunCommands", "deterministicRunCommandError", "inFlightRunCommandId", "clearInFlightRunCommand", "requestId", ) functions = "\n".join( javascript_function(webui, name) for name in function_names ) script = ( """ const assert = (condition, message) => { if (!condition) throw new Error(message); }; const values = new Map(); const sessionStorage = { getItem: (key) => values.has(key) ? values.get(key) : null, setItem: (key, value) => values.set(key, String(value)), removeItem: (key) => values.delete(key), }; let uuidSequence = 0; const window = { crypto: { randomUUID: () => `uuid-${++uuidSequence}` } }; const pendingReleaseRunKey = "test-pending-create"; const pendingRunCommandsKey = "test-pending-commands"; const releaseState = { currentRun: null, runStoreAvailable: true }; const elements = { runStatus: { textContent: "" }, runOutput: { innerHTML: "", insertAdjacentHTML: (_position, html) => { elements.runOutput.innerHTML = html + elements.runOutput.innerHTML; }, }, releaseRun: { value: "" }, }; const pill = (label) => label; const escapeHtml = (value) => String(value); const restoreReleaseRunSelection = () => {}; const renderReleaseRun = (run) => { elements.runOutput.innerHTML = `RUN:${run.run_id}`; }; const renderReleaseRunList = () => {}; let failCreate = true; let failList = true; let failureStatus = null; const postedRequestIds = []; const postedPaths = []; async function postJson(path, payload) { postedPaths.push(path); postedRequestIds.push(payload.request_id); if (failCreate) { const error = new Error(failureStatus ? `server rejected ${failureStatus}` : "uncertain transport failure"); if (failureStatus) error.status = failureStatus; throw error; } return { run_id: "rr-known", immutable: { input: {} }, state: { steps: [] } }; } async function api(_path) { if (failList) throw new Error("list refresh failed"); return { runs: [] }; } """ + functions + """ (async () => { const input = { repo_versions: { "govoplan-files": "1.2.4", "govoplan-core": "1.2.3" }, channel: "stable" }; const first = pendingReleaseRunPayload(input); const same = pendingReleaseRunPayload({ ...input, repo_versions: { "govoplan-core": "1.2.3", "govoplan-files": "1.2.4" } }); assert(first.request_id === same.request_id, "same create input did not reuse its request ID"); const uncertain = await submitReleaseRunCreate(first, { automatic: true }); assert(uncertain === false, "uncertain create unexpectedly succeeded"); assert(readPendingReleaseRun().payload.request_id === first.request_id, "uncertain create ID was cleared"); failCreate = false; const recovered = await recoverPendingReleaseRun(); assert(recovered === true, "pending create did not recover"); assert(postedRequestIds.length === 2 && postedRequestIds.every((value) => value === first.request_id), "create replay changed request ID"); assert(readPendingReleaseRun() === null, "successful replay did not clear pending create"); assert(elements.runOutput.innerHTML.includes("Run saved; list refresh unavailable"), "list failure did not preserve saved state"); assert(!elements.runOutput.innerHTML.includes("Run creation failed"), "list failure was labeled as create failure"); assert(releaseState.currentRun.run_id === "rr-known", "saved run was not retained"); const commandKey = "retry:rr-known:core:preflight"; const commandOne = inFlightRunCommandId(commandKey, "retry"); const commandReplay = inFlightRunCommandId(commandKey, "retry"); assert(commandOne === commandReplay, "uncertain command did not reuse its request ID"); const retryRequest = pendingRunCommandRequest(commandKey, commandOne); assert(retryRequest.path.endsWith("/steps/core%3Apreflight/retry"), "retry recovery path was not executable"); const reconcileRequest = pendingRunCommandRequest("reconcile:rr-known:core:tag:effect_succeeded", "reconcile-request"); assert(reconcileRequest.path.endsWith("/steps/core%3Atag/reconcile"), "reconciliation recovery path was not executable"); assert(reconcileRequest.body.confirm === "RECONCILE" && reconcileRequest.body.outcome === "effect_succeeded", "reconciliation recovery body changed"); const executeKey = "execute:rr-known:catalog:selective-generator:GENERATE"; const executeId = inFlightRunCommandId(executeKey, "execute", { signing_keys: ["release-key=/private/key"] }); assert(!values.get(pendingRunCommandsKey).includes("signing_keys"), "signing key path was persisted in session storage"); const executeRequest = pendingRunCommandRequest(executeKey, executeId); assert(!Object.hasOwn(executeRequest.body, "signing_keys"), "execute replay reconstructed signing material"); values.set(pendingRunCommandsKey, JSON.stringify({ [commandKey]: commandOne, [executeKey]: { request_id: executeId, body: { signing_keys: ["legacy-secret"] } } })); const migratedCommands = readInFlightRunCommands(); assert(migratedCommands[executeKey] === executeId, "legacy pending execute request ID was lost"); assert(!values.get(pendingRunCommandsKey).includes("legacy-secret"), "legacy persisted signing material was not purged"); clearInFlightRunCommand(executeKey, executeId); failCreate = true; const uncertainCommand = await recoverPendingRunCommands(); assert(uncertainCommand === false, "uncertain command recovery unexpectedly succeeded"); assert(readInFlightRunCommands()[commandKey] === commandOne, "uncertain command ID was cleared"); failCreate = false; const recoveredCommand = await recoverPendingRunCommands(); assert(recoveredCommand === true, "pending command did not recover after reload"); assert(readInFlightRunCommands()[commandKey] === undefined, "replayed command ID was not cleared"); assert(postedRequestIds.slice(-2).every((value) => value === commandOne), "command replay changed request ID"); assert(postedPaths.at(-1).endsWith("/steps/core%3Apreflight/retry"), "command replay used the wrong endpoint"); assert(releaseState.currentRun.run_id === "rr-known", "recovered command did not select its run"); const conflictKey = "resume:rr-known"; const conflictId = inFlightRunCommandId(conflictKey, "resume"); failCreate = true; failureStatus = 409; const rejectedCommand = await recoverPendingRunCommands(); assert(rejectedCommand === false, "state-conflict recovery unexpectedly succeeded"); assert(readInFlightRunCommands()[conflictKey] === undefined, "deterministic 409 retained a stale command ID"); assert(elements.runOutput.innerHTML.includes("Saved command was not accepted"), "deterministic conflict was not explained"); failCreate = false; failureStatus = null; const commandAfterSuccess = inFlightRunCommandId(commandKey, "retry"); assert(commandAfterSuccess !== commandOne, "successful command did not clear its request ID"); })().catch((error) => { console.error(error.stack || error.message); process.exit(1); }); """ ) result = subprocess.run( # noqa: S603 [node, "-e", script], check=False, capture_output=True, text=True, ) self.assertEqual(0, result.returncode, result.stderr) def create_run( store: ReleaseRunStore, *, request_id: str | None = None, input_snapshot: dict[str, object] | None = None, plan_snapshot: dict[str, object] | None = None, ) -> dict[str, object]: return store.create( request_id=request_id or f"test-create-request-{next(CREATE_REQUESTS):06d}", input_snapshot=input_snapshot if input_snapshot is not None else run_input(), plan_snapshot=plan_snapshot if plan_snapshot is not None else release_plan(), ) def api_run_input( *, request_id: str | None = None, **overrides: object ) -> dict[str, object]: payload = run_input() payload.pop("workspace_fingerprint") payload["request_id"] = request_id or ( f"api-create-request-{next(CREATE_REQUESTS):06d}" ) payload.update(overrides) return payload def javascript_function(source: str, name: str) -> str: match = re.search( rf"\n\s+(?:async\s+)?function\s+{re.escape(name)}\s*\(", source ) if match is None: raise AssertionError(f"JavaScript function {name} was not found") following = re.search( r"\n\s+(?:async\s+)?function\s+[A-Za-z_$]", source[match.end() :], ) if following is None: raise AssertionError(f"JavaScript function after {name} was not found") end = match.end() + following.start() return source[match.start() : end].strip() def run_input( *, workspace_fingerprint: str = WORKSPACE_FINGERPRINT ) -> dict[str, object]: return { "channel": "stable", "repo_versions": { "govoplan-files": "1.2.4", "govoplan-core": "1.2.3", }, "online": False, "remote_tags": False, "public_catalog": False, "include_migrations": True, "workspace_fingerprint": workspace_fingerprint, } def repository_receipt( *, repo: str = "govoplan-core", target_tag: str = "v1.2.3" ) -> dict[str, object]: return { "kind": "repository_state", "repo": repo, "head": "a" * 40, "branch": "main", "remote": "origin", "remote_sha256": "b" * 64, "worktree_clean": True, "target_tag": target_tag, "tag_object": None, } def published_candidate_result( **kwargs: object, ) -> tuple[dict[str, object], dict[str, object]]: candidate = kwargs["candidate_receipt"] website = kwargs["expected_website_receipt"] if not isinstance(candidate, dict) or not isinstance(website, dict): raise AssertionError("publication test receipts are unavailable") receipt = { "kind": "catalog_publication", "candidate_id": candidate["candidate_id"], "catalog_sha256": candidate["catalog_sha256"], "keyring_sha256": "c" * 64, "publication_commit_sha": "d" * 40, "publication_tag_object_sha": "e" * 40, "publication_tag_commit_sha": "d" * 40, "branch": website["branch"], "tag_name": "catalog-stable-7", "remote": "origin", "remote_sha256": website["remote_sha256"], } return {"status": "published"}, receipt def bind_test_plan_source_states(**kwargs: object) -> dict[str, object]: plan = kwargs["plan"] if not isinstance(plan, dict): raise AssertionError("test plan is not an object") steps = plan.get("dry_run_steps") if not isinstance(steps, list): raise AssertionError("test plan has no steps") for step in steps: if not isinstance(step, dict): continue if isinstance(step.get("repo"), str): step["source_binding"] = repository_receipt() elif step.get("id") == "catalog:validate-sign-publish": step["source_binding"] = repository_receipt( repo="addideas-govoplan-website", target_tag="" ) return plan def release_plan() -> dict[str, object]: return { "generated_at": "2026-07-22T00:00:00Z", "target_channel": "stable", "status": "attention", "units": [ {"repo": "govoplan-core", "target_version": "1.2.3"}, {"repo": "govoplan-files", "target_version": "1.2.4"}, ], "compatibility": [], "gate_findings": [], "recommended_action": { "id": "preview_source_release", "title": "Preview source release tags", "detail": "Plan-visible gates pass.", "remediation": "Preview before mutation.", }, "source_preflight_ready": True, "dry_run_steps": [ { "id": "core:preflight", "title": "Inspect Core", "detail": "Run a read-only preflight.", "command": "git status --short --branch", "cwd": "/workspace/govoplan-core", "mutating": False, "repo": "govoplan-core", "status": "planned", }, { "id": "core:tag", "title": "Create Core tag", "detail": "Create an annotated tag after confirmation.", "command": "git tag -a v1.2.3", "cwd": "/workspace/govoplan-core", "mutating": True, "repo": "govoplan-core", "status": "planned", }, ], "notes": [], } def mutating_first_plan() -> dict[str, object]: plan = release_plan() plan["dry_run_steps"] = [plan["dry_run_steps"][1]] # type: ignore[index] return plan def catalog_release_plan() -> dict[str, object]: plan = release_plan() plan["dry_run_steps"] = [ { "id": "catalog:selective-generator", "title": "Generate candidate", "detail": "Build artifacts and sign a candidate.", "command": "release-catalog selective", "cwd": "/workspace/govoplan", "mutating": True, "repo": None, "status": "planned", }, { "id": "catalog:validate-sign-publish", "title": "Publish candidate", "detail": "Publish only the receipt-bound candidate.", "command": "release-catalog publish-candidate", "cwd": "/workspace/govoplan", "mutating": True, "repo": None, "status": "planned", }, ] return plan if __name__ == "__main__": unittest.main()