Merge remote-tracking branch 'prgs/master' into feat/issue-659-maintenance-drain-mode
This commit is contained in:
@@ -35,6 +35,17 @@ BREAK_GLASS_ENV = "GITEA_BREAKGLASS_RESTART_AUTHORIZATION"
|
||||
QUIET_SESSIONS: list[dict] = []
|
||||
QUIET_LEASES: list[dict] = []
|
||||
|
||||
# #669: broad restarts need a prior narrow-attempt log (unless break-glass).
|
||||
PRIOR_NARROW_ATTEMPTS_JSON = json.dumps(
|
||||
[
|
||||
{
|
||||
"action": "client_reconnect",
|
||||
"outcome": "insufficient",
|
||||
"reason": "still flapping after reconnect",
|
||||
}
|
||||
]
|
||||
)
|
||||
|
||||
|
||||
class _FakeDB:
|
||||
"""Minimal control-plane DB stand-in for the restart inventory."""
|
||||
@@ -128,6 +139,7 @@ class TestConjunction(_RestartToolHarness):
|
||||
preview = self._call(
|
||||
role="operator",
|
||||
restart_class="full_mcp_restart",
|
||||
prior_recovery_attempts_json=PRIOR_NARROW_ATTEMPTS_JSON,
|
||||
env={CONTROLLER_APPROVAL_ENV: "operator-approved"},
|
||||
)
|
||||
self.assertTrue(preview["allow_restart"],
|
||||
@@ -136,6 +148,7 @@ class TestConjunction(_RestartToolHarness):
|
||||
result = self._call(
|
||||
role="operator",
|
||||
restart_class="full_mcp_restart",
|
||||
prior_recovery_attempts_json=PRIOR_NARROW_ATTEMPTS_JSON,
|
||||
dry_run=False,
|
||||
drain_proof_json=self._clean_proof_for(preview),
|
||||
env={CONTROLLER_APPROVAL_ENV: "operator-approved"},
|
||||
@@ -342,6 +355,7 @@ class TestExistingPathsStillWork(_RestartToolHarness):
|
||||
result = self._call(
|
||||
role="operator",
|
||||
restart_class="full_mcp_restart",
|
||||
prior_recovery_attempts_json=PRIOR_NARROW_ATTEMPTS_JSON,
|
||||
dry_run=False,
|
||||
env={CONTROLLER_APPROVAL_ENV: "operator-approved"},
|
||||
)
|
||||
|
||||
@@ -0,0 +1,217 @@
|
||||
"""Unit tests for the scoped recovery playbook (#669)."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import recovery_playbook as rp
|
||||
import restart_coordinator as rc
|
||||
|
||||
|
||||
def test_ladder_covers_eleven_ordered_rungs():
|
||||
ranks = [r.rank for r in rp.RECOVERY_LADDER]
|
||||
assert ranks == list(range(len(rp.RECOVERY_LADDER)))
|
||||
assert len(rp.RECOVERY_LADDER) == 11
|
||||
assert rp.RECOVERY_LADDER[0].action is rp.RecoveryAction.CLIENT_RECONNECT
|
||||
assert rp.RECOVERY_LADDER[-1].action is rp.RecoveryAction.HOST_RESTART
|
||||
|
||||
|
||||
def test_ladder_document_links_parent_issues():
|
||||
doc = rp.ladder_document()
|
||||
assert "#655" in doc["parent_issues"]
|
||||
assert "#652" in doc["parent_issues"]
|
||||
assert "#653" in doc["parent_issues"]
|
||||
assert doc["enforcement_issue"] == "#669"
|
||||
assert "full_mcp_restart" in doc["broad_restart_actions"]
|
||||
|
||||
|
||||
def test_recommend_transport_eof_starts_at_client_reconnect():
|
||||
plan = rp.recommend_actions(symptoms=["transport_eof"])
|
||||
assert plan["recommended_actions"][0]["action"] == "client_reconnect"
|
||||
assert plan["recommended_actions"][0]["issue_links"]
|
||||
|
||||
|
||||
def test_recommend_skips_successful_prior_attempts():
|
||||
attempts = [
|
||||
rp.build_attempt_record(
|
||||
"client_reconnect", outcome="success", reason="reconnected"
|
||||
)
|
||||
]
|
||||
plan = rp.recommend_actions(
|
||||
symptoms=["transport_eof"], prior_recovery_attempts=attempts
|
||||
)
|
||||
actions = [a["action"] for a in plan["recommended_actions"]]
|
||||
assert "client_reconnect" not in actions
|
||||
assert actions[0] == "capability_refresh"
|
||||
|
||||
|
||||
def test_escalation_denied_without_attempt_log():
|
||||
result = rp.assess_escalation("full_mcp_restart", prior_recovery_attempts=[])
|
||||
assert result.allowed is False
|
||||
assert result.require_attempt_log is True
|
||||
assert any("#669" in r for r in result.reasons)
|
||||
assert result.recommended_next # soft recommendations still provided
|
||||
|
||||
|
||||
def test_escalation_allowed_after_insufficient_narrower():
|
||||
attempts = [
|
||||
rp.build_attempt_record(
|
||||
"client_reconnect",
|
||||
outcome="insufficient",
|
||||
reason="still flapping",
|
||||
),
|
||||
rp.build_attempt_record(
|
||||
"session_reconnect",
|
||||
outcome="failed",
|
||||
reason="namespace still dead",
|
||||
),
|
||||
]
|
||||
result = rp.assess_escalation(
|
||||
"full_mcp_restart", prior_recovery_attempts=attempts
|
||||
)
|
||||
assert result.allowed is True
|
||||
assert len(result.qualifying_attempts) == 2
|
||||
|
||||
|
||||
def test_escalation_break_glass_bypasses_attempt_log():
|
||||
result = rp.assess_escalation(
|
||||
"host_restart", prior_recovery_attempts=[], break_glass=True
|
||||
)
|
||||
assert result.allowed is True
|
||||
assert result.break_glass is True
|
||||
|
||||
|
||||
def test_narrow_action_does_not_require_attempt_log():
|
||||
result = rp.assess_escalation(
|
||||
"client_reconnect", prior_recovery_attempts=[]
|
||||
)
|
||||
assert result.allowed is True
|
||||
assert result.require_attempt_log is False
|
||||
|
||||
|
||||
def test_same_rank_attempt_does_not_qualify_for_escalation():
|
||||
attempts = [
|
||||
rp.build_attempt_record(
|
||||
"full_mcp_restart", outcome="failed", reason="already failed full"
|
||||
)
|
||||
]
|
||||
result = rp.assess_escalation(
|
||||
"full_mcp_restart", prior_recovery_attempts=attempts
|
||||
)
|
||||
assert result.allowed is False
|
||||
|
||||
|
||||
def test_success_outcome_does_not_qualify_for_escalation():
|
||||
attempts = [
|
||||
rp.build_attempt_record(
|
||||
"client_reconnect", outcome="success", reason="fixed"
|
||||
)
|
||||
]
|
||||
result = rp.assess_escalation(
|
||||
"full_mcp_restart", prior_recovery_attempts=attempts
|
||||
)
|
||||
assert result.allowed is False
|
||||
|
||||
|
||||
def test_recovery_metrics_fraction_avoided():
|
||||
attempts = [
|
||||
rp.build_attempt_record("client_reconnect", outcome="success"),
|
||||
rp.build_attempt_record("session_reconnect", outcome="success"),
|
||||
rp.build_attempt_record("full_mcp_restart", outcome="success"),
|
||||
]
|
||||
metrics = rp.recovery_metrics(attempts)
|
||||
assert metrics["successes_total"] == 3
|
||||
assert metrics["successes_avoided_full_restart"] == 2
|
||||
assert metrics["successes_full_or_host_restart"] == 1
|
||||
assert abs(metrics["fraction_avoided_full_restart"] - (2 / 3)) < 1e-9
|
||||
|
||||
|
||||
def test_coordinator_denies_full_restart_without_attempt_log():
|
||||
inv = {
|
||||
"inventory_complete": True,
|
||||
"sessions": [],
|
||||
"leases": [],
|
||||
"prior_recovery_attempts": [],
|
||||
}
|
||||
report = rc.evaluate_restart_impact(
|
||||
inv,
|
||||
restart_class=rc.RestartClass.FULL_MCP_RESTART,
|
||||
requester_role="controller",
|
||||
requester_permissions=rc.permissions_for_role("controller"),
|
||||
controller_approved=True,
|
||||
operator_authorized=True,
|
||||
)
|
||||
assert report.allow_restart is False
|
||||
assert report.attempt_log_satisfied is False
|
||||
assert report.verdict == rc.VERDICT_UNSAFE
|
||||
blob = " ".join(report.reasons + report.authorization_reasons)
|
||||
assert "#669" in blob or "attempt log" in blob
|
||||
|
||||
|
||||
def test_coordinator_allows_full_restart_with_attempt_log():
|
||||
inv = {
|
||||
"inventory_complete": True,
|
||||
"sessions": [],
|
||||
"leases": [],
|
||||
"prior_recovery_attempts": [
|
||||
{
|
||||
"action": "client_reconnect",
|
||||
"outcome": "insufficient",
|
||||
"reason": "still broken",
|
||||
}
|
||||
],
|
||||
}
|
||||
report = rc.evaluate_restart_impact(
|
||||
inv,
|
||||
restart_class=rc.RestartClass.FULL_MCP_RESTART,
|
||||
requester_role="controller",
|
||||
requester_permissions=rc.permissions_for_role("controller"),
|
||||
controller_approved=True,
|
||||
operator_authorized=True,
|
||||
)
|
||||
assert report.attempt_log_satisfied is True
|
||||
assert report.allow_restart is True
|
||||
assert report.verdict == rc.VERDICT_SAFE
|
||||
|
||||
|
||||
def test_coordinator_break_glass_allows_without_log():
|
||||
inv = {
|
||||
"inventory_complete": True,
|
||||
"sessions": [],
|
||||
"leases": [],
|
||||
"prior_recovery_attempts": [],
|
||||
}
|
||||
report = rc.evaluate_restart_impact(
|
||||
inv,
|
||||
restart_class=rc.RestartClass.FULL_MCP_RESTART,
|
||||
requester_role="controller",
|
||||
requester_permissions=rc.permissions_for_role("controller"),
|
||||
controller_approved=True,
|
||||
operator_authorized=True,
|
||||
break_glass=True,
|
||||
)
|
||||
assert report.break_glass is True
|
||||
assert report.attempt_log_satisfied is True
|
||||
assert report.allow_restart is True
|
||||
|
||||
|
||||
def test_coordinator_client_reconnect_unaffected():
|
||||
inv = {
|
||||
"inventory_complete": True,
|
||||
"sessions": [],
|
||||
"leases": [],
|
||||
"prior_recovery_attempts": [],
|
||||
}
|
||||
report = rc.evaluate_restart_impact(
|
||||
inv,
|
||||
restart_class=rc.RestartClass.CLIENT_RECONNECT,
|
||||
requester_role="author",
|
||||
requester_permissions=rc.permissions_for_role("author"),
|
||||
)
|
||||
assert report.attempt_log_satisfied is True
|
||||
assert report.allow_restart is True
|
||||
|
||||
|
||||
def test_restart_class_alias_accepted():
|
||||
assert (
|
||||
rp.resolve_action("full_mcp_restart")
|
||||
is rp.RecoveryAction.FULL_MCP_RESTART
|
||||
)
|
||||
@@ -0,0 +1,506 @@
|
||||
"""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()
|
||||
@@ -0,0 +1,465 @@
|
||||
"""Unit tests for Phase 3 Notifications and Human-Attention Console (#648)."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import pytest
|
||||
from starlette.testclient import TestClient
|
||||
|
||||
from webui.app import create_app
|
||||
from webui.notifications import (
|
||||
ATTENTION_HUMAN_REQUIRED,
|
||||
ATTENTION_OPERATOR,
|
||||
ATTENTION_ROUTINE,
|
||||
CATEGORY_AUTH,
|
||||
CATEGORY_BLOCKER,
|
||||
CATEGORY_LEASE,
|
||||
CATEGORY_SYSTEM,
|
||||
CATEGORY_VALIDATION,
|
||||
CATEGORY_WORKFLOW,
|
||||
NotificationItem,
|
||||
NotificationSnapshot,
|
||||
classify_attention_event,
|
||||
load_notifications_snapshot,
|
||||
snapshot_to_dict,
|
||||
)
|
||||
from webui.notification_views import render_notifications_page
|
||||
from webui.project_registry import load_registry
|
||||
from webui.queue_loader import QueueItem, QueueSnapshot
|
||||
from webui.lease_loader import CollisionWarning, LeaseSnapshot
|
||||
from webui.system_health import DependencyProbe, SystemHealthSnapshot, VersionInfo, StaleRuntime
|
||||
|
||||
|
||||
def test_classify_attention_event_rules():
|
||||
# 1. Critical escalation boundaries -> human-required
|
||||
att_cls, req_human = classify_attention_event(
|
||||
CATEGORY_AUTH, "Auth error", "Unauthorized access attempt", is_auth_failure=True
|
||||
)
|
||||
assert att_cls == ATTENTION_HUMAN_REQUIRED
|
||||
assert req_human is True
|
||||
|
||||
att_cls, req_human = classify_attention_event(
|
||||
CATEGORY_SYSTEM, "Hard stop", "Hard stop triggered", is_hard_stop=True
|
||||
)
|
||||
assert att_cls == ATTENTION_HUMAN_REQUIRED
|
||||
assert req_human is True
|
||||
|
||||
att_cls, req_human = classify_attention_event(
|
||||
CATEGORY_VALIDATION, "Validation Error", "Report validation failed", is_validation_failure=True
|
||||
)
|
||||
assert att_cls == ATTENTION_HUMAN_REQUIRED
|
||||
assert req_human is True
|
||||
|
||||
# 2. Operational issues -> operator
|
||||
att_cls, req_human = classify_attention_event(
|
||||
CATEGORY_BLOCKER, "PR Blocked", "Merge conflict detected", is_blocker=True
|
||||
)
|
||||
assert att_cls == ATTENTION_OPERATOR
|
||||
assert req_human is False
|
||||
|
||||
att_cls, req_human = classify_attention_event(
|
||||
CATEGORY_LEASE, "Lease Expired", "Session lease expired", is_stale=True
|
||||
)
|
||||
assert att_cls == ATTENTION_OPERATOR
|
||||
assert req_human is False
|
||||
|
||||
# 3. Routine workflow transitions -> routine
|
||||
att_cls, req_human = classify_attention_event(
|
||||
CATEGORY_WORKFLOW, "PR Active", "PR in review"
|
||||
)
|
||||
assert att_cls == ATTENTION_ROUTINE
|
||||
assert req_human is False
|
||||
|
||||
|
||||
def test_notification_snapshot_aggregation():
|
||||
reg = load_registry()
|
||||
proj_id = reg.projects[0].id if reg.projects else "gitea-tools"
|
||||
|
||||
mock_queue = QueueSnapshot(
|
||||
project_id=proj_id,
|
||||
repo_label="org/repo",
|
||||
prs=(
|
||||
QueueItem(
|
||||
number=101,
|
||||
title="Blocked PR",
|
||||
badges=("blocked",),
|
||||
extra={},
|
||||
),
|
||||
QueueItem(
|
||||
number=102,
|
||||
title="Normal PR",
|
||||
badges=("in-review",),
|
||||
extra={},
|
||||
),
|
||||
),
|
||||
issues=(),
|
||||
pr_pagination=None,
|
||||
issue_pagination=None,
|
||||
)
|
||||
|
||||
mock_leases = LeaseSnapshot(
|
||||
project_id=proj_id,
|
||||
repo_label="org/repo",
|
||||
issue_lock=None,
|
||||
claim_inventory={},
|
||||
reviewer_leases=(
|
||||
{
|
||||
"pr_number": 101,
|
||||
"status": "expired",
|
||||
"is_expired": True,
|
||||
},
|
||||
),
|
||||
duplicate_prs=(
|
||||
CollisionWarning(
|
||||
kind="duplicate_pr",
|
||||
message="Multiple open PRs for issue #101",
|
||||
issue_number=101,
|
||||
pr_numbers=(101, 103),
|
||||
),
|
||||
),
|
||||
duplicate_branches=(),
|
||||
collision_history=(),
|
||||
fetch_error=None,
|
||||
)
|
||||
|
||||
mock_version = VersionInfo(
|
||||
git_sha="abc1234",
|
||||
git_describe="v1.0.0",
|
||||
control_plane_schema_version=1,
|
||||
python_version="3.11",
|
||||
known=True,
|
||||
)
|
||||
|
||||
mock_stale = StaleRuntime(
|
||||
daemon_head="abc1234",
|
||||
checkout_head="abc1234",
|
||||
remote_head="abc1234",
|
||||
stale=False,
|
||||
determinable=True,
|
||||
mutation_safe=True,
|
||||
reasons=(),
|
||||
)
|
||||
|
||||
mock_health = SystemHealthSnapshot(
|
||||
status="degraded",
|
||||
ready=False,
|
||||
readiness_complete=True,
|
||||
readiness_reasons=("Auth failure",),
|
||||
service="webui",
|
||||
mode="test",
|
||||
version=mock_version,
|
||||
started_at="2026-07-25T00:00:00Z",
|
||||
uptime_seconds=100.0,
|
||||
timestamp="2026-07-25T00:00:00Z",
|
||||
deep_probes_requested=True,
|
||||
dependencies=(
|
||||
DependencyProbe(
|
||||
name="auth_service",
|
||||
kind="auth",
|
||||
status="unauthorized",
|
||||
detail="Token expired",
|
||||
required=True,
|
||||
),
|
||||
),
|
||||
mcp_namespaces=(),
|
||||
stale_runtime=mock_stale,
|
||||
probe_errors=(),
|
||||
)
|
||||
|
||||
snapshot = load_notifications_snapshot(
|
||||
proj_id,
|
||||
load_queue=lambda _id: mock_queue,
|
||||
load_leases=lambda **_kwargs: mock_leases,
|
||||
load_health=lambda **_kwargs: mock_health,
|
||||
)
|
||||
|
||||
assert snapshot.project_id == proj_id
|
||||
assert snapshot.total_count == 5
|
||||
assert snapshot.human_required_count >= 1 # auth probe failure
|
||||
assert snapshot.operator_count >= 3 # blocked PR + expired lease + duplicate PR collision
|
||||
assert snapshot.routine_count >= 1 # normal PR
|
||||
|
||||
# Inbox items should include operator and human-required items only
|
||||
inbox_classes = {item.attention_class for item in snapshot.inbox_items}
|
||||
assert ATTENTION_ROUTINE not in inbox_classes
|
||||
assert ATTENTION_OPERATOR in inbox_classes
|
||||
assert ATTENTION_HUMAN_REQUIRED in inbox_classes
|
||||
|
||||
|
||||
def test_snapshot_to_dict_and_redaction():
|
||||
item = NotificationItem(
|
||||
id="notif-1",
|
||||
attention_class=ATTENTION_HUMAN_REQUIRED,
|
||||
category=CATEGORY_AUTH,
|
||||
title="Auth Error",
|
||||
summary="Failed auth header: Bearer secret_token_12345",
|
||||
work_kind="system",
|
||||
work_number=None,
|
||||
project_id="test-proj",
|
||||
repo_label="org/repo",
|
||||
created_at="2026-07-25T16:00:00Z",
|
||||
requires_human=True,
|
||||
)
|
||||
snap = NotificationSnapshot(
|
||||
project_id="test-proj",
|
||||
repo_label="org/repo",
|
||||
items=(item,),
|
||||
human_required_count=1,
|
||||
operator_count=0,
|
||||
routine_count=0,
|
||||
total_count=1,
|
||||
)
|
||||
|
||||
data = snapshot_to_dict(snap)
|
||||
assert data["project_id"] == "test-proj"
|
||||
assert data["human_required_count"] == 1
|
||||
assert len(data["inbox_items"]) == 1
|
||||
|
||||
# Redaction test
|
||||
summary = data["inbox_items"][0]["summary"]
|
||||
assert "secret_token_12345" not in summary
|
||||
assert "<redacted>" in summary or "Bearer" in summary
|
||||
|
||||
|
||||
def test_notifications_html_views():
|
||||
item = NotificationItem(
|
||||
id="notif-1",
|
||||
attention_class=ATTENTION_HUMAN_REQUIRED,
|
||||
category=CATEGORY_AUTH,
|
||||
title="Critical Auth Failure",
|
||||
summary="Auth failure details",
|
||||
work_kind="issue",
|
||||
work_number=42,
|
||||
project_id="test-proj",
|
||||
repo_label="org/repo",
|
||||
created_at="2026-07-25T16:00:00Z",
|
||||
requires_human=True,
|
||||
)
|
||||
snap = NotificationSnapshot(
|
||||
project_id="test-proj",
|
||||
repo_label="org/repo",
|
||||
items=(item,),
|
||||
human_required_count=1,
|
||||
operator_count=0,
|
||||
routine_count=0,
|
||||
total_count=1,
|
||||
)
|
||||
|
||||
html = render_notifications_page(snap, filter_class="inbox")
|
||||
assert "Notifications & Attention Inbox" in html or "Notifications & Attention Inbox" in html
|
||||
assert "Critical Auth Failure" in html
|
||||
assert "HUMAN REQUIRED" in html
|
||||
assert "Human Required" in html
|
||||
|
||||
|
||||
def test_notifications_app_routes():
|
||||
app = create_app()
|
||||
client = TestClient(app)
|
||||
|
||||
# 1. HTML Route
|
||||
res = client.get("/notifications")
|
||||
assert res.status_code == 200
|
||||
assert "Notifications" in res.text
|
||||
assert "Attention Inbox" in res.text
|
||||
|
||||
# 2. API Route /api/v1/notifications
|
||||
res_api = client.get("/api/v1/notifications")
|
||||
assert res_api.status_code == 200
|
||||
json_data = res_api.json()
|
||||
assert "human_required_count" in json_data
|
||||
assert "operator_count" in json_data
|
||||
assert "routine_count" in json_data
|
||||
assert "inbox_items" in json_data
|
||||
|
||||
# 3. Compatibility Alias /api/notifications
|
||||
res_alias = client.get("/api/notifications")
|
||||
assert res_alias.status_code == 200
|
||||
assert res_alias.json()["project_id"] == json_data["project_id"]
|
||||
|
||||
|
||||
def test_classify_ignores_human_authored_title_and_summary_keywords():
|
||||
"""B1: keywords in human-authored titles must not escalate routine work (#905)."""
|
||||
# Routine transition whose title/summary mention critical-boundary words
|
||||
att_cls, req_human = classify_attention_event(
|
||||
CATEGORY_WORKFLOW,
|
||||
"record irrecoverable decision lock provenance",
|
||||
"PR #999 'record irrecoverable decision lock provenance' is in routine state in-review.",
|
||||
)
|
||||
assert att_cls == ATTENTION_ROUTINE
|
||||
assert req_human is False
|
||||
|
||||
att_cls, req_human = classify_attention_event(
|
||||
CATEGORY_WORKFLOW,
|
||||
"fix unauthorized token path",
|
||||
"Issue #1 'fix unauthorized token path' state: claimed. hard stop docs only.",
|
||||
)
|
||||
assert att_cls == ATTENTION_ROUTINE
|
||||
assert req_human is False
|
||||
|
||||
# Structured flags still escalate (machine-driven)
|
||||
att_cls, req_human = classify_attention_event(
|
||||
CATEGORY_SYSTEM,
|
||||
"anything",
|
||||
"anything with hard stop in text",
|
||||
is_hard_stop=True,
|
||||
)
|
||||
assert att_cls == ATTENTION_HUMAN_REQUIRED
|
||||
assert req_human is True
|
||||
|
||||
|
||||
def test_notification_ids_are_unique_across_probe_errors_and_collisions():
|
||||
"""B2: published notification ids must be unique within a snapshot (#905)."""
|
||||
reg = load_registry()
|
||||
proj_id = reg.projects[0].id if reg.projects else "gitea-tools"
|
||||
|
||||
mock_queue = QueueSnapshot(
|
||||
project_id=proj_id,
|
||||
repo_label="org/repo",
|
||||
prs=(),
|
||||
issues=(),
|
||||
pr_pagination=None,
|
||||
issue_pagination=None,
|
||||
)
|
||||
mock_leases = LeaseSnapshot(
|
||||
project_id=proj_id,
|
||||
repo_label="org/repo",
|
||||
issue_lock=None,
|
||||
claim_inventory={},
|
||||
reviewer_leases=(),
|
||||
duplicate_prs=(
|
||||
CollisionWarning(
|
||||
kind="duplicate_pr",
|
||||
message="Multiple open PRs for issue #10",
|
||||
issue_number=10,
|
||||
pr_numbers=(10, 11),
|
||||
),
|
||||
CollisionWarning(
|
||||
kind="duplicate_branch",
|
||||
message="Another collision without issue",
|
||||
issue_number=None,
|
||||
pr_numbers=(12, 13),
|
||||
),
|
||||
CollisionWarning(
|
||||
kind="duplicate_pr",
|
||||
message="Second issue collision",
|
||||
issue_number=10,
|
||||
pr_numbers=(14, 15),
|
||||
),
|
||||
),
|
||||
duplicate_branches=(),
|
||||
collision_history=(),
|
||||
fetch_error=None,
|
||||
)
|
||||
mock_version = VersionInfo(
|
||||
git_sha="abc1234",
|
||||
git_describe="v1.0.0",
|
||||
control_plane_schema_version=1,
|
||||
python_version="3.11",
|
||||
known=True,
|
||||
)
|
||||
mock_stale = StaleRuntime(
|
||||
daemon_head="abc1234",
|
||||
checkout_head="abc1234",
|
||||
remote_head="abc1234",
|
||||
stale=False,
|
||||
determinable=True,
|
||||
mutation_safe=True,
|
||||
reasons=(),
|
||||
)
|
||||
mock_health = SystemHealthSnapshot(
|
||||
status="degraded",
|
||||
ready=False,
|
||||
readiness_complete=True,
|
||||
readiness_reasons=(),
|
||||
service="webui",
|
||||
mode="test",
|
||||
version=mock_version,
|
||||
started_at="2026-07-25T00:00:00Z",
|
||||
uptime_seconds=100.0,
|
||||
timestamp="2026-07-25T00:00:00Z",
|
||||
deep_probes_requested=True,
|
||||
dependencies=(),
|
||||
mcp_namespaces=(),
|
||||
stale_runtime=mock_stale,
|
||||
probe_errors=("error alpha", "error beta"),
|
||||
)
|
||||
|
||||
snapshot = load_notifications_snapshot(
|
||||
proj_id,
|
||||
load_queue=lambda _id: mock_queue,
|
||||
load_leases=lambda **_kwargs: mock_leases,
|
||||
load_health=lambda **_kwargs: mock_health,
|
||||
)
|
||||
ids = [item.id for item in snapshot.items]
|
||||
assert len(ids) == len(set(ids)), f"duplicate notification ids: {ids}"
|
||||
assert any(i.startswith(f"notif-sys-err-{proj_id}-") for i in ids)
|
||||
assert any(i.startswith("notif-collision-") for i in ids)
|
||||
|
||||
|
||||
def test_probe_errors_do_not_set_fetch_error():
|
||||
"""B3: probe_errors must not be reported as fetch_error (#905)."""
|
||||
reg = load_registry()
|
||||
proj_id = reg.projects[0].id if reg.projects else "gitea-tools"
|
||||
|
||||
mock_queue = QueueSnapshot(
|
||||
project_id=proj_id,
|
||||
repo_label="org/repo",
|
||||
prs=(),
|
||||
issues=(),
|
||||
pr_pagination=None,
|
||||
issue_pagination=None,
|
||||
fetch_error=None,
|
||||
)
|
||||
mock_leases = LeaseSnapshot(
|
||||
project_id=proj_id,
|
||||
repo_label="org/repo",
|
||||
issue_lock=None,
|
||||
claim_inventory={},
|
||||
reviewer_leases=(),
|
||||
duplicate_prs=(),
|
||||
duplicate_branches=(),
|
||||
collision_history=(),
|
||||
fetch_error=None,
|
||||
)
|
||||
mock_version = VersionInfo(
|
||||
git_sha="abc1234",
|
||||
git_describe="v1.0.0",
|
||||
control_plane_schema_version=1,
|
||||
python_version="3.11",
|
||||
known=True,
|
||||
)
|
||||
mock_stale = StaleRuntime(
|
||||
daemon_head="abc1234",
|
||||
checkout_head="abc1234",
|
||||
remote_head="abc1234",
|
||||
stale=False,
|
||||
determinable=True,
|
||||
mutation_safe=True,
|
||||
reasons=(),
|
||||
)
|
||||
mock_health = SystemHealthSnapshot(
|
||||
status="degraded",
|
||||
ready=False,
|
||||
readiness_complete=True,
|
||||
readiness_reasons=(),
|
||||
service="webui",
|
||||
mode="test",
|
||||
version=mock_version,
|
||||
started_at="2026-07-25T00:00:00Z",
|
||||
uptime_seconds=100.0,
|
||||
timestamp="2026-07-25T00:00:00Z",
|
||||
deep_probes_requested=True,
|
||||
dependencies=(),
|
||||
mcp_namespaces=(),
|
||||
stale_runtime=mock_stale,
|
||||
probe_errors=("probe blew up",),
|
||||
)
|
||||
|
||||
snapshot = load_notifications_snapshot(
|
||||
proj_id,
|
||||
load_queue=lambda _id: mock_queue,
|
||||
load_leases=lambda **_kwargs: mock_leases,
|
||||
load_health=lambda **_kwargs: mock_health,
|
||||
)
|
||||
assert snapshot.fetch_error is None
|
||||
# probe errors still appear as items
|
||||
assert any("probe blew up" in item.summary for item in snapshot.items)
|
||||
@@ -0,0 +1,452 @@
|
||||
"""Read-only restart console: views, gates, and honesty rules (#667).
|
||||
|
||||
The console consumes the #655 substrate. These tests hold it to the three
|
||||
properties that make a status surface trustworthy:
|
||||
|
||||
* an unreadable source is reported unavailable, never rendered as green;
|
||||
* authorization is probed the way execution would probe it, so an allow is
|
||||
never shown for something that could not run;
|
||||
* the surface performs no mutation, including no write to the control-plane DB.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import sqlite3
|
||||
import sys
|
||||
import tempfile
|
||||
import unittest
|
||||
from datetime import datetime, timedelta, timezone
|
||||
from pathlib import Path
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
||||
|
||||
from starlette.testclient import TestClient # noqa: E402
|
||||
|
||||
import restart_coordinator # noqa: E402
|
||||
from webui import console_authz, restart_console, restart_views # noqa: E402
|
||||
from webui.app import create_app # noqa: E402
|
||||
|
||||
NOW = datetime(2026, 7, 25, 21, 0, 0, tzinfo=timezone.utc)
|
||||
|
||||
|
||||
def _principal(role: str) -> console_authz.Principal:
|
||||
return console_authz.Principal(
|
||||
subject="[email protected]",
|
||||
role=role,
|
||||
identity_source=console_authz.IDENTITY_LOCAL_DEV,
|
||||
authenticated=True,
|
||||
)
|
||||
|
||||
|
||||
def _inventory(*, complete: bool = True, sessions=(), leases=()):
|
||||
def _read(**_kwargs):
|
||||
return {
|
||||
"sessions": list(sessions),
|
||||
"leases": list(leases),
|
||||
"terminal_lock": None,
|
||||
"prior_recovery_attempts": [],
|
||||
"inventory_complete": complete,
|
||||
"incomplete_reasons": (
|
||||
[] if complete else ["fixture: inventory withheld"]
|
||||
),
|
||||
}
|
||||
|
||||
return _read
|
||||
|
||||
|
||||
def _live_session(session_id: str = "prgs-author-1234-abcd") -> dict:
|
||||
return {
|
||||
"session_id": session_id,
|
||||
"role": "author",
|
||||
"profile": "prgs-author",
|
||||
"pid": os.getpid(),
|
||||
"status": "active",
|
||||
"last_heartbeat_at": (NOW - timedelta(seconds=30)).isoformat(),
|
||||
}
|
||||
|
||||
|
||||
def drain_proof_fixture() -> dict:
|
||||
"""A structurally complete but unsigned drain proof."""
|
||||
return {
|
||||
"version": "drain-proof/v1",
|
||||
"proof_id": "deadbeef" * 8,
|
||||
"clean": True,
|
||||
"issued_at": (NOW - timedelta(minutes=1)).isoformat(),
|
||||
"expires_at": (NOW + timedelta(minutes=5)).isoformat(),
|
||||
"requesting_session_id": "s-live",
|
||||
"impact_fingerprint": "f" * 64,
|
||||
"checks": [],
|
||||
"failed_checks": [],
|
||||
}
|
||||
|
||||
|
||||
class RestartClassMatrixTest(unittest.TestCase):
|
||||
def test_every_policy_class_is_rendered(self) -> None:
|
||||
views = restart_console.build_restart_class_views("operator")
|
||||
self.assertEqual(len(views), len(restart_coordinator.RESTART_CLASS_POLICIES))
|
||||
|
||||
def test_viewer_capability_is_role_scoped_not_generic(self) -> None:
|
||||
"""A worker role must not be shown as able to request a full restart."""
|
||||
author = {
|
||||
v.restart_class: v
|
||||
for v in restart_console.build_restart_class_views("author")
|
||||
}
|
||||
operator = {
|
||||
v.restart_class: v
|
||||
for v in restart_console.build_restart_class_views("operator")
|
||||
}
|
||||
full = restart_coordinator.RestartClass.FULL_MCP_RESTART.value
|
||||
|
||||
self.assertFalse(author[full].viewer_may_request)
|
||||
self.assertFalse(author[full].viewer_may_execute)
|
||||
self.assertTrue(operator[full].viewer_may_request)
|
||||
self.assertTrue(operator[full].viewer_may_execute)
|
||||
|
||||
def test_unknown_role_may_do_nothing(self) -> None:
|
||||
views = restart_console.build_restart_class_views("not-a-role")
|
||||
self.assertTrue(all(not v.viewer_may_request for v in views))
|
||||
self.assertTrue(all(not v.viewer_may_execute for v in views))
|
||||
|
||||
|
||||
class AuthorizationProbeTest(unittest.TestCase):
|
||||
def test_probe_asks_for_execution_so_phase_gate_is_reported(self) -> None:
|
||||
"""An admin clears the role bar and still cannot execute in Phase 1.
|
||||
|
||||
This is the case that distinguishes the two probes. Asked without
|
||||
``for_execution`` an admin is *allowed* for ``system.restart_namespace``,
|
||||
which on a control surface reads as a live button. Asked the way
|
||||
execution asks, the same principal is refused ``phase_not_active``. The
|
||||
console must report the second answer.
|
||||
"""
|
||||
by_id = {
|
||||
a.action_id: a
|
||||
for a in restart_console.build_action_authorizations(
|
||||
_principal(console_authz.ADMIN)
|
||||
)
|
||||
}
|
||||
restart = by_id["system.restart_namespace"]
|
||||
|
||||
self.assertFalse(restart.execution_enabled)
|
||||
self.assertEqual(restart.reason_code, console_authz.DENY_PHASE_NOT_ACTIVE)
|
||||
|
||||
permissive = console_authz.authorize(
|
||||
"system.restart_namespace", _principal(console_authz.ADMIN)
|
||||
)
|
||||
self.assertTrue(
|
||||
permissive.allowed,
|
||||
"guard precondition: without for_execution an admin is allowed, "
|
||||
"which is exactly why the console must not probe that way",
|
||||
)
|
||||
|
||||
def test_operator_is_refused_the_admin_only_restart_action(self) -> None:
|
||||
"""Role refusal precedes the phase gate and is reported as such."""
|
||||
by_id = {
|
||||
a.action_id: a
|
||||
for a in restart_console.build_action_authorizations(
|
||||
_principal(console_authz.OPERATOR)
|
||||
)
|
||||
}
|
||||
self.assertEqual(
|
||||
by_id["system.restart_namespace"].reason_code,
|
||||
console_authz.DENY_INSUFFICIENT_ROLE,
|
||||
)
|
||||
|
||||
def test_anonymous_is_denied_unauthenticated(self) -> None:
|
||||
by_id = {
|
||||
a.action_id: a for a in restart_console.build_action_authorizations(None)
|
||||
}
|
||||
self.assertEqual(
|
||||
by_id["system.restart_namespace"].reason_code,
|
||||
console_authz.DENY_UNAUTHENTICATED,
|
||||
)
|
||||
|
||||
def test_no_authorization_ever_reports_execution_enabled(self) -> None:
|
||||
for role in (
|
||||
console_authz.VIEWER,
|
||||
console_authz.OPERATOR,
|
||||
console_authz.CONTROLLER,
|
||||
console_authz.ADMIN,
|
||||
):
|
||||
for auth in restart_console.build_action_authorizations(_principal(role)):
|
||||
self.assertFalse(
|
||||
auth.execution_enabled,
|
||||
f"{role} reported execution_enabled for {auth.action_id}",
|
||||
)
|
||||
|
||||
|
||||
class ImpactPreviewTest(unittest.TestCase):
|
||||
def test_impact_renders_from_coordinator_dto(self) -> None:
|
||||
impact, source = restart_console.load_impact_report(
|
||||
principal=_principal(console_authz.OPERATOR),
|
||||
read_inventory=_inventory(sessions=[_live_session()]),
|
||||
now=NOW,
|
||||
)
|
||||
self.assertTrue(source.available)
|
||||
self.assertIsNotNone(impact)
|
||||
self.assertEqual(
|
||||
impact["restart_class"],
|
||||
restart_coordinator.RestartClass.FULL_MCP_RESTART.value,
|
||||
)
|
||||
self.assertIn("verdict", impact)
|
||||
self.assertFalse(impact["restart_performed"])
|
||||
self.assertTrue(impact["dry_run"])
|
||||
|
||||
def test_incomplete_inventory_is_surfaced_and_denies(self) -> None:
|
||||
impact, source = restart_console.load_impact_report(
|
||||
principal=_principal(console_authz.OPERATOR),
|
||||
read_inventory=_inventory(complete=False),
|
||||
now=NOW,
|
||||
)
|
||||
self.assertFalse(impact["inventory_complete"])
|
||||
self.assertFalse(impact["allow_restart"])
|
||||
self.assertTrue(source.detail, "incomplete inventory must explain itself")
|
||||
|
||||
def test_inventory_reader_failure_is_unavailable_not_empty(self) -> None:
|
||||
"""A reader that raises must not be rendered as 'no sessions affected'."""
|
||||
|
||||
def _boom(**_kwargs):
|
||||
raise RuntimeError("control-plane unreachable")
|
||||
|
||||
impact, source = restart_console.load_impact_report(
|
||||
principal=_principal(console_authz.OPERATOR),
|
||||
read_inventory=_boom,
|
||||
now=NOW,
|
||||
)
|
||||
self.assertIsNone(impact)
|
||||
self.assertFalse(source.available)
|
||||
self.assertIn("control-plane unreachable", source.detail)
|
||||
|
||||
|
||||
class ControlPlaneReadTest(unittest.TestCase):
|
||||
def test_missing_database_is_incomplete_not_empty(self) -> None:
|
||||
inventory = restart_console.read_control_plane_inventory(
|
||||
db_path="/nonexistent/control-plane.sqlite3"
|
||||
)
|
||||
self.assertFalse(inventory["inventory_complete"])
|
||||
self.assertEqual(inventory["sessions"], [])
|
||||
self.assertTrue(inventory["incomplete_reasons"])
|
||||
|
||||
def test_reader_never_creates_the_database(self) -> None:
|
||||
"""Reading status must not bring a control-plane DB into existence.
|
||||
|
||||
The path deliberately sits in a directory that already exists: a
|
||||
read-write ``sqlite3.connect`` would happily create the file there, so
|
||||
this fails if the reader ever stops opening the database ``mode=ro``.
|
||||
A nested-missing-directory path would pass for the wrong reason,
|
||||
because sqlite cannot create the parent directory either way.
|
||||
"""
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
path = os.path.join(tmp, "control_plane.sqlite3")
|
||||
self.assertTrue(os.path.isdir(os.path.dirname(path)))
|
||||
|
||||
inventory = restart_console.read_control_plane_inventory(db_path=path)
|
||||
|
||||
self.assertFalse(
|
||||
os.path.exists(path),
|
||||
"reading restart status created a control-plane database",
|
||||
)
|
||||
self.assertFalse(inventory["inventory_complete"])
|
||||
|
||||
def test_reads_active_sessions_from_a_real_database(self) -> None:
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
path = os.path.join(tmp, "cp.sqlite3")
|
||||
conn = sqlite3.connect(path)
|
||||
conn.execute(
|
||||
"CREATE TABLE sessions (session_id TEXT, role TEXT, profile TEXT,"
|
||||
" pid INTEGER, status TEXT, last_heartbeat_at TEXT)"
|
||||
)
|
||||
conn.execute(
|
||||
"CREATE TABLE work_items (work_item_id INTEGER, kind TEXT,"
|
||||
" number INTEGER)"
|
||||
)
|
||||
conn.execute(
|
||||
"CREATE TABLE leases (lease_id TEXT, session_id TEXT, role TEXT,"
|
||||
" phase TEXT, status TEXT, worktree_path TEXT,"
|
||||
" work_item_id INTEGER, expires_at TEXT)"
|
||||
)
|
||||
conn.execute(
|
||||
"INSERT INTO sessions VALUES (?,?,?,?,?,?)",
|
||||
("s-live", "author", "prgs-author", 4242, "active", NOW.isoformat()),
|
||||
)
|
||||
conn.execute(
|
||||
"INSERT INTO sessions VALUES (?,?,?,?,?,?)",
|
||||
("s-done", "author", "prgs-author", 11, "closed", NOW.isoformat()),
|
||||
)
|
||||
conn.execute("INSERT INTO work_items VALUES (1, 'issue', 667)")
|
||||
conn.execute(
|
||||
"INSERT INTO leases VALUES (?,?,?,?,?,?,?,?)",
|
||||
(
|
||||
"l-1",
|
||||
"s-live",
|
||||
"author",
|
||||
"allocated",
|
||||
"active",
|
||||
None,
|
||||
1,
|
||||
NOW.isoformat(),
|
||||
),
|
||||
)
|
||||
conn.commit()
|
||||
conn.close()
|
||||
|
||||
inventory = restart_console.read_control_plane_inventory(db_path=path)
|
||||
|
||||
self.assertTrue(inventory["inventory_complete"])
|
||||
self.assertEqual([s["session_id"] for s in inventory["sessions"]], ["s-live"])
|
||||
self.assertEqual(inventory["leases"][0]["work_number"], 667)
|
||||
|
||||
|
||||
class DrainAndReconcileTest(unittest.TestCase):
|
||||
def test_absent_drain_proof_is_not_a_pass(self) -> None:
|
||||
drain, source = restart_console.load_drain_status(proof=None, now=NOW)
|
||||
self.assertIsNone(drain)
|
||||
self.assertFalse(source.available)
|
||||
self.assertIn("denies", source.detail)
|
||||
|
||||
def test_tampered_drain_proof_is_reported_invalid(self) -> None:
|
||||
proof = drain_proof_fixture()
|
||||
proof["clean"] = True
|
||||
proof["proof_id"] = "0" * 64
|
||||
drain, source = restart_console.load_drain_status(proof=proof, now=NOW)
|
||||
self.assertTrue(source.available)
|
||||
self.assertFalse(drain["valid"])
|
||||
|
||||
def test_absent_reconcile_proof_is_unavailable(self) -> None:
|
||||
reconcile, source = restart_console.load_reconcile_status(load_proof=None)
|
||||
self.assertIsNone(reconcile)
|
||||
self.assertFalse(source.available)
|
||||
|
||||
def test_reconcile_proof_is_rendered_when_supplied(self) -> None:
|
||||
payload = {
|
||||
"overall_status": "degraded",
|
||||
"mode": "log_only",
|
||||
"resolved_count": 3,
|
||||
"unresolved_count": 2,
|
||||
"items": [
|
||||
{
|
||||
"dimension": "leases",
|
||||
"status": "unresolved",
|
||||
"summary": "2 orphaned leases",
|
||||
"follow_up_required": True,
|
||||
}
|
||||
],
|
||||
}
|
||||
reconcile, source = restart_console.load_reconcile_status(
|
||||
load_proof=lambda: payload
|
||||
)
|
||||
self.assertTrue(source.available)
|
||||
self.assertEqual(reconcile["unresolved_count"], 2)
|
||||
|
||||
|
||||
class RenderingTest(unittest.TestCase):
|
||||
def _snapshot(self, **kwargs):
|
||||
params = {
|
||||
"principal": _principal(console_authz.OPERATOR),
|
||||
"read_inventory": _inventory(sessions=[_live_session()]),
|
||||
"now": NOW,
|
||||
}
|
||||
params.update(kwargs)
|
||||
return restart_console.load_restart_console_snapshot(**params)
|
||||
|
||||
def test_page_renders_every_section(self) -> None:
|
||||
html = restart_views.render_restart_console_page(self._snapshot())
|
||||
for heading in (
|
||||
"Impact preview",
|
||||
"Drain proof",
|
||||
"Post-restart reconcile",
|
||||
"Restart classes",
|
||||
"Approval controls",
|
||||
"Break-glass",
|
||||
):
|
||||
self.assertIn(heading, html)
|
||||
|
||||
def test_hostile_session_id_is_escaped(self) -> None:
|
||||
hostile = "<script>alert('x')</script>"
|
||||
html = restart_views.render_restart_console_page(
|
||||
self._snapshot(read_inventory=_inventory(sessions=[_live_session(hostile)]))
|
||||
)
|
||||
self.assertNotIn("<script>alert", html)
|
||||
self.assertIn("<script>", html)
|
||||
|
||||
def test_unavailable_impact_says_unsafe_rather_than_clean(self) -> None:
|
||||
def _boom(**_kwargs):
|
||||
raise RuntimeError("nope")
|
||||
|
||||
snapshot = self._snapshot(read_inventory=_boom)
|
||||
html = restart_views.render_restart_console_page(snapshot)
|
||||
self.assertIn("blast radius of a restart is unknown", html)
|
||||
self.assertIn("unavailable", html)
|
||||
|
||||
def test_break_glass_is_hidden_from_unprivileged_viewers(self) -> None:
|
||||
viewer_html = restart_views.render_restart_console_page(
|
||||
self._snapshot(principal=_principal(console_authz.VIEWER))
|
||||
)
|
||||
self.assertIn("visible to operator-class", viewer_html)
|
||||
self.assertNotIn(
|
||||
f"#{restart_console.BREAK_GLASS_ISSUE}", viewer_html
|
||||
)
|
||||
|
||||
def test_break_glass_shown_to_operator_is_marked_unavailable(self) -> None:
|
||||
html = restart_views.render_restart_console_page(self._snapshot())
|
||||
self.assertIn("unavailable", html)
|
||||
self.assertIn(f"#{restart_console.BREAK_GLASS_ISSUE}", html)
|
||||
|
||||
def test_snapshot_always_declares_itself_read_only(self) -> None:
|
||||
self.assertTrue(self._snapshot().read_only)
|
||||
|
||||
|
||||
class RestartConsoleRouteTest(unittest.TestCase):
|
||||
def setUp(self) -> None:
|
||||
self.client = TestClient(create_app())
|
||||
|
||||
def test_page_route_renders(self) -> None:
|
||||
res = self.client.get("/runtime/restart")
|
||||
self.assertEqual(res.status_code, 200)
|
||||
self.assertIn("Restart status and impact", res.text)
|
||||
|
||||
def test_api_route_exports_snapshot(self) -> None:
|
||||
res = self.client.get("/api/v1/system/restart/status")
|
||||
self.assertEqual(res.status_code, 200)
|
||||
payload = res.json()
|
||||
self.assertTrue(payload["read_only"])
|
||||
self.assertEqual(payload["links"]["issue"], 667)
|
||||
self.assertEqual(
|
||||
len(payload["restart_classes"]),
|
||||
len(restart_coordinator.RESTART_CLASS_POLICIES),
|
||||
)
|
||||
|
||||
def test_restart_class_is_selectable(self) -> None:
|
||||
res = self.client.get(
|
||||
"/api/v1/system/restart/status?restart_class=client_reconnect"
|
||||
)
|
||||
self.assertEqual(res.status_code, 200)
|
||||
self.assertEqual(res.json()["impact"]["restart_class"], "client_reconnect")
|
||||
|
||||
def test_unknown_restart_class_fails_closed(self) -> None:
|
||||
res = self.client.get(
|
||||
"/api/v1/system/restart/status?restart_class=obliterate-everything"
|
||||
)
|
||||
self.assertEqual(res.status_code, 200)
|
||||
impact = res.json()["impact"]
|
||||
self.assertFalse(impact["allow_restart"])
|
||||
|
||||
def test_anonymous_api_reader_gets_no_execution_grant(self) -> None:
|
||||
payload = self.client.get("/api/v1/system/restart/status").json()
|
||||
self.assertFalse(payload["break_glass"]["available"])
|
||||
for auth in payload["authorizations"]:
|
||||
self.assertFalse(auth["execution_enabled"])
|
||||
|
||||
def test_route_is_registered_in_nav(self) -> None:
|
||||
from webui.nav import nav_hrefs
|
||||
|
||||
self.assertIn("/runtime/restart", nav_hrefs())
|
||||
|
||||
def test_no_write_method_is_exposed(self) -> None:
|
||||
"""The surface is read-only: nothing accepts a POST."""
|
||||
for path in ("/runtime/restart", "/api/v1/system/restart/status"):
|
||||
self.assertEqual(self.client.post(path).status_code, 405, path)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user