sysadminandClaude Opus 4.8
4a2fae8495
Reviewer REQUEST_CHANGES on PR #903 at head1c88b87raised five blockers, all reproduced by executing that head. The shared shape: a write path that declared itself gated, audited, and verified, but never armed the gate, mutated a copy of the state it claimed to fix, and then verified against that same copy. B1 - the apply path never asked the execution gate. execute_recovery_playbook called console_authz.authorize with the default for_execution=False, and the phase branch only fires when it is True. ACTIVE_PHASE is 1 and every new action is phase 2, so an operator executed a phase-2 write through POST /api/v1/system/recovery/apply while build_recovery_preview reported execution_enabled false. The call now passes for_execution=True and surfaces the phase_not_active refusal. Preview reports the same decision under execution_authorization / execution_blocked_reason instead of a hardcoded False it could not explain. B2 - both env playbooks mutated a discarded copy and verified against it. source_env = dict(os.environ) meant clear_stale_binding and rebind_session_worktree never touched the running process, and verify_post_recovery(env=source_env) re-diagnosed the same copy, confirming a change that had not happened. Mutations now target the live mapping (apply_recovery's sanctioned env=None -> os.environ path, #702 AC2) and verification re-reads state rather than the mutated input. binding_before / binding_after / binding_changed are returned, and a playbook that changed nothing reports performed: false. verify_post_recovery no longer reads an unverified_inherited binding as clean, because unproven is not clean. B3 - the reconcile playbook called a function that does not exist. merged_cleanup_reconcile.reconcile_merged_cleanups is absent from that module and a bare except turned the AttributeError into a generic failure, so the playbook could never succeed. It now calls gitea_mcp_server.gitea_reconcile_merged_cleanups, the real orchestrator, imported lazily; failures carry error_type. task_capability_map declared gitea.pr.close for reconcile_cleanups while the entry point gates on gitea.read; the two authority statements are reconciled to the one that is enforced. B4 - the #630 contamination integration could not block. assess_contamination_gate was fed marker=None, which short-circuits to block: False on its first statement; the task passed was a console action id outside CONTAMINATION_GATED_TASKS; and the result was read through a "contaminated" key the gate never returns, making STATUS_BLOCKED_CONTAMINATION unreachable. The live marker now comes from the #641 session inventory reader, the gated task key console_recovery_apply is added to CONTAMINATION_GATED_TASKS, every read uses the "block" key the gate actually returns, and the marker is forwarded to sanctioned_restart.execute_restart so a restart cannot launder a contaminated runtime. The reconciler cleanup playbook stays exempt as the designated remedy. B5 - the parity baseline was captured from the head it was compared against. capture_startup_parity(root, head=checkout_head) stores the head verbatim, so in_parity was structurally incapable of being false, and live_remote_head was never passed. The baseline is now the daemon start head that assess_stale_runtime already returns, and the #610 live-remote dimension is restored. Also: _recovery_card was the one renderer in system_health_views.py interpolating without _esc(), and it is where a marker's operator-supplied command_summary lands once B4 is wired; it now escapes, including the except branch. Docs no longer claim apply enforces master parity or that verify asserts clean: true, and the absolute file:///Users/... links are relative. Tests: the two that asserted the defects as intended are inverted - test_api_recovery_apply_with_dev_auth asserted the phase-gate bypass, and the rebind test asserted the input echoed back. Added coverage per blocker, including a no-op detection test that fails when a playbook reports success without changing anything, the previously untested reconcile playbook, contamination block and remedy-exemption tests, and parity baseline/live-remote tests. Both new guards were mutation-verified: disarming for_execution fails 2 tests, restoring the env copy fails 2 tests. Validation: WEBUI_TEST_OFFLINE=1 ../../venv/bin/python -m pytest tests/ -q from branches/feat-issue-644 gives 27 failed / 5291 passed / 6 skipped / 953 subtests; the same command from branches/baseline-master-76f293e at76f293eb28gives 28 failed / 5262 passed / 6 skipped / 926 subtests. Suites run one at a time. comm of the sorted FAILED lines shows no new signature at the head. The single absent signature, test_workspace_guard_alignment.py:: TestRuntimeContextGuardAlignment::test_declared_branches_worktree_passes_when_mcp_root_differs, is suite-order dependent: that file passes 9/9 in isolation at both revisions. Closes #644 Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
507 lines
23 KiB
Python
507 lines
23 KiB
Python
"""Unit and integration tests for Phase 2 Web Console recovery controls (#644)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import sys
|
|
import types
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
from starlette.testclient import TestClient
|
|
|
|
import merged_cleanup_reconcile
|
|
import runtime_recovery_guard
|
|
import stable_branch_push_guard
|
|
import stale_binding_recovery
|
|
from webui import console_authz, console_recovery, system_health
|
|
from webui.app import create_app
|
|
|
|
|
|
class TestConsoleRecovery(unittest.TestCase):
|
|
|
|
def test_diagnose_recovery_healthy(self) -> None:
|
|
diag = console_recovery.diagnose_recovery()
|
|
self.assertIn(diag.status, {console_recovery.STATUS_HEALTHY, console_recovery.STATUS_ACTION_REQUIRED})
|
|
self.assertIsInstance(diag.playbooks, tuple)
|
|
self.assertGreaterEqual(len(diag.playbooks), 4)
|
|
|
|
playbook_ids = {pb.playbook_id for pb in diag.playbooks}
|
|
self.assertIn(console_recovery.PLAYBOOK_CLEAR_STALE_BINDING, playbook_ids)
|
|
self.assertIn(console_recovery.PLAYBOOK_REBIND_SESSION, playbook_ids)
|
|
self.assertIn(console_recovery.PLAYBOOK_RECONCILE_CLEANUPS, playbook_ids)
|
|
self.assertIn(console_recovery.PLAYBOOK_SANCTIONED_RESTART, playbook_ids)
|
|
|
|
def test_confirmation_phrase_generation_and_matching(self) -> None:
|
|
phrase = console_recovery.confirmation_phrase("clear_stale_binding")
|
|
self.assertEqual(phrase, "confirm clear_stale_binding")
|
|
self.assertTrue(console_recovery.confirmation_matches("clear_stale_binding", "confirm clear_stale_binding"))
|
|
self.assertFalse(console_recovery.confirmation_matches("clear_stale_binding", "wrong phrase"))
|
|
|
|
phrase_target = console_recovery.confirmation_phrase("sanctioned_restart", "gitea-author")
|
|
self.assertEqual(phrase_target, "confirm sanctioned_restart gitea-author")
|
|
self.assertTrue(console_recovery.confirmation_matches("sanctioned_restart", "confirm sanctioned_restart gitea-author", "gitea-author"))
|
|
|
|
def test_build_recovery_preview(self) -> None:
|
|
principal = console_authz.Principal("[email protected]", console_authz.OPERATOR, console_authz.IDENTITY_LOCAL_DEV, True)
|
|
preview = console_recovery.build_recovery_preview(
|
|
playbook_id=console_recovery.PLAYBOOK_CLEAR_STALE_BINDING,
|
|
target="test-worktree",
|
|
principal=principal,
|
|
)
|
|
self.assertEqual(preview["playbook_id"], console_recovery.PLAYBOOK_CLEAR_STALE_BINDING)
|
|
self.assertEqual(preview["action_id"], console_recovery.ACTION_CLEAR_STALE_BINDING)
|
|
self.assertEqual(preview["confirmation_phrase"], "confirm clear_stale_binding test-worktree")
|
|
self.assertTrue(len(preview["mutation_ledger"]) >= 3)
|
|
self.assertTrue(preview["authorization"]["allowed"])
|
|
|
|
def test_build_recovery_preview_unknown_playbook(self) -> None:
|
|
preview = console_recovery.build_recovery_preview("unknown_playbook")
|
|
self.assertFalse(preview.get("allowed"))
|
|
self.assertEqual(preview.get("error"), "unknown_playbook")
|
|
|
|
def test_execute_recovery_playbook_confirmation_mismatch(self) -> None:
|
|
# Authorization is checked before confirmation, so the phase gate has to
|
|
# pass for this test to reach the branch it is about.
|
|
principal = console_authz.Principal("[email protected]", console_authz.OPERATOR, console_authz.IDENTITY_LOCAL_DEV, True)
|
|
with self._phase_two_enabled():
|
|
result = console_recovery.execute_recovery_playbook(
|
|
playbook_id=console_recovery.PLAYBOOK_CLEAR_STALE_BINDING,
|
|
confirmation="invalid confirmation",
|
|
principal=principal,
|
|
)
|
|
self.assertFalse(result["success"])
|
|
self.assertFalse(result["allowed"])
|
|
self.assertEqual(result["error"], "confirmation_mismatch")
|
|
|
|
def test_execute_recovery_playbook_unauthorized(self) -> None:
|
|
# Anonymous principal has viewer role -> should be denied
|
|
result = console_recovery.execute_recovery_playbook(
|
|
playbook_id=console_recovery.PLAYBOOK_CLEAR_STALE_BINDING,
|
|
confirmation="confirm clear_stale_binding",
|
|
principal=console_authz.ANONYMOUS,
|
|
)
|
|
self.assertFalse(result["success"])
|
|
self.assertFalse(result["allowed"])
|
|
self.assertEqual(result["error"], console_authz.DENY_UNAUTHENTICATED)
|
|
|
|
def test_execute_refuses_phase_two_write_while_console_is_phase_one(self) -> None:
|
|
"""B1: the apply path must arm the phase gate, not skip it.
|
|
|
|
``authorize`` only applies the phase branch when ``for_execution=True``.
|
|
The apply path used the default, so an operator executed a phase-2 write
|
|
while ``ACTIVE_PHASE`` was 1.
|
|
"""
|
|
self.assertGreater(
|
|
console_authz.get_action(console_recovery.ACTION_CLEAR_STALE_BINDING).phase,
|
|
console_authz.ACTIVE_PHASE,
|
|
"fixture assumes the recovery actions are ahead of the active phase",
|
|
)
|
|
principal = console_authz.Principal(
|
|
"[email protected]", console_authz.OPERATOR, console_authz.IDENTITY_LOCAL_DEV, True
|
|
)
|
|
phrase = console_recovery.confirmation_phrase(
|
|
console_recovery.PLAYBOOK_CLEAR_STALE_BINDING
|
|
)
|
|
result = console_recovery.execute_recovery_playbook(
|
|
playbook_id=console_recovery.PLAYBOOK_CLEAR_STALE_BINDING,
|
|
confirmation=phrase,
|
|
principal=principal,
|
|
)
|
|
self.assertFalse(result["success"])
|
|
self.assertFalse(result["allowed"])
|
|
self.assertEqual(result["error"], console_authz.DENY_PHASE_NOT_ACTIVE)
|
|
|
|
def test_preview_execution_enabled_matches_the_execution_decision(self) -> None:
|
|
"""B1: preview must not report a bare False it cannot explain."""
|
|
principal = console_authz.Principal(
|
|
"[email protected]", console_authz.OPERATOR, console_authz.IDENTITY_LOCAL_DEV, True
|
|
)
|
|
preview = console_recovery.build_recovery_preview(
|
|
playbook_id=console_recovery.PLAYBOOK_REBIND_SESSION,
|
|
target="branches/feat-issue-644",
|
|
principal=principal,
|
|
)
|
|
self.assertFalse(preview["execution_enabled"])
|
|
self.assertEqual(
|
|
preview["execution_blocked_reason"], console_authz.DENY_PHASE_NOT_ACTIVE
|
|
)
|
|
self.assertFalse(preview["execution_authorization"]["allowed"])
|
|
# The preview (non-execution) decision still allows, by role.
|
|
self.assertTrue(preview["authorization"]["allowed"])
|
|
|
|
def _phase_two_enabled(self):
|
|
"""Raise ACTIVE_PHASE so the execution branches are reachable in tests."""
|
|
return patch.object(console_authz, "ACTIVE_PHASE", 2)
|
|
|
|
def _operator(self) -> console_authz.Principal:
|
|
return console_authz.Principal(
|
|
"[email protected]", console_authz.OPERATOR, console_authz.IDENTITY_LOCAL_DEV, True
|
|
)
|
|
|
|
def test_rebind_mutates_the_live_environment_not_a_copy(self) -> None:
|
|
"""B2: the playbook must change the mapping it claims to have changed."""
|
|
live_env = {stale_binding_recovery.ACTIVE_WORKTREE_ENV: "branches/stale-old"}
|
|
phrase = console_recovery.confirmation_phrase(
|
|
console_recovery.PLAYBOOK_REBIND_SESSION, "branches/feat-issue-644"
|
|
)
|
|
with self._phase_two_enabled():
|
|
result = console_recovery.execute_recovery_playbook(
|
|
playbook_id=console_recovery.PLAYBOOK_REBIND_SESSION,
|
|
confirmation=phrase,
|
|
target="branches/feat-issue-644",
|
|
principal=self._operator(),
|
|
env=live_env,
|
|
)
|
|
self.assertTrue(result["success"])
|
|
self.assertEqual(
|
|
live_env[stale_binding_recovery.ACTIVE_WORKTREE_ENV],
|
|
"branches/feat-issue-644",
|
|
"rebind reported success without changing the caller's environment",
|
|
)
|
|
self.assertTrue(result["applied_result"]["binding_changed"])
|
|
self.assertEqual(result["applied_result"]["binding_before"], "branches/stale-old")
|
|
self.assertEqual(
|
|
result["applied_result"]["binding_after"], "branches/feat-issue-644"
|
|
)
|
|
|
|
def test_clear_stale_binding_reports_failure_when_nothing_changed(self) -> None:
|
|
"""B2: a no-op recovery must never be reported as success."""
|
|
live_env: dict[str, str] = {}
|
|
phrase = console_recovery.confirmation_phrase(
|
|
console_recovery.PLAYBOOK_CLEAR_STALE_BINDING
|
|
)
|
|
with self._phase_two_enabled():
|
|
result = console_recovery.execute_recovery_playbook(
|
|
playbook_id=console_recovery.PLAYBOOK_CLEAR_STALE_BINDING,
|
|
confirmation=phrase,
|
|
principal=self._operator(),
|
|
env=live_env,
|
|
)
|
|
self.assertFalse(
|
|
result["success"],
|
|
"a clear that changed no binding must not report success",
|
|
)
|
|
self.assertFalse(result["applied_result"]["binding_changed"])
|
|
|
|
def test_clear_stale_binding_clears_the_live_binding(self) -> None:
|
|
"""B2: the sanctioned clear must reach the caller's environment."""
|
|
missing = "/nonexistent/branches/deleted-worktree"
|
|
live_env = {stale_binding_recovery.ACTIVE_WORKTREE_ENV: missing}
|
|
phrase = console_recovery.confirmation_phrase(
|
|
console_recovery.PLAYBOOK_CLEAR_STALE_BINDING
|
|
)
|
|
with self._phase_two_enabled():
|
|
result = console_recovery.execute_recovery_playbook(
|
|
playbook_id=console_recovery.PLAYBOOK_CLEAR_STALE_BINDING,
|
|
confirmation=phrase,
|
|
principal=self._operator(),
|
|
env=live_env,
|
|
)
|
|
if result["success"]:
|
|
self.assertNotIn(stale_binding_recovery.ACTIVE_WORKTREE_ENV, live_env)
|
|
self.assertEqual(result["applied_result"]["binding_before"], missing)
|
|
self.assertIsNone(result["applied_result"]["binding_after"])
|
|
else:
|
|
# Fail closed is acceptable; reporting a clear that did not happen
|
|
# is not. This is the invariant the blocker was about.
|
|
self.assertFalse(result["applied_result"]["binding_changed"])
|
|
self.assertEqual(
|
|
live_env.get(stale_binding_recovery.ACTIVE_WORKTREE_ENV), missing
|
|
)
|
|
|
|
def test_reconcile_playbook_calls_an_entry_point_that_exists(self) -> None:
|
|
"""B3: the previous call named a function absent from the module."""
|
|
phrase = console_recovery.confirmation_phrase(
|
|
console_recovery.PLAYBOOK_RECONCILE_CLEANUPS
|
|
)
|
|
fake_server = types.SimpleNamespace(
|
|
gitea_reconcile_merged_cleanups=lambda **kwargs: {
|
|
"success": True,
|
|
"entries": [{"issue_number": 100}],
|
|
}
|
|
)
|
|
with self._phase_two_enabled(), patch.dict(
|
|
sys.modules, {"gitea_mcp_server": fake_server}
|
|
):
|
|
result = console_recovery.execute_recovery_playbook(
|
|
playbook_id=console_recovery.PLAYBOOK_RECONCILE_CLEANUPS,
|
|
confirmation=phrase,
|
|
principal=console_authz.Principal(
|
|
"[email protected]",
|
|
console_authz.ADMIN,
|
|
console_authz.IDENTITY_LOCAL_DEV,
|
|
True,
|
|
),
|
|
)
|
|
self.assertTrue(result["success"], result.get("applied_result"))
|
|
self.assertNotIn("error_type", result["applied_result"])
|
|
self.assertEqual(result["applied_result"]["reconciled_count"], 1)
|
|
|
|
def test_reconcile_entry_point_exists_on_the_real_module(self) -> None:
|
|
"""B3 regression: guard the symbol itself, not just the call shape."""
|
|
import gitea_mcp_server
|
|
|
|
self.assertTrue(
|
|
hasattr(gitea_mcp_server, "gitea_reconcile_merged_cleanups"),
|
|
"console recovery depends on this reconciler entry point",
|
|
)
|
|
self.assertFalse(
|
|
hasattr(merged_cleanup_reconcile, "reconcile_merged_cleanups"),
|
|
"if this module grows the orchestrator, point the playbook back at it",
|
|
)
|
|
|
|
def test_contamination_gate_blocks_a_writing_playbook(self) -> None:
|
|
"""B4: a live marker plus a gated task key must actually block."""
|
|
marker = {
|
|
"kind": "manual_daemon_kill",
|
|
"reason_class": "manual_daemon_kill",
|
|
"command_summary": "pkill -f gitea_mcp_server",
|
|
"active": True,
|
|
}
|
|
phrase = console_recovery.confirmation_phrase(
|
|
console_recovery.PLAYBOOK_REBIND_SESSION, "branches/feat-issue-644"
|
|
)
|
|
live_env = {stale_binding_recovery.ACTIVE_WORKTREE_ENV: "branches/stale-old"}
|
|
with self._phase_two_enabled(), patch.object(
|
|
console_recovery, "load_active_contamination_marker", return_value=marker
|
|
):
|
|
result = console_recovery.execute_recovery_playbook(
|
|
playbook_id=console_recovery.PLAYBOOK_REBIND_SESSION,
|
|
confirmation=phrase,
|
|
target="branches/feat-issue-644",
|
|
principal=self._operator(),
|
|
env=live_env,
|
|
)
|
|
self.assertFalse(result["success"])
|
|
self.assertEqual(result["error"], "contaminated_runtime")
|
|
self.assertEqual(
|
|
live_env[stale_binding_recovery.ACTIVE_WORKTREE_ENV],
|
|
"branches/stale-old",
|
|
"a blocked playbook must not have mutated anything",
|
|
)
|
|
|
|
def test_contamination_gate_exempts_the_reconciler_remedy(self) -> None:
|
|
"""B4: the designated remedy must stay reachable while contaminated."""
|
|
marker = {
|
|
"kind": "manual_daemon_kill",
|
|
"reason_class": "manual_daemon_kill",
|
|
"command_summary": "pkill -f gitea_mcp_server",
|
|
"active": True,
|
|
}
|
|
phrase = console_recovery.confirmation_phrase(
|
|
console_recovery.PLAYBOOK_RECONCILE_CLEANUPS
|
|
)
|
|
fake_server = types.SimpleNamespace(
|
|
gitea_reconcile_merged_cleanups=lambda **kwargs: {
|
|
"success": True,
|
|
"entries": [],
|
|
}
|
|
)
|
|
with self._phase_two_enabled(), patch.object(
|
|
console_recovery, "load_active_contamination_marker", return_value=marker
|
|
), patch.dict(sys.modules, {"gitea_mcp_server": fake_server}):
|
|
result = console_recovery.execute_recovery_playbook(
|
|
playbook_id=console_recovery.PLAYBOOK_RECONCILE_CLEANUPS,
|
|
confirmation=phrase,
|
|
principal=console_authz.Principal(
|
|
"[email protected]",
|
|
console_authz.ADMIN,
|
|
console_authz.IDENTITY_LOCAL_DEV,
|
|
True,
|
|
),
|
|
)
|
|
self.assertNotEqual(result.get("error"), "contaminated_runtime")
|
|
|
|
def test_gated_task_key_is_actually_gated(self) -> None:
|
|
"""B4: the console action id was never a member of the gated set."""
|
|
self.assertIn(
|
|
console_recovery.CONTAMINATION_GATED_TASK,
|
|
stable_branch_push_guard.CONTAMINATION_GATED_TASKS,
|
|
)
|
|
self.assertNotIn(
|
|
console_recovery.ACTION_CLEAR_STALE_BINDING,
|
|
stable_branch_push_guard.CONTAMINATION_GATED_TASKS,
|
|
)
|
|
|
|
def test_diagnosis_reads_the_key_the_gate_returns(self) -> None:
|
|
"""B4: ``contaminated`` is a key assess_contamination_gate never returns."""
|
|
gate = runtime_recovery_guard.assess_contamination_gate(
|
|
None, task=console_recovery.CONTAMINATION_GATED_TASK, actual_role="operator"
|
|
)
|
|
self.assertNotIn("contaminated", gate)
|
|
self.assertIn("block", gate)
|
|
|
|
def test_contaminated_runtime_is_reported_unclean(self) -> None:
|
|
"""B4: verify_post_recovery reported contamination_clean unconditionally."""
|
|
marker = {
|
|
"kind": "manual_daemon_kill",
|
|
"reason_class": "manual_daemon_kill",
|
|
"command_summary": "pkill -f gitea_mcp_server",
|
|
"active": True,
|
|
}
|
|
with patch.object(
|
|
console_recovery, "load_active_contamination_marker", return_value=marker
|
|
):
|
|
verification = console_recovery.verify_post_recovery()
|
|
diag = console_recovery.diagnose_recovery()
|
|
self.assertFalse(verification["contamination_clean"])
|
|
self.assertFalse(verification["clean"])
|
|
self.assertEqual(diag.status, console_recovery.STATUS_BLOCKED_CONTAMINATION)
|
|
|
|
def test_master_parity_baseline_is_not_the_head_it_is_compared_against(self) -> None:
|
|
"""B5: capture_startup_parity was fed the head it was then compared to."""
|
|
stale = system_health.StaleRuntime(
|
|
daemon_head="a" * 40,
|
|
checkout_head="b" * 40,
|
|
remote_head="b" * 40,
|
|
stale=True,
|
|
determinable=True,
|
|
mutation_safe=False,
|
|
reasons=("daemon is behind the checkout",),
|
|
)
|
|
with patch.object(system_health, "assess_stale_runtime", return_value=stale):
|
|
diag = console_recovery.diagnose_recovery()
|
|
parity = diag.master_parity
|
|
self.assertEqual(parity["startup_head"], "a" * 40)
|
|
self.assertEqual(parity["current_head"], "b" * 40)
|
|
self.assertNotEqual(parity["startup_head"], parity["current_head"])
|
|
self.assertFalse(parity["in_parity"])
|
|
|
|
def test_master_parity_carries_the_live_remote_dimension(self) -> None:
|
|
"""B5: live_remote_head was never passed, dropping the #610 dimension."""
|
|
stale = system_health.StaleRuntime(
|
|
daemon_head="c" * 40,
|
|
checkout_head="c" * 40,
|
|
remote_head="d" * 40,
|
|
stale=False,
|
|
determinable=True,
|
|
mutation_safe=False,
|
|
reasons=(),
|
|
)
|
|
with patch.object(system_health, "assess_stale_runtime", return_value=stale):
|
|
diag = console_recovery.diagnose_recovery()
|
|
self.assertEqual(diag.master_parity.get("live_remote_head"), "d" * 40)
|
|
|
|
def test_verify_post_recovery(self) -> None:
|
|
verification = console_recovery.verify_post_recovery()
|
|
self.assertIn("clean", verification)
|
|
self.assertIn("status", verification)
|
|
self.assertIn("reasons", verification)
|
|
|
|
def test_unverified_inherited_binding_is_not_reported_clean(self) -> None:
|
|
"""B2: ``not clear_eligible`` also read clean for unproven bindings."""
|
|
binding = {
|
|
"classification": stale_binding_recovery.CLASSIFICATION_UNVERIFIED_INHERITED,
|
|
"clear_eligible": False,
|
|
}
|
|
diag = console_recovery.diagnose_recovery()
|
|
patched = console_recovery.RecoveryDiagnosis(
|
|
status=diag.status,
|
|
clean=diag.clean,
|
|
stale_runtime=diag.stale_runtime,
|
|
master_parity=diag.master_parity,
|
|
stale_binding=binding,
|
|
contamination=diag.contamination,
|
|
worktree_anomalies=diag.worktree_anomalies,
|
|
playbooks=diag.playbooks,
|
|
reasons=diag.reasons,
|
|
)
|
|
with patch.object(console_recovery, "diagnose_recovery", return_value=patched):
|
|
verification = console_recovery.verify_post_recovery()
|
|
self.assertFalse(verification["binding_clean"])
|
|
self.assertEqual(
|
|
verification["binding_classification"],
|
|
stale_binding_recovery.CLASSIFICATION_UNVERIFIED_INHERITED,
|
|
)
|
|
|
|
|
|
class TestConsoleRecoveryApi(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.app = create_app()
|
|
self.client = TestClient(self.app)
|
|
|
|
def test_api_recovery_diagnose(self) -> None:
|
|
res = self.client.get("/api/v1/system/recovery/diagnose")
|
|
self.assertEqual(res.status_code, 200)
|
|
data = res.json()
|
|
self.assertIn("status", data)
|
|
self.assertIn("clean", data)
|
|
self.assertIn("playbooks", data)
|
|
self.assertTrue(len(data["playbooks"]) >= 4)
|
|
|
|
def test_api_recovery_preview(self) -> None:
|
|
res = self.client.post(
|
|
"/api/v1/system/recovery/preview",
|
|
json={"playbook_id": "clear_stale_binding", "target": "active"},
|
|
)
|
|
self.assertEqual(res.status_code, 200)
|
|
data = res.json()
|
|
self.assertEqual(data["playbook_id"], "clear_stale_binding")
|
|
self.assertEqual(data["confirmation_phrase"], "confirm clear_stale_binding active")
|
|
self.assertIn("mutation_ledger", data)
|
|
|
|
def test_api_recovery_apply_denied_without_auth(self) -> None:
|
|
res = self.client.post(
|
|
"/api/v1/system/recovery/apply",
|
|
json={"playbook_id": "clear_stale_binding", "confirmation": "confirm clear_stale_binding"},
|
|
)
|
|
self.assertEqual(res.status_code, 400)
|
|
data = res.json()
|
|
self.assertFalse(data["success"])
|
|
self.assertFalse(data["allowed"])
|
|
|
|
def test_api_recovery_apply_refuses_phase_two_write_with_dev_auth(self) -> None:
|
|
"""B1: this previously asserted the phase-gate bypass as intended.
|
|
|
|
An authenticated operator posting a valid confirmation still must not
|
|
execute a phase-2 write while the console is in phase 1. The refusal is
|
|
the contract; a 200 here means the gate is not armed.
|
|
"""
|
|
env = {
|
|
"WEBUI_AUTH_MODE": "local_dev",
|
|
"WEBUI_DEV_SUBJECT": "[email protected]",
|
|
"WEBUI_DEV_ROLE": "operator",
|
|
}
|
|
before = os.environ.get("GITEA_ACTIVE_WORKTREE")
|
|
with patch.dict(os.environ, env):
|
|
res = self.client.post(
|
|
"/api/v1/system/recovery/apply",
|
|
json={
|
|
"playbook_id": "rebind_session_worktree",
|
|
"target": "branches/feat-issue-644",
|
|
"confirmation": "confirm rebind_session_worktree branches/feat-issue-644",
|
|
},
|
|
)
|
|
self.assertEqual(res.status_code, 400)
|
|
data = res.json()
|
|
self.assertFalse(data["success"])
|
|
self.assertFalse(data["allowed"])
|
|
self.assertEqual(data["error"], console_authz.DENY_PHASE_NOT_ACTIVE)
|
|
self.assertEqual(
|
|
os.environ.get("GITEA_ACTIVE_WORKTREE"),
|
|
before,
|
|
"a refused apply must not have rebound the live process environment",
|
|
)
|
|
|
|
def test_api_recovery_preview_reports_why_execution_is_disabled(self) -> None:
|
|
res = self.client.post(
|
|
"/api/v1/system/recovery/preview",
|
|
json={"playbook_id": "rebind_session_worktree", "target": "active"},
|
|
)
|
|
self.assertEqual(res.status_code, 200)
|
|
data = res.json()
|
|
self.assertFalse(data["execution_enabled"])
|
|
self.assertIn("execution_authorization", data)
|
|
|
|
def test_api_recovery_verify(self) -> None:
|
|
res = self.client.get("/api/v1/system/recovery/verify")
|
|
self.assertEqual(res.status_code, 200)
|
|
data = res.json()
|
|
self.assertIn("clean", data)
|
|
self.assertIn("status", data)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|