Merge branch 'master' into docs/issue-656-mcp-restart-governance
This commit is contained in:
@@ -0,0 +1,243 @@
|
||||
"""Allocator epic / child-only container pre-rank exclusion (#844).
|
||||
|
||||
Covers:
|
||||
* Issue #631-shaped child-only epic is excluded before ranking.
|
||||
* Implementable child issues remain eligible and can be selected.
|
||||
* Ordinary issues that merely mention "epic" in title/body are not excluded.
|
||||
* Excluded containers never receive assignments or workflow leases.
|
||||
* Structured skip reason ``epic_or_child_only_container`` is reported.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import tempfile
|
||||
import unittest
|
||||
|
||||
from allocator_service import (
|
||||
OUTCOME_ASSIGNED,
|
||||
OUTCOME_PREVIEW,
|
||||
SKIP_EPIC_OR_CHILD_ONLY_CONTAINER,
|
||||
WorkCandidate,
|
||||
allocate_next_work,
|
||||
classify_epic_or_child_only_container,
|
||||
)
|
||||
from control_plane_db import ControlPlaneDB
|
||||
|
||||
REMOTE = "prgs"
|
||||
ORG = "Scaled-Tech-Consulting"
|
||||
REPO = "Gitea-Tools"
|
||||
|
||||
# Minimal body mirroring issue #631 authoritative scope language.
|
||||
_EPIC_631_BODY = """
|
||||
## Scope (umbrella)
|
||||
|
||||
This epic owns the **product roadmap and linkage** for the Web Console.
|
||||
Implementation is delivered via child issues only.
|
||||
|
||||
## Explicit non-goals
|
||||
|
||||
* Do not implement product features in this epic issue itself.
|
||||
* No product feature implementation is claimed complete solely on this epic.
|
||||
"""
|
||||
|
||||
_CHILD_BODY = """
|
||||
## Problem
|
||||
|
||||
Operators need a workflow-event timeline model for Phase 1.
|
||||
|
||||
## Acceptance criteria
|
||||
|
||||
- [ ] Timeline model API exists
|
||||
"""
|
||||
|
||||
|
||||
def _issue(
|
||||
number: int,
|
||||
*,
|
||||
title: str = "",
|
||||
body: str = "",
|
||||
labels: tuple[str, ...] = ("status:ready", "type:feature"),
|
||||
priority: int = 20,
|
||||
) -> WorkCandidate:
|
||||
return WorkCandidate(
|
||||
kind="issue",
|
||||
number=number,
|
||||
state="open",
|
||||
labels=labels,
|
||||
title=title or f"issue {number}",
|
||||
body=body,
|
||||
priority=priority,
|
||||
)
|
||||
|
||||
|
||||
class ClassifyEpicContainerTest(unittest.TestCase):
|
||||
def test_631_shaped_body_and_title_is_container(self) -> None:
|
||||
c = _issue(
|
||||
631,
|
||||
title="Epic: MCP Control Plane Web Console",
|
||||
body=_EPIC_631_BODY,
|
||||
)
|
||||
is_c, detail = classify_epic_or_child_only_container(c)
|
||||
self.assertTrue(is_c)
|
||||
self.assertIsNotNone(detail)
|
||||
self.assertIn("body_marker", detail or "")
|
||||
|
||||
def test_body_markers_without_epic_title(self) -> None:
|
||||
c = _issue(
|
||||
900,
|
||||
title="Control plane roadmap tracker",
|
||||
body="Implementation is delivered via child issues only.",
|
||||
)
|
||||
is_c, _ = classify_epic_or_child_only_container(c)
|
||||
self.assertTrue(is_c)
|
||||
|
||||
def test_epic_label_alone_is_container(self) -> None:
|
||||
c = _issue(
|
||||
901,
|
||||
title="Roadmap linkage",
|
||||
body="Track children.",
|
||||
labels=("status:ready", "type:epic"),
|
||||
)
|
||||
is_c, detail = classify_epic_or_child_only_container(c)
|
||||
self.assertTrue(is_c)
|
||||
self.assertIn("type:epic", detail or "")
|
||||
|
||||
def test_title_epic_prefix_alone_not_container(self) -> None:
|
||||
"""Title-only 'Epic:' without body scope evidence stays eligible (#844)."""
|
||||
c = _issue(
|
||||
902,
|
||||
title="Epic: something mentioned only in title",
|
||||
body="Implement a concrete fix for the allocator skip list.",
|
||||
)
|
||||
is_c, detail = classify_epic_or_child_only_container(c)
|
||||
self.assertFalse(is_c)
|
||||
self.assertIsNone(detail)
|
||||
|
||||
def test_incidental_epic_word_not_container(self) -> None:
|
||||
c = _issue(
|
||||
903,
|
||||
title="Document epic handoff conventions",
|
||||
body=(
|
||||
"Update the docs so implementable issues that mention an epic "
|
||||
"remain independently executable."
|
||||
),
|
||||
)
|
||||
is_c, _ = classify_epic_or_child_only_container(c)
|
||||
self.assertFalse(is_c)
|
||||
|
||||
def test_prs_never_classified(self) -> None:
|
||||
pr = WorkCandidate(
|
||||
kind="pr",
|
||||
number=10,
|
||||
state="open",
|
||||
title="Epic: fake",
|
||||
body="Implementation is delivered via child issues only.",
|
||||
head_sha="a" * 40,
|
||||
priority=5,
|
||||
)
|
||||
is_c, _ = classify_epic_or_child_only_container(pr)
|
||||
self.assertFalse(is_c)
|
||||
|
||||
|
||||
class AllocateEpicContainerExclusionTest(unittest.TestCase):
|
||||
def setUp(self) -> None:
|
||||
self._tmp = tempfile.TemporaryDirectory()
|
||||
self.addCleanup(self._tmp.cleanup)
|
||||
self.db = ControlPlaneDB(os.path.join(self._tmp.name, "cp.sqlite3"))
|
||||
|
||||
def _alloc(self, candidates, **kwargs):
|
||||
defaults = dict(
|
||||
session_id="sess-844",
|
||||
role="author",
|
||||
remote=REMOTE,
|
||||
org=ORG,
|
||||
repo=REPO,
|
||||
profile_name="prgs-author",
|
||||
username="jcwalker3",
|
||||
claims={},
|
||||
apply=False,
|
||||
)
|
||||
defaults.update(kwargs)
|
||||
return allocate_next_work(self.db, candidates=candidates, **defaults)
|
||||
|
||||
def test_631_shaped_epic_excluded_child_selected(self) -> None:
|
||||
epic = _issue(
|
||||
631,
|
||||
title="Epic: MCP Control Plane Web Console",
|
||||
body=_EPIC_631_BODY,
|
||||
)
|
||||
child = _issue(
|
||||
637,
|
||||
title="Web Console: Workflow-event timeline model (Phase 1)",
|
||||
body=_CHILD_BODY,
|
||||
)
|
||||
res = self._alloc([epic, child], apply=False)
|
||||
self.assertTrue(res["success"], res)
|
||||
self.assertEqual(res["outcome"], OUTCOME_PREVIEW)
|
||||
self.assertEqual(res["selected"]["number"], 637)
|
||||
skipped = {s["number"]: s for s in res["skipped"]}
|
||||
self.assertIn(631, skipped)
|
||||
self.assertEqual(
|
||||
skipped[631]["reason_code"], SKIP_EPIC_OR_CHILD_ONLY_CONTAINER
|
||||
)
|
||||
self.assertIn(SKIP_EPIC_OR_CHILD_ONLY_CONTAINER, skipped[631]["reason"])
|
||||
|
||||
def test_container_cannot_receive_assignment_or_lease(self) -> None:
|
||||
epic = _issue(
|
||||
631,
|
||||
title="Epic: MCP Control Plane Web Console",
|
||||
body=_EPIC_631_BODY,
|
||||
)
|
||||
res = self._alloc([epic], apply=True)
|
||||
self.assertTrue(res["success"], res)
|
||||
# Only container present → no safe work; never assigned_work.
|
||||
self.assertNotEqual(res["outcome"], OUTCOME_ASSIGNED)
|
||||
self.assertIsNone(res.get("assignment"))
|
||||
self.assertIsNone(res.get("selected"))
|
||||
skipped = {s["number"]: s for s in res["skipped"]}
|
||||
self.assertEqual(
|
||||
skipped[631]["reason_code"], SKIP_EPIC_OR_CHILD_ONLY_CONTAINER
|
||||
)
|
||||
# No lease row for the epic.
|
||||
leases = self.db.list_active_leases(
|
||||
remote=REMOTE, org=ORG, repo=REPO
|
||||
) if hasattr(self.db, "list_active_leases") else []
|
||||
# Prefer generic inventory if available.
|
||||
if not leases and hasattr(self.db, "list_leases"):
|
||||
leases = self.db.list_leases(remote=REMOTE, org=ORG, repo=REPO)
|
||||
for lease in leases or []:
|
||||
work_number = lease.get("work_number") if isinstance(lease, dict) else None
|
||||
self.assertNotEqual(work_number, 631)
|
||||
|
||||
def test_incidental_epic_title_remains_eligible(self) -> None:
|
||||
ordinary = _issue(
|
||||
700,
|
||||
title="Document epic handoff conventions",
|
||||
body="Write runbook text about epic vs child issues.",
|
||||
)
|
||||
res = self._alloc([ordinary], apply=False)
|
||||
self.assertTrue(res["success"], res)
|
||||
self.assertEqual(res["selected"]["number"], 700)
|
||||
self.assertEqual(res["skipped"], [])
|
||||
|
||||
def test_apply_selects_child_not_epic(self) -> None:
|
||||
epic = _issue(
|
||||
631,
|
||||
title="Epic: MCP Control Plane Web Console",
|
||||
body=_EPIC_631_BODY,
|
||||
)
|
||||
child = _issue(
|
||||
637,
|
||||
title="Web Console: Workflow-event timeline model (Phase 1)",
|
||||
body=_CHILD_BODY,
|
||||
)
|
||||
res = self._alloc([epic, child], apply=True)
|
||||
self.assertTrue(res["success"], res)
|
||||
self.assertEqual(res["outcome"], OUTCOME_ASSIGNED)
|
||||
self.assertEqual(res["selected"]["number"], 637)
|
||||
self.assertEqual(res["assignment"]["work_number"], 637)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -1266,6 +1266,378 @@ class TestSecondRemediationIntegration(unittest.TestCase):
|
||||
self.assertIn("delete_acknowledged", delete_actions[0])
|
||||
self.assertTrue(delete_actions[0].get("verified_absent"))
|
||||
|
||||
def test_issue_851_worktree_removed_when_remote_blocked_only_by_worktree_binding(self):
|
||||
"""#851: remote blocked by worktree_binding must not skip safe worktree removal.
|
||||
|
||||
Lifecycle: remove clean owned worktree → reassess ownership → delete
|
||||
remote only if independently safe. Unrelated entries stay untouched.
|
||||
"""
|
||||
from mcp_server import gitea_reconcile_merged_cleanups
|
||||
|
||||
target_branch = "fix/issue-844-exclude-epic-containers"
|
||||
foreign_branch = "fix/issue-999-unrelated-active"
|
||||
worktree_path = "/tmp/branches/fix-issue-844-exclude-epic-containers"
|
||||
ownership_calls = []
|
||||
remove_calls = []
|
||||
delete_api_calls = []
|
||||
|
||||
def fake_collect(**kwargs):
|
||||
ownership_calls.append(dict(kwargs))
|
||||
# Ownership is reassessed *after* independent worktree removal (#851).
|
||||
# Target worktree is already gone → no worktree_binding remains.
|
||||
# Foreign branch keeps an active author lease → remote delete blocked.
|
||||
if kwargs.get("branch") == foreign_branch:
|
||||
# Match session-bound org/repo + host used by the tool resolve path.
|
||||
return {
|
||||
"records": [
|
||||
{
|
||||
"category": guard.OWNERSHIP_CATEGORY_AUTHOR_LEASE,
|
||||
"status": "active",
|
||||
"remote": kwargs.get("remote") or "prgs",
|
||||
"host": kwargs.get("host") or "gitea.example.com",
|
||||
"org": kwargs.get("org") or "Scaled-Tech-Consulting",
|
||||
"repo": kwargs.get("repo") or "Gitea-Tools",
|
||||
"branch": foreign_branch,
|
||||
"reclaim_allowed": False,
|
||||
}
|
||||
],
|
||||
"inventory_error": False,
|
||||
}
|
||||
return {"records": [], "inventory_error": False}
|
||||
|
||||
def fake_remove(project_root, branch, worktree_path=None):
|
||||
remove_calls.append(
|
||||
{"branch": branch, "worktree_path": worktree_path}
|
||||
)
|
||||
return {
|
||||
"success": True,
|
||||
"performed": True,
|
||||
"message": f"removed worktree {worktree_path}",
|
||||
"worktree_path": worktree_path,
|
||||
}
|
||||
|
||||
def fake_probe(h, o, r, auth, br):
|
||||
return guard.classify_branch_readback_http_status(
|
||||
404, not_found_scope=guard.NOT_FOUND_SCOPE_BRANCH
|
||||
)
|
||||
|
||||
def fake_api(method, url, auth, **kwargs):
|
||||
if method == "DELETE":
|
||||
delete_api_calls.append(url)
|
||||
return {}
|
||||
|
||||
report = {
|
||||
"entries": [
|
||||
{
|
||||
"pr_number": 848,
|
||||
"head_branch": target_branch,
|
||||
"remote_branch": {"safe_to_delete_remote": True},
|
||||
"local_worktree": {
|
||||
"safe_to_remove_worktree": True,
|
||||
"worktree_path": worktree_path,
|
||||
},
|
||||
},
|
||||
{
|
||||
"pr_number": 999,
|
||||
"head_branch": foreign_branch,
|
||||
"remote_branch": {"safe_to_delete_remote": True},
|
||||
"local_worktree": {
|
||||
"safe_to_remove_worktree": False,
|
||||
"worktree_path": None,
|
||||
},
|
||||
},
|
||||
],
|
||||
"reviewer_scratch_entries": [],
|
||||
}
|
||||
patch(
|
||||
"mcp_server.get_profile",
|
||||
return_value={
|
||||
"profile_name": "prgs-reconciler",
|
||||
"role": "reconciler",
|
||||
"allowed_operations": [
|
||||
"gitea.read",
|
||||
"gitea.branch.delete",
|
||||
"gitea.pr.close",
|
||||
],
|
||||
"forbidden_operations": [],
|
||||
},
|
||||
).start()
|
||||
patch("mcp_server.api_get_all", return_value=[]).start()
|
||||
patch(
|
||||
"mcp_server.merged_cleanup_reconcile.build_reconciliation_report",
|
||||
return_value=report,
|
||||
).start()
|
||||
patch(
|
||||
"mcp_server.merged_cleanup_reconcile.discover_reviewer_scratch_worktrees",
|
||||
return_value=[],
|
||||
).start()
|
||||
patch(
|
||||
"mcp_server.audit_reconciliation_mode.check_cleanup_execution_allowed",
|
||||
return_value=(True, []),
|
||||
).start()
|
||||
patch("mcp_server.verify_preflight_purity", return_value=None).start()
|
||||
patch(
|
||||
"mcp_server._collect_branch_ownership_records",
|
||||
side_effect=fake_collect,
|
||||
).start()
|
||||
patch("mcp_server._probe_remote_branch", side_effect=fake_probe).start()
|
||||
patch(
|
||||
"mcp_server.merged_cleanup_reconcile.remove_local_worktree",
|
||||
side_effect=fake_remove,
|
||||
).start()
|
||||
self.mock_api.side_effect = fake_api
|
||||
|
||||
res = gitea_reconcile_merged_cleanups(
|
||||
dry_run=False,
|
||||
execute_confirmed=True,
|
||||
remote="prgs",
|
||||
)
|
||||
self.assertTrue(res.get("performed") or res.get("executed"))
|
||||
actions = res.get("actions") or []
|
||||
|
||||
remove_actions = [
|
||||
a for a in actions if a.get("action") == "remove_local_worktree"
|
||||
]
|
||||
self.assertEqual(len(remove_actions), 1, actions)
|
||||
self.assertTrue(remove_actions[0].get("success"))
|
||||
self.assertEqual(remove_calls[0]["branch"], target_branch)
|
||||
self.assertEqual(remove_calls[0]["worktree_path"], worktree_path)
|
||||
|
||||
# Target remote delete succeeds after worktree removal + reassessment.
|
||||
target_deletes = [
|
||||
a
|
||||
for a in actions
|
||||
if a.get("action") == "delete_remote_branch"
|
||||
and a.get("branch") == target_branch
|
||||
]
|
||||
self.assertEqual(len(target_deletes), 1, actions)
|
||||
self.assertTrue(target_deletes[0].get("success"))
|
||||
self.assertTrue(target_deletes[0].get("after_worktree_removal"))
|
||||
self.assertTrue(target_deletes[0].get("ownership_reassessed"))
|
||||
self.assertTrue(target_deletes[0].get("verified_absent"))
|
||||
|
||||
# Foreign branch remains protected (author lease) and is not deleted.
|
||||
foreign_deletes = [
|
||||
a
|
||||
for a in actions
|
||||
if a.get("action") == "delete_remote_branch"
|
||||
and a.get("branch") == foreign_branch
|
||||
]
|
||||
self.assertEqual(len(foreign_deletes), 1, actions)
|
||||
self.assertFalse(foreign_deletes[0].get("success"))
|
||||
self.assertEqual(
|
||||
foreign_deletes[0].get("blocker_kind"), "active_branch_ownership"
|
||||
)
|
||||
self.assertIn(
|
||||
guard.OWNERSHIP_CATEGORY_AUTHOR_LEASE,
|
||||
foreign_deletes[0].get("blocking_categories") or [],
|
||||
)
|
||||
# Only the target branch should hit the DELETE API.
|
||||
self.assertEqual(len(delete_api_calls), 1)
|
||||
|
||||
# Ownership collected for target (post-removal) and foreign; worktree
|
||||
# removal happened before target remote delete in the action log.
|
||||
target_idx = next(
|
||||
i
|
||||
for i, a in enumerate(actions)
|
||||
if a.get("action") == "remove_local_worktree"
|
||||
)
|
||||
delete_idx = next(
|
||||
i
|
||||
for i, a in enumerate(actions)
|
||||
if a.get("action") == "delete_remote_branch"
|
||||
and a.get("branch") == target_branch
|
||||
and a.get("success")
|
||||
)
|
||||
self.assertLess(target_idx, delete_idx)
|
||||
|
||||
def test_issue_851_dirty_worktree_not_removed_and_remote_stays_protected(self):
|
||||
"""#851: dirty/foreign worktrees remain protected; no unsafe cleanup."""
|
||||
from mcp_server import gitea_reconcile_merged_cleanups
|
||||
|
||||
branch = "fix/issue-851-dirty"
|
||||
remove_calls = []
|
||||
|
||||
def fake_collect(**kwargs):
|
||||
return {
|
||||
"records": [
|
||||
{
|
||||
"category": guard.OWNERSHIP_CATEGORY_WORKTREE_BINDING,
|
||||
"status": "active",
|
||||
"remote": kwargs.get("remote") or "prgs",
|
||||
"host": kwargs.get("host") or "gitea.example.com",
|
||||
"org": kwargs.get("org") or "Scaled-Tech-Consulting",
|
||||
"repo": kwargs.get("repo") or "Gitea-Tools",
|
||||
"branch": branch,
|
||||
"reclaim_allowed": False,
|
||||
}
|
||||
],
|
||||
"inventory_error": False,
|
||||
}
|
||||
|
||||
report = {
|
||||
"entries": [
|
||||
{
|
||||
"pr_number": 851,
|
||||
"head_branch": branch,
|
||||
"remote_branch": {"safe_to_delete_remote": True},
|
||||
"local_worktree": {
|
||||
"safe_to_remove_worktree": False,
|
||||
"worktree_path": "/tmp/dirty-wt",
|
||||
},
|
||||
}
|
||||
],
|
||||
"reviewer_scratch_entries": [],
|
||||
}
|
||||
patch(
|
||||
"mcp_server.get_profile",
|
||||
return_value={
|
||||
"profile_name": "prgs-reconciler",
|
||||
"role": "reconciler",
|
||||
"allowed_operations": [
|
||||
"gitea.read",
|
||||
"gitea.branch.delete",
|
||||
],
|
||||
"forbidden_operations": [],
|
||||
},
|
||||
).start()
|
||||
patch("mcp_server.api_get_all", return_value=[]).start()
|
||||
patch(
|
||||
"mcp_server.merged_cleanup_reconcile.build_reconciliation_report",
|
||||
return_value=report,
|
||||
).start()
|
||||
patch(
|
||||
"mcp_server.merged_cleanup_reconcile.discover_reviewer_scratch_worktrees",
|
||||
return_value=[],
|
||||
).start()
|
||||
patch(
|
||||
"mcp_server.audit_reconciliation_mode.check_cleanup_execution_allowed",
|
||||
return_value=(True, []),
|
||||
).start()
|
||||
patch("mcp_server.verify_preflight_purity", return_value=None).start()
|
||||
patch(
|
||||
"mcp_server._collect_branch_ownership_records",
|
||||
side_effect=fake_collect,
|
||||
).start()
|
||||
patch(
|
||||
"mcp_server.merged_cleanup_reconcile.remove_local_worktree",
|
||||
side_effect=lambda *a, **k: remove_calls.append(k) or {
|
||||
"success": True,
|
||||
"performed": True,
|
||||
},
|
||||
).start()
|
||||
self.mock_api.side_effect = lambda *a, **k: {}
|
||||
|
||||
res = gitea_reconcile_merged_cleanups(
|
||||
dry_run=False,
|
||||
execute_confirmed=True,
|
||||
remote="prgs",
|
||||
)
|
||||
actions = res.get("actions") or []
|
||||
self.assertEqual(remove_calls, [])
|
||||
self.assertFalse(
|
||||
any(a.get("action") == "remove_local_worktree" for a in actions)
|
||||
)
|
||||
deletes = [
|
||||
a for a in actions if a.get("action") == "delete_remote_branch"
|
||||
]
|
||||
self.assertEqual(len(deletes), 1)
|
||||
self.assertFalse(deletes[0].get("success"))
|
||||
self.assertEqual(deletes[0].get("blocker_kind"), "active_branch_ownership")
|
||||
self.assertIn(
|
||||
guard.OWNERSHIP_CATEGORY_WORKTREE_BINDING,
|
||||
deletes[0].get("blocking_categories") or [],
|
||||
)
|
||||
|
||||
def test_issue_851_idempotent_resume_when_worktree_already_absent(self):
|
||||
"""#851: partial failures remain resumable and idempotent."""
|
||||
from mcp_server import gitea_reconcile_merged_cleanups
|
||||
|
||||
branch = "fix/issue-851-resume"
|
||||
ownership_calls = []
|
||||
|
||||
def fake_collect(**kwargs):
|
||||
ownership_calls.append(kwargs)
|
||||
return {"records": [], "inventory_error": False}
|
||||
|
||||
def fake_remove(project_root, branch, worktree_path=None):
|
||||
return {
|
||||
"success": False,
|
||||
"performed": False,
|
||||
"message": f"worktree not found: {worktree_path}",
|
||||
}
|
||||
|
||||
def fake_probe(h, o, r, auth, br):
|
||||
return guard.classify_branch_readback_http_status(
|
||||
404, not_found_scope=guard.NOT_FOUND_SCOPE_BRANCH
|
||||
)
|
||||
|
||||
report = {
|
||||
"entries": [
|
||||
{
|
||||
"pr_number": 851,
|
||||
"head_branch": branch,
|
||||
"remote_branch": {"safe_to_delete_remote": True},
|
||||
"local_worktree": {
|
||||
"safe_to_remove_worktree": True,
|
||||
"worktree_path": "/tmp/already-gone",
|
||||
},
|
||||
}
|
||||
],
|
||||
"reviewer_scratch_entries": [],
|
||||
}
|
||||
patch(
|
||||
"mcp_server.get_profile",
|
||||
return_value={
|
||||
"profile_name": "prgs-reconciler",
|
||||
"role": "reconciler",
|
||||
"allowed_operations": [
|
||||
"gitea.read",
|
||||
"gitea.branch.delete",
|
||||
],
|
||||
"forbidden_operations": [],
|
||||
},
|
||||
).start()
|
||||
patch("mcp_server.api_get_all", return_value=[]).start()
|
||||
patch(
|
||||
"mcp_server.merged_cleanup_reconcile.build_reconciliation_report",
|
||||
return_value=report,
|
||||
).start()
|
||||
patch(
|
||||
"mcp_server.merged_cleanup_reconcile.discover_reviewer_scratch_worktrees",
|
||||
return_value=[],
|
||||
).start()
|
||||
patch(
|
||||
"mcp_server.audit_reconciliation_mode.check_cleanup_execution_allowed",
|
||||
return_value=(True, []),
|
||||
).start()
|
||||
patch("mcp_server.verify_preflight_purity", return_value=None).start()
|
||||
patch(
|
||||
"mcp_server._collect_branch_ownership_records",
|
||||
side_effect=fake_collect,
|
||||
).start()
|
||||
patch("mcp_server._probe_remote_branch", side_effect=fake_probe).start()
|
||||
patch(
|
||||
"mcp_server.merged_cleanup_reconcile.remove_local_worktree",
|
||||
side_effect=fake_remove,
|
||||
).start()
|
||||
self.mock_api.side_effect = lambda *a, **k: {}
|
||||
|
||||
res = gitea_reconcile_merged_cleanups(
|
||||
dry_run=False,
|
||||
execute_confirmed=True,
|
||||
remote="prgs",
|
||||
)
|
||||
actions = res.get("actions") or []
|
||||
removes = [a for a in actions if a.get("action") == "remove_local_worktree"]
|
||||
deletes = [a for a in actions if a.get("action") == "delete_remote_branch"]
|
||||
self.assertEqual(len(removes), 1)
|
||||
self.assertFalse(removes[0].get("success"))
|
||||
self.assertEqual(len(deletes), 1)
|
||||
self.assertTrue(deletes[0].get("success"))
|
||||
self.assertTrue(deletes[0].get("after_worktree_removal"))
|
||||
self.assertTrue(ownership_calls)
|
||||
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
|
||||
@@ -0,0 +1,682 @@
|
||||
"""Cross-role allocation handoff consumable by independent workers (#843).
|
||||
|
||||
Regression coverage for the controller→required-role consume path:
|
||||
|
||||
* controller allocates author work; independent author adopts successfully
|
||||
* author adoption succeeds after allocating controller process exits
|
||||
* author adoption without sharing controller session identity
|
||||
* wrong-role adoption rejected
|
||||
* concurrent/second adoption rejected without state corruption
|
||||
* terminal allocation adoption rejected
|
||||
* successful adoption produces authoritative ownership evidence
|
||||
* genuine abandoned-lease recovery remains valid
|
||||
* process_work_queue / allocate results include consume identifiers
|
||||
* same-role allocation behavior remains compatible
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import tempfile
|
||||
import unittest
|
||||
from datetime import timedelta
|
||||
from unittest.mock import patch
|
||||
|
||||
from allocator_service import (
|
||||
ALLOCATION_MODE_CROSS_ROLE,
|
||||
ALLOCATION_MODE_ROLE_SCOPED,
|
||||
OUTCOME_ASSIGNED,
|
||||
ROLE_AUTHOR,
|
||||
ROLE_CONTROLLER,
|
||||
ROLE_REVIEWER,
|
||||
WorkCandidate,
|
||||
allocate_next_work,
|
||||
)
|
||||
from control_plane_db import ControlPlaneDB, ForeignLeaseError, _ts, _utc_now
|
||||
import lease_lifecycle as ll
|
||||
|
||||
|
||||
class CrossRoleHandoffTest(unittest.TestCase):
|
||||
def setUp(self) -> None:
|
||||
self._tmp = tempfile.TemporaryDirectory()
|
||||
self.db_path = os.path.join(self._tmp.name, "cp.sqlite3")
|
||||
self.db = ControlPlaneDB(self.db_path)
|
||||
self.db.upsert_session(
|
||||
session_id="ctrl-session",
|
||||
role="controller",
|
||||
profile="prgs-controller",
|
||||
pid=99999999, # dead-looking pid
|
||||
)
|
||||
self.db.upsert_session(
|
||||
session_id="author-worker",
|
||||
role="author",
|
||||
profile="prgs-author",
|
||||
pid=os.getpid(),
|
||||
)
|
||||
self.db.upsert_session(
|
||||
session_id="author-worker-2",
|
||||
role="author",
|
||||
profile="prgs-author",
|
||||
pid=os.getpid(),
|
||||
)
|
||||
self.db.upsert_session(
|
||||
session_id="reviewer-worker",
|
||||
role="reviewer",
|
||||
profile="prgs-reviewer",
|
||||
pid=os.getpid(),
|
||||
)
|
||||
self.wt = self._tmp.name
|
||||
|
||||
def tearDown(self) -> None:
|
||||
self._tmp.cleanup()
|
||||
|
||||
def _ready_issue(self, number: int = 843, title: str = "handoff target") -> WorkCandidate:
|
||||
return WorkCandidate(
|
||||
kind="issue",
|
||||
number=number,
|
||||
labels=("status:ready", "type:bug"),
|
||||
title=title,
|
||||
priority=20,
|
||||
)
|
||||
|
||||
def _controller_allocate(self, number: int = 843, **kwargs):
|
||||
defaults = dict(
|
||||
db=self.db,
|
||||
session_id="ctrl-session",
|
||||
role=ROLE_CONTROLLER,
|
||||
remote="prgs",
|
||||
org="Scaled-Tech-Consulting",
|
||||
repo="Gitea-Tools",
|
||||
candidates=[self._ready_issue(number)],
|
||||
apply=True,
|
||||
profile_name="prgs-controller",
|
||||
username="controller-user",
|
||||
allocation_mode=ALLOCATION_MODE_CROSS_ROLE,
|
||||
)
|
||||
defaults.update(kwargs)
|
||||
return allocate_next_work(**defaults)
|
||||
|
||||
def test_controller_allocates_author_independent_author_adopts(self) -> None:
|
||||
res = self._controller_allocate()
|
||||
self.assertEqual(res["outcome"], OUTCOME_ASSIGNED)
|
||||
self.assertEqual(res["required_role"], ROLE_AUTHOR)
|
||||
self.assertIn("consume_allocation", res)
|
||||
consume = res["consume_allocation"]
|
||||
self.assertEqual(consume["tool"], "gitea_adopt_workflow_lease")
|
||||
self.assertEqual(consume["required_role"], ROLE_AUTHOR)
|
||||
self.assertFalse(consume["controller_session_required"])
|
||||
lid = res["assignment"]["lease_id"]
|
||||
self.assertEqual(consume["lease_id"], lid)
|
||||
self.assertIn(lid, res["next_valid_command"])
|
||||
|
||||
adopted = ll.adopt_lease(
|
||||
self.db,
|
||||
lease_id=lid,
|
||||
adopter_session_id="author-worker",
|
||||
role=ROLE_AUTHOR,
|
||||
worktree_path=self.wt,
|
||||
)
|
||||
self.assertTrue(adopted["success"])
|
||||
self.assertEqual(adopted["outcome"], "adopted_cross_role_handoff")
|
||||
self.assertEqual(adopted["adopted_by_session_id"], "author-worker")
|
||||
self.assertEqual(adopted["adopted_from_session_id"], "ctrl-session")
|
||||
raw = adopted["read_after_write"]
|
||||
self.assertEqual(raw["session_id"], "author-worker")
|
||||
self.assertEqual(raw["adopted_by_session_id"], "author-worker")
|
||||
self.assertEqual(raw["status"], "active")
|
||||
self.assertEqual(raw["phase"], "adopted")
|
||||
|
||||
# Authoritative re-read
|
||||
state = self.db.get_lease_workflow_state(lid)
|
||||
self.assertEqual(state["lease"]["session_id"], "author-worker")
|
||||
self.assertEqual(state["lease"]["adopted_by_session_id"], "author-worker")
|
||||
self.assertEqual(state["assignment"]["session_id"], "author-worker")
|
||||
self.assertEqual(state["provenance"]["handoff_status"], "adopted")
|
||||
|
||||
def test_author_adoption_after_controller_process_exits(self) -> None:
|
||||
res = self._controller_allocate(number=900)
|
||||
lid = res["assignment"]["lease_id"]
|
||||
# Force owner_pid dead + freshness stale_dead_process
|
||||
import sqlite3
|
||||
|
||||
conn = sqlite3.connect(self.db_path)
|
||||
try:
|
||||
conn.execute(
|
||||
"UPDATE leases SET owner_pid = 99999999 WHERE lease_id = ?",
|
||||
(lid,),
|
||||
)
|
||||
conn.commit()
|
||||
finally:
|
||||
conn.close()
|
||||
state = self.db.get_lease_workflow_state(lid)
|
||||
fr = ll.classify_lease_freshness(
|
||||
state["lease"], pid_checker=lambda _p: False
|
||||
)
|
||||
self.assertEqual(fr["freshness"], "stale_dead_process")
|
||||
|
||||
adopted = ll.adopt_lease(
|
||||
self.db,
|
||||
lease_id=lid,
|
||||
adopter_session_id="author-worker",
|
||||
role=ROLE_AUTHOR,
|
||||
worktree_path=self.wt,
|
||||
)
|
||||
self.assertEqual(adopted["outcome"], "adopted_cross_role_handoff")
|
||||
self.assertEqual(adopted["adopted_by_session_id"], "author-worker")
|
||||
# No abandon required
|
||||
state2 = self.db.get_lease_workflow_state(lid)
|
||||
self.assertEqual(state2["lease"]["status"], "active")
|
||||
self.assertNotEqual(state2["lease"]["status"], "abandoned")
|
||||
|
||||
def test_adoption_without_sharing_controller_session_identity(self) -> None:
|
||||
res = self._controller_allocate(number=901)
|
||||
lid = res["assignment"]["lease_id"]
|
||||
adopted = ll.adopt_lease(
|
||||
self.db,
|
||||
lease_id=lid,
|
||||
adopter_session_id="author-worker",
|
||||
role=ROLE_AUTHOR,
|
||||
worktree_path=self.wt,
|
||||
)
|
||||
self.assertNotEqual(adopted["adopted_by_session_id"], "ctrl-session")
|
||||
self.assertFalse(adopted["same_owner"])
|
||||
self.assertEqual(adopted["adopted_from_session_id"], "ctrl-session")
|
||||
|
||||
def test_wrong_role_adoption_rejected(self) -> None:
|
||||
res = self._controller_allocate(number=902)
|
||||
lid = res["assignment"]["lease_id"]
|
||||
with self.assertRaises(ll.LeaseLifecycleError) as ctx:
|
||||
ll.adopt_lease(
|
||||
self.db,
|
||||
lease_id=lid,
|
||||
adopter_session_id="reviewer-worker",
|
||||
role=ROLE_REVIEWER,
|
||||
)
|
||||
self.assertIn("wrong role", str(ctx.exception).lower())
|
||||
# State unchanged
|
||||
state = self.db.get_lease_workflow_state(lid)
|
||||
self.assertEqual(state["lease"]["session_id"], "ctrl-session")
|
||||
self.assertIsNone(state["lease"].get("adopted_by_session_id") or None)
|
||||
self.assertEqual(state["provenance"]["handoff_status"], "pending")
|
||||
|
||||
def test_second_adoption_rejected_without_corruption(self) -> None:
|
||||
res = self._controller_allocate(number=903)
|
||||
lid = res["assignment"]["lease_id"]
|
||||
first = ll.adopt_lease(
|
||||
self.db,
|
||||
lease_id=lid,
|
||||
adopter_session_id="author-worker",
|
||||
role=ROLE_AUTHOR,
|
||||
worktree_path=self.wt,
|
||||
)
|
||||
self.assertEqual(first["outcome"], "adopted_cross_role_handoff")
|
||||
with self.assertRaises(ll.LeaseLifecycleError):
|
||||
ll.adopt_lease(
|
||||
self.db,
|
||||
lease_id=lid,
|
||||
adopter_session_id="author-worker-2",
|
||||
role=ROLE_AUTHOR,
|
||||
worktree_path=self.wt,
|
||||
)
|
||||
state = self.db.get_lease_workflow_state(lid)
|
||||
self.assertEqual(state["lease"]["session_id"], "author-worker")
|
||||
self.assertEqual(state["lease"]["adopted_by_session_id"], "author-worker")
|
||||
self.assertEqual(state["assignment"]["session_id"], "author-worker")
|
||||
self.assertEqual(state["lease"]["status"], "active")
|
||||
|
||||
def test_terminal_allocation_adoption_rejected(self) -> None:
|
||||
res = self._controller_allocate(number=904)
|
||||
lid = res["assignment"]["lease_id"]
|
||||
# Abandon as terminal
|
||||
proof = ll.AbandonProof(
|
||||
dead_process=True,
|
||||
missing_worktree=True,
|
||||
no_open_pr=True,
|
||||
no_live_mutation_risk=True,
|
||||
owner_pid=99999999,
|
||||
worktree_path="/nonexistent/for-843",
|
||||
)
|
||||
# Attach dead pid / missing wt for abandon eligibility
|
||||
import sqlite3
|
||||
|
||||
conn = sqlite3.connect(self.db_path)
|
||||
try:
|
||||
conn.execute(
|
||||
"UPDATE leases SET owner_pid = 99999999, worktree_path = ? WHERE lease_id = ?",
|
||||
("/nonexistent/for-843", lid),
|
||||
)
|
||||
conn.commit()
|
||||
finally:
|
||||
conn.close()
|
||||
abandoned = ll.abandon_lease(
|
||||
self.db,
|
||||
lease_id=lid,
|
||||
requester_session_id="author-worker",
|
||||
proof=proof,
|
||||
)
|
||||
self.assertEqual(abandoned["outcome"], "abandoned")
|
||||
with self.assertRaises(ll.LeaseLifecycleError) as ctx:
|
||||
ll.adopt_lease(
|
||||
self.db,
|
||||
lease_id=lid,
|
||||
adopter_session_id="author-worker",
|
||||
role=ROLE_AUTHOR,
|
||||
)
|
||||
self.assertIn("abandoned", str(ctx.exception).lower())
|
||||
|
||||
def test_successful_adoption_read_after_write_ownership(self) -> None:
|
||||
res = self._controller_allocate(number=905)
|
||||
lid = res["assignment"]["lease_id"]
|
||||
adopted = ll.adopt_lease(
|
||||
self.db,
|
||||
lease_id=lid,
|
||||
adopter_session_id="author-worker",
|
||||
role=ROLE_AUTHOR,
|
||||
worktree_path=self.wt,
|
||||
)
|
||||
raw = adopted["read_after_write"]
|
||||
self.assertEqual(raw["lease_id"], lid)
|
||||
self.assertEqual(raw["session_id"], "author-worker")
|
||||
self.assertEqual(raw["adopted_by_session_id"], "author-worker")
|
||||
self.assertEqual(raw["adopted_from_session_id"], "ctrl-session")
|
||||
# Re-fetch proves durable write
|
||||
state = self.db.get_lease_workflow_state(lid)
|
||||
self.assertEqual(state["lease"]["session_id"], raw["session_id"])
|
||||
self.assertEqual(
|
||||
state["lease"]["adopted_by_session_id"], raw["adopted_by_session_id"]
|
||||
)
|
||||
|
||||
def test_genuine_abandoned_recovery_still_valid(self) -> None:
|
||||
"""Same-role author lease abandoned remains reclaimable via abandon path."""
|
||||
same = allocate_next_work(
|
||||
self.db,
|
||||
session_id="author-worker",
|
||||
role=ROLE_AUTHOR,
|
||||
remote="prgs",
|
||||
org="Scaled-Tech-Consulting",
|
||||
repo="Gitea-Tools",
|
||||
candidates=[self._ready_issue(906, "same-role")],
|
||||
apply=True,
|
||||
profile_name="prgs-author",
|
||||
username="author-user",
|
||||
allocation_mode=ALLOCATION_MODE_ROLE_SCOPED,
|
||||
)
|
||||
self.assertEqual(same["outcome"], OUTCOME_ASSIGNED)
|
||||
lid = same["assignment"]["lease_id"]
|
||||
import sqlite3
|
||||
|
||||
conn = sqlite3.connect(self.db_path)
|
||||
try:
|
||||
conn.execute(
|
||||
"UPDATE leases SET owner_pid = 99999999, worktree_path = ? WHERE lease_id = ?",
|
||||
("/nonexistent/same-role", lid),
|
||||
)
|
||||
conn.commit()
|
||||
finally:
|
||||
conn.close()
|
||||
proof = ll.AbandonProof(
|
||||
dead_process=True,
|
||||
missing_worktree=True,
|
||||
no_open_pr=True,
|
||||
no_live_mutation_risk=True,
|
||||
owner_pid=99999999,
|
||||
worktree_path="/nonexistent/same-role",
|
||||
)
|
||||
abandoned = ll.abandon_lease(
|
||||
self.db,
|
||||
lease_id=lid,
|
||||
requester_session_id="author-worker-2",
|
||||
proof=proof,
|
||||
)
|
||||
self.assertEqual(abandoned["outcome"], "abandoned")
|
||||
# Foreign author cannot handoff-consume an abandoned non-handoff lease
|
||||
with self.assertRaises(ll.LeaseLifecycleError):
|
||||
ll.adopt_lease(
|
||||
self.db,
|
||||
lease_id=lid,
|
||||
adopter_session_id="author-worker-2",
|
||||
role=ROLE_AUTHOR,
|
||||
)
|
||||
# Reclaim path still works for expired/abandoned after force-expire
|
||||
reclaimed = ll.reclaim_expired_lease(
|
||||
self.db,
|
||||
lease_id=lid,
|
||||
session_id="author-worker-2",
|
||||
role=ROLE_AUTHOR,
|
||||
worktree_path=self.wt,
|
||||
)
|
||||
self.assertEqual(reclaimed["outcome"], "reclaimed")
|
||||
self.assertEqual(reclaimed["assignment"]["session_id"], "author-worker-2")
|
||||
|
||||
def test_allocate_payload_includes_consume_identifiers(self) -> None:
|
||||
res = self._controller_allocate(number=907)
|
||||
self.assertIn("consume_allocation", res)
|
||||
c = res["consume_allocation"]
|
||||
for key in (
|
||||
"tool",
|
||||
"lease_id",
|
||||
"assignment_id",
|
||||
"required_role",
|
||||
"required_profile",
|
||||
"required_namespace",
|
||||
"instructions",
|
||||
"handoff_status",
|
||||
):
|
||||
self.assertIn(key, c)
|
||||
self.assertEqual(c["required_namespace"], "gitea-author")
|
||||
self.assertEqual(c["required_profile"], "prgs-author")
|
||||
self.assertIn("gitea_adopt_workflow_lease", c["instructions"])
|
||||
self.assertTrue(res["lease_proof"]["cross_role_handoff"])
|
||||
self.assertEqual(res["lease_proof"]["handoff_status"], "pending")
|
||||
|
||||
def test_same_role_allocation_remains_compatible(self) -> None:
|
||||
res = allocate_next_work(
|
||||
self.db,
|
||||
session_id="author-worker",
|
||||
role=ROLE_AUTHOR,
|
||||
remote="prgs",
|
||||
org="Scaled-Tech-Consulting",
|
||||
repo="Gitea-Tools",
|
||||
candidates=[self._ready_issue(908)],
|
||||
apply=True,
|
||||
profile_name="prgs-author",
|
||||
username="author-user",
|
||||
)
|
||||
self.assertEqual(res["outcome"], OUTCOME_ASSIGNED)
|
||||
self.assertNotIn("consume_allocation", res)
|
||||
lid = res["assignment"]["lease_id"]
|
||||
state = self.db.get_lease_workflow_state(lid)
|
||||
# No cross-role handoff provenance
|
||||
prov = state.get("provenance") or {}
|
||||
self.assertFalse(prov.get("cross_role_handoff"))
|
||||
# Owner resume still works
|
||||
resume = ll.adopt_lease(
|
||||
self.db,
|
||||
lease_id=lid,
|
||||
adopter_session_id="author-worker",
|
||||
role=ROLE_AUTHOR,
|
||||
worktree_path=self.wt,
|
||||
)
|
||||
self.assertTrue(resume["same_owner"])
|
||||
self.assertEqual(resume["outcome"], "adopted_owner_resume")
|
||||
|
||||
def test_inspect_points_required_role_at_consume(self) -> None:
|
||||
res = self._controller_allocate(number=909)
|
||||
lid = res["assignment"]["lease_id"]
|
||||
decision = ll.inspect_lease(
|
||||
self.db, lid, caller_session_id="author-worker"
|
||||
)
|
||||
self.assertEqual(
|
||||
decision["safe_next_action"], ll.SAFE_CONSUME_CROSS_ROLE
|
||||
)
|
||||
self.assertFalse(decision["block"])
|
||||
self.assertEqual(decision["required_role"], ROLE_AUTHOR)
|
||||
|
||||
def test_db_cas_rejects_concurrent_second_consume(self) -> None:
|
||||
res = self._controller_allocate(number=910)
|
||||
lid = res["assignment"]["lease_id"]
|
||||
# First consume via DB layer directly
|
||||
first = self.db.adopt_lease(
|
||||
lease_id=lid,
|
||||
adopter_session_id="author-worker",
|
||||
role=ROLE_AUTHOR,
|
||||
worktree_path=self.wt,
|
||||
provenance={
|
||||
"cross_role_handoff": True,
|
||||
"handoff_status": "adopted",
|
||||
"required_role": "author",
|
||||
},
|
||||
)
|
||||
self.assertEqual(first["outcome"], "adopted_cross_role_handoff")
|
||||
# Second CAS must fail
|
||||
with self.assertRaises(ForeignLeaseError):
|
||||
self.db.adopt_lease(
|
||||
lease_id=lid,
|
||||
adopter_session_id="author-worker-2",
|
||||
role=ROLE_AUTHOR,
|
||||
worktree_path=self.wt,
|
||||
provenance={
|
||||
"cross_role_handoff": True,
|
||||
"handoff_status": "pending",
|
||||
"required_role": "author",
|
||||
},
|
||||
)
|
||||
state = self.db.get_lease_workflow_state(lid)
|
||||
self.assertEqual(state["lease"]["session_id"], "author-worker")
|
||||
|
||||
|
||||
class MCPBoundaryAdoptRoleBindingTest(unittest.TestCase):
|
||||
"""#843 F1: MCP-boundary role binding for ``gitea_adopt_workflow_lease``.
|
||||
|
||||
The library-level wrong-role test calls ``lease_lifecycle.adopt_lease``
|
||||
directly. These tests prove the MCP entry point derives the adopter role
|
||||
authoritatively from the active authenticated profile and rejects any
|
||||
caller-supplied role that disagrees, so a reviewer/merger profile cannot
|
||||
consume an author handoff by passing ``role="author"``.
|
||||
"""
|
||||
|
||||
AUTHOR_PROFILE = {
|
||||
"profile_name": "prgs-author",
|
||||
"role": "author",
|
||||
"allowed_operations": [
|
||||
"gitea.read",
|
||||
"gitea.pr.create",
|
||||
"gitea.branch.push",
|
||||
],
|
||||
"forbidden_operations": [],
|
||||
}
|
||||
REVIEWER_PROFILE = {
|
||||
"profile_name": "prgs-reviewer",
|
||||
"role": "reviewer",
|
||||
"allowed_operations": [
|
||||
"gitea.read",
|
||||
"gitea.pr.review",
|
||||
"gitea.pr.approve",
|
||||
"gitea.pr.request_changes",
|
||||
],
|
||||
"forbidden_operations": ["gitea.pr.create", "gitea.branch.push"],
|
||||
}
|
||||
MERGER_PROFILE = {
|
||||
"profile_name": "prgs-merger",
|
||||
"role": "merger",
|
||||
"allowed_operations": ["gitea.read", "gitea.pr.merge"],
|
||||
"forbidden_operations": ["gitea.pr.create", "gitea.branch.push"],
|
||||
}
|
||||
FOREIGN_AUTHOR_PROFILE = {
|
||||
"profile_name": "dadeschools-author",
|
||||
"role": "author",
|
||||
"allowed_operations": [
|
||||
"gitea.read",
|
||||
"gitea.pr.create",
|
||||
"gitea.branch.push",
|
||||
],
|
||||
"forbidden_operations": [],
|
||||
}
|
||||
|
||||
def setUp(self) -> None:
|
||||
self._tmp = tempfile.TemporaryDirectory()
|
||||
self.db_path = os.path.join(self._tmp.name, "cp.sqlite3")
|
||||
self.db = ControlPlaneDB(self.db_path)
|
||||
self.db.upsert_session(
|
||||
session_id="ctrl-session",
|
||||
role="controller",
|
||||
profile="prgs-controller",
|
||||
pid=99999999,
|
||||
)
|
||||
self.db.upsert_session(
|
||||
session_id="author-worker",
|
||||
role="author",
|
||||
profile="prgs-author",
|
||||
pid=os.getpid(),
|
||||
)
|
||||
self.wt = self._tmp.name
|
||||
|
||||
def tearDown(self) -> None:
|
||||
self._tmp.cleanup()
|
||||
|
||||
def _ready_issue(self, number: int) -> WorkCandidate:
|
||||
return WorkCandidate(
|
||||
kind="issue",
|
||||
number=number,
|
||||
labels=("status:ready", "type:bug"),
|
||||
title="handoff target",
|
||||
priority=20,
|
||||
)
|
||||
|
||||
def _handoff_lease(self, number: int = 843) -> str:
|
||||
res = allocate_next_work(
|
||||
db=self.db,
|
||||
session_id="ctrl-session",
|
||||
role=ROLE_CONTROLLER,
|
||||
remote="prgs",
|
||||
org="Scaled-Tech-Consulting",
|
||||
repo="Gitea-Tools",
|
||||
candidates=[self._ready_issue(number)],
|
||||
apply=True,
|
||||
profile_name="prgs-controller",
|
||||
username="controller-user",
|
||||
allocation_mode=ALLOCATION_MODE_CROSS_ROLE,
|
||||
)
|
||||
self.assertEqual(res["outcome"], OUTCOME_ASSIGNED)
|
||||
self.assertEqual(res["required_role"], ROLE_AUTHOR)
|
||||
return res["assignment"]["lease_id"]
|
||||
|
||||
def _call_adopt_tool(self, profile: dict, **kwargs):
|
||||
import gitea_mcp_server as mcp_server
|
||||
|
||||
with (
|
||||
patch.object(mcp_server, "get_profile", return_value=profile),
|
||||
patch.object(
|
||||
mcp_server,
|
||||
"_control_plane_db_or_error",
|
||||
return_value=(self.db, []),
|
||||
),
|
||||
):
|
||||
return mcp_server.gitea_adopt_workflow_lease(
|
||||
remote="prgs", **kwargs
|
||||
)
|
||||
|
||||
def _assert_handoff_untouched(self, lease_id: str) -> None:
|
||||
state = self.db.get_lease_workflow_state(lease_id)
|
||||
self.assertEqual(state["lease"]["session_id"], "ctrl-session")
|
||||
self.assertIsNone(state["lease"].get("adopted_by_session_id") or None)
|
||||
self.assertEqual(state["lease"]["status"], "active")
|
||||
self.assertEqual(state["provenance"]["handoff_status"], "pending")
|
||||
|
||||
def test_reviewer_profile_cannot_consume_author_handoff_via_role_author(
|
||||
self,
|
||||
) -> None:
|
||||
lid = self._handoff_lease(920)
|
||||
result = self._call_adopt_tool(
|
||||
self.REVIEWER_PROFILE,
|
||||
lease_id=lid,
|
||||
session_id="reviewer-worker",
|
||||
role="author",
|
||||
worktree_path=self.wt,
|
||||
)
|
||||
self.assertFalse(result["success"])
|
||||
self.assertEqual(result["outcome"], "blocked")
|
||||
self.assertEqual(result["profile_role_kind"], "reviewer")
|
||||
self.assertEqual(result["supplied_role"], "author")
|
||||
self.assertIn("does not match", result["reasons"][0])
|
||||
self._assert_handoff_untouched(lid)
|
||||
|
||||
def test_merger_profile_cannot_consume_author_handoff_via_role_author(
|
||||
self,
|
||||
) -> None:
|
||||
lid = self._handoff_lease(921)
|
||||
result = self._call_adopt_tool(
|
||||
self.MERGER_PROFILE,
|
||||
lease_id=lid,
|
||||
session_id="merger-worker",
|
||||
role="author",
|
||||
worktree_path=self.wt,
|
||||
)
|
||||
self.assertFalse(result["success"])
|
||||
self.assertEqual(result["outcome"], "blocked")
|
||||
self.assertEqual(result["profile_role_kind"], "merger")
|
||||
self._assert_handoff_untouched(lid)
|
||||
|
||||
def test_reviewer_profile_rejected_without_role_argument(self) -> None:
|
||||
"""Even without a spoofed role, the profile-derived role binds."""
|
||||
lid = self._handoff_lease(922)
|
||||
result = self._call_adopt_tool(
|
||||
self.REVIEWER_PROFILE,
|
||||
lease_id=lid,
|
||||
session_id="reviewer-worker",
|
||||
worktree_path=self.wt,
|
||||
)
|
||||
self.assertFalse(result["success"])
|
||||
self.assertEqual(result["outcome"], "blocked")
|
||||
self.assertIn("wrong role", result["reasons"][0].lower())
|
||||
self._assert_handoff_untouched(lid)
|
||||
|
||||
def test_author_profile_mismatching_supplied_role_rejected(self) -> None:
|
||||
lid = self._handoff_lease(923)
|
||||
result = self._call_adopt_tool(
|
||||
self.AUTHOR_PROFILE,
|
||||
lease_id=lid,
|
||||
session_id="author-worker",
|
||||
role="reviewer",
|
||||
worktree_path=self.wt,
|
||||
)
|
||||
self.assertFalse(result["success"])
|
||||
self.assertEqual(result["outcome"], "blocked")
|
||||
self.assertEqual(result["profile_role_kind"], "author")
|
||||
self.assertEqual(result["supplied_role"], "reviewer")
|
||||
self._assert_handoff_untouched(lid)
|
||||
|
||||
def test_foreign_profile_name_rejected_for_author_handoff(self) -> None:
|
||||
"""Provenance required_profile binds even when the role matches."""
|
||||
lid = self._handoff_lease(924)
|
||||
result = self._call_adopt_tool(
|
||||
self.FOREIGN_AUTHOR_PROFILE,
|
||||
lease_id=lid,
|
||||
session_id="foreign-author-worker",
|
||||
worktree_path=self.wt,
|
||||
)
|
||||
self.assertFalse(result["success"])
|
||||
self.assertEqual(result["outcome"], "blocked")
|
||||
self.assertIn("wrong profile", result["reasons"][0].lower())
|
||||
self._assert_handoff_untouched(lid)
|
||||
|
||||
def test_author_profile_consumes_author_handoff(self) -> None:
|
||||
lid = self._handoff_lease(925)
|
||||
result = self._call_adopt_tool(
|
||||
self.AUTHOR_PROFILE,
|
||||
lease_id=lid,
|
||||
session_id="author-worker",
|
||||
role="author",
|
||||
worktree_path=self.wt,
|
||||
)
|
||||
self.assertTrue(result["success"])
|
||||
self.assertEqual(result["outcome"], "adopted_cross_role_handoff")
|
||||
self.assertEqual(result["adopted_by_session_id"], "author-worker")
|
||||
self.assertEqual(result["adopted_from_session_id"], "ctrl-session")
|
||||
state = self.db.get_lease_workflow_state(lid)
|
||||
self.assertEqual(state["lease"]["session_id"], "author-worker")
|
||||
self.assertEqual(
|
||||
state["lease"]["adopted_by_session_id"], "author-worker"
|
||||
)
|
||||
self.assertEqual(state["assignment"]["session_id"], "author-worker")
|
||||
self.assertEqual(state["provenance"]["handoff_status"], "adopted")
|
||||
|
||||
def test_author_profile_consumes_author_handoff_without_role_argument(
|
||||
self,
|
||||
) -> None:
|
||||
lid = self._handoff_lease(926)
|
||||
result = self._call_adopt_tool(
|
||||
self.AUTHOR_PROFILE,
|
||||
lease_id=lid,
|
||||
session_id="author-worker",
|
||||
worktree_path=self.wt,
|
||||
)
|
||||
self.assertTrue(result["success"])
|
||||
self.assertEqual(result["outcome"], "adopted_cross_role_handoff")
|
||||
state = self.db.get_lease_workflow_state(lid)
|
||||
self.assertEqual(state["lease"]["session_id"], "author-worker")
|
||||
self.assertEqual(state["lease"]["role"], "author")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -12,6 +12,59 @@ import merged_cleanup_reconcile as mcr # noqa: E402
|
||||
|
||||
|
||||
class TestMergedCleanupAssessment(unittest.TestCase):
|
||||
def test_issue_851_plan_order_worktree_then_reassess_then_remote(self):
|
||||
"""#851 dry-run plan: remove worktree, reassess ownership, then remote."""
|
||||
plan = mcr.plan_cleanup_execution_order(
|
||||
remote_assessment={"safe_to_delete_remote": True},
|
||||
local_assessment={"safe_to_remove_worktree": True},
|
||||
)
|
||||
actions = [s["action"] for s in plan]
|
||||
self.assertEqual(
|
||||
actions,
|
||||
[
|
||||
"remove_local_worktree",
|
||||
"reassess_branch_ownership",
|
||||
"delete_remote_branch",
|
||||
],
|
||||
)
|
||||
self.assertEqual(plan[0]["phase"], 1)
|
||||
self.assertEqual(plan[-1]["phase"], 3)
|
||||
self.assertIn("independently_safe", plan[0]["reason"])
|
||||
self.assertIn("reassessment", plan[-1]["reason"])
|
||||
|
||||
def test_issue_851_plan_remote_only_when_worktree_not_safe(self):
|
||||
plan = mcr.plan_cleanup_execution_order(
|
||||
remote_assessment={"safe_to_delete_remote": True},
|
||||
local_assessment={"safe_to_remove_worktree": False},
|
||||
)
|
||||
self.assertEqual([s["action"] for s in plan], ["delete_remote_branch"])
|
||||
self.assertNotIn("reassess_branch_ownership", [s["action"] for s in plan])
|
||||
|
||||
def test_issue_851_plan_worktree_only_when_remote_not_safe(self):
|
||||
plan = mcr.plan_cleanup_execution_order(
|
||||
remote_assessment={"safe_to_delete_remote": False},
|
||||
local_assessment={"safe_to_remove_worktree": True},
|
||||
)
|
||||
self.assertEqual([s["action"] for s in plan], ["remove_local_worktree"])
|
||||
|
||||
def test_issue_851_entry_includes_planned_execution_order(self):
|
||||
entry = mcr.build_pr_cleanup_entry(
|
||||
pr={
|
||||
"number": 848,
|
||||
"title": "Closes #844",
|
||||
"body": "",
|
||||
"merged_at": "2026-07-23T00:00:00Z",
|
||||
"head": {"ref": "fix/issue-844-x", "sha": "a" * 40},
|
||||
},
|
||||
project_root="/tmp/not-a-real-root",
|
||||
open_pr_heads=set(),
|
||||
remote_branch_exists=True,
|
||||
head_on_master=True,
|
||||
delete_capability_allowed=True,
|
||||
)
|
||||
self.assertIn("planned_execution_order", entry)
|
||||
self.assertIsInstance(entry["planned_execution_order"], list)
|
||||
|
||||
def test_extract_linked_issue_from_closes(self):
|
||||
issue = mcr.extract_linked_issue(
|
||||
"feat: cleanup (Closes #269)",
|
||||
|
||||
@@ -0,0 +1,135 @@
|
||||
"""Tests for the Phase 1 operator console application shell (#638)."""
|
||||
import sys
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
||||
|
||||
from starlette.routing import Route
|
||||
from starlette.testclient import TestClient
|
||||
|
||||
from webui import layout
|
||||
from webui.app import create_app
|
||||
from webui.nav import NAV_GROUPS, STUB_PAGES, nav_hrefs
|
||||
|
||||
|
||||
class TestShellNav(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.client = TestClient(create_app())
|
||||
|
||||
def test_nav_group_labels_present(self):
|
||||
text = self.client.get("/").text
|
||||
for group in NAV_GROUPS:
|
||||
with self.subTest(group=group.label):
|
||||
self.assertIn(f">{group.label}<", text)
|
||||
|
||||
def test_phase1_group_labels_cover_expected_ia(self):
|
||||
labels = {group.label for group in NAV_GROUPS}
|
||||
for expected in (
|
||||
"Health",
|
||||
"Traffic",
|
||||
"Runtime/Sessions",
|
||||
"Projects",
|
||||
"Inventory",
|
||||
"Timeline",
|
||||
"Policy",
|
||||
"Insights",
|
||||
):
|
||||
with self.subTest(label=expected):
|
||||
self.assertIn(expected, labels)
|
||||
|
||||
def test_every_nav_href_resolves_to_a_get_route(self):
|
||||
app = create_app()
|
||||
get_paths = {
|
||||
route.path
|
||||
for route in app.routes
|
||||
if isinstance(route, Route) and "GET" in route.methods
|
||||
}
|
||||
for href in nav_hrefs():
|
||||
with self.subTest(href=href):
|
||||
self.assertIn(href, get_paths, f"nav href {href} has no GET route")
|
||||
|
||||
def test_legacy_hrefs_still_navigable(self):
|
||||
text = self.client.get("/").text
|
||||
for href in ("/queue", "/projects", "/prompts", "/runtime",
|
||||
"/audit", "/worktrees", "/leases", "/actions"):
|
||||
with self.subTest(href=href):
|
||||
self.assertIn(f'href="{href}"', text)
|
||||
|
||||
|
||||
class TestShellBadges(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.client = TestClient(create_app())
|
||||
|
||||
def test_mode_badge_present(self):
|
||||
self.assertIn("mode: read-only", self.client.get("/").text)
|
||||
|
||||
def test_environment_badge_present(self):
|
||||
self.assertIn("env:", self.client.get("/").text)
|
||||
|
||||
def test_default_environment_is_local(self):
|
||||
self.assertEqual(layout.environment_label(), "local")
|
||||
|
||||
def test_remote_bind_reports_remote_environment(self):
|
||||
import os
|
||||
|
||||
prior = os.environ.get("WEBUI_HOST")
|
||||
os.environ["WEBUI_HOST"] = "10.0.0.5"
|
||||
try:
|
||||
self.assertEqual(layout.environment_label(), "remote")
|
||||
finally:
|
||||
if prior is None:
|
||||
os.environ.pop("WEBUI_HOST", None)
|
||||
else:
|
||||
os.environ["WEBUI_HOST"] = prior
|
||||
|
||||
def test_docs_link_present(self):
|
||||
text = self.client.get("/").text
|
||||
self.assertIn(layout.DOCS_URL, text)
|
||||
self.assertIn(">Docs<", text)
|
||||
|
||||
|
||||
class TestShellStubs(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.client = TestClient(create_app())
|
||||
|
||||
def test_stub_routes_render_200(self):
|
||||
for path, (title, _desc) in STUB_PAGES.items():
|
||||
with self.subTest(path=path):
|
||||
response = self.client.get(path)
|
||||
self.assertEqual(response.status_code, 200, path)
|
||||
self.assertIn(title, response.text)
|
||||
self.assertIn("placeholder", response.text)
|
||||
|
||||
def test_stub_routes_are_read_only(self):
|
||||
for path in STUB_PAGES:
|
||||
with self.subTest(path=path):
|
||||
response = self.client.post(path)
|
||||
self.assertEqual(response.status_code, 405)
|
||||
self.assertEqual(response.json()["error"], "read-only-mvp")
|
||||
|
||||
def test_stub_pages_carry_nav_and_badges(self):
|
||||
response = self.client.get("/inventory")
|
||||
self.assertIn("mode: read-only", response.text)
|
||||
self.assertIn('href="/queue"', response.text)
|
||||
|
||||
|
||||
class TestShellHome(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.client = TestClient(create_app())
|
||||
|
||||
def test_home_summarizes_console(self):
|
||||
text = self.client.get("/").text
|
||||
self.assertIn("Operator console", text)
|
||||
self.assertIn("Phase 1", text)
|
||||
|
||||
def test_home_links_legacy_pages(self):
|
||||
text = self.client.get("/").text
|
||||
self.assertIn("MVP legacy pages", text)
|
||||
for href in ("/queue", "/audit", "/leases"):
|
||||
with self.subTest(href=href):
|
||||
self.assertIn(f'href="{href}"', text)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,499 @@
|
||||
"""Tests for the read-only system-health API (#634).
|
||||
|
||||
Covers the acceptance criteria directly: a structured payload with readiness
|
||||
and a dependency list (AC1), version and uptime when knowable (AC2), stale
|
||||
runtime reported without a false mutation-safe claim (AC3), and the healthy /
|
||||
degraded-dependency / redaction cases (AC4).
|
||||
"""
|
||||
import json
|
||||
import os
|
||||
import sqlite3
|
||||
import sys
|
||||
import tempfile
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
from unittest import mock
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
||||
|
||||
from starlette.testclient import TestClient
|
||||
|
||||
import control_plane_db
|
||||
from webui.app import create_app
|
||||
from webui.deployment_boundary import scan_text_for_client_secrets
|
||||
from webui.system_health import (
|
||||
API_PATH,
|
||||
STATUS_DEGRADED,
|
||||
STATUS_DOWN,
|
||||
STATUS_OK,
|
||||
STATUS_SKIPPED,
|
||||
DependencyProbe,
|
||||
StaleRuntime,
|
||||
assess_stale_runtime,
|
||||
clear_probe_cache,
|
||||
load_system_health,
|
||||
namespace_summaries,
|
||||
probe_control_plane_db,
|
||||
probe_gitea,
|
||||
process_uptime,
|
||||
redact,
|
||||
redact_url,
|
||||
snapshot_to_dict,
|
||||
)
|
||||
|
||||
|
||||
def _probe(name, status, *, required=True, detail="detail", kind="test"):
|
||||
return DependencyProbe(
|
||||
name=name,
|
||||
kind=kind,
|
||||
status=status,
|
||||
detail=detail,
|
||||
required=required,
|
||||
latency_ms=1.5,
|
||||
metadata={},
|
||||
)
|
||||
|
||||
|
||||
_ALL_HEALTHY = (
|
||||
_probe("control_plane_db", STATUS_OK, kind="sqlite"),
|
||||
_probe("repository", STATUS_OK, kind="git"),
|
||||
_probe("gitea", STATUS_OK, required=False, kind="http"),
|
||||
)
|
||||
|
||||
_CLEAN_PARITY = StaleRuntime(
|
||||
daemon_head="abc123",
|
||||
checkout_head="abc123",
|
||||
remote_head="abc123",
|
||||
stale=False,
|
||||
determinable=True,
|
||||
mutation_safe=True,
|
||||
reasons=(),
|
||||
)
|
||||
|
||||
|
||||
class CleanParityMixin:
|
||||
"""Pin parity for tests about aggregation rather than staleness.
|
||||
|
||||
Without this the assertions depend on the real checkout: a worktree whose
|
||||
branch is ahead of its upstream is genuinely stale, which would degrade the
|
||||
overall status and make these cases fail for an unrelated reason.
|
||||
"""
|
||||
|
||||
def setUp(self):
|
||||
super().setUp()
|
||||
patcher = mock.patch(
|
||||
"webui.system_health.assess_stale_runtime",
|
||||
return_value=_CLEAN_PARITY,
|
||||
)
|
||||
patcher.start()
|
||||
self.addCleanup(patcher.stop)
|
||||
|
||||
|
||||
class TestDependencyAggregation(CleanParityMixin, unittest.TestCase):
|
||||
"""AC1 — readiness and dependency list derived from probe results."""
|
||||
|
||||
def test_all_healthy_is_ok_and_ready(self):
|
||||
snapshot = load_system_health(probes=_ALL_HEALTHY, daemon_head="abc123")
|
||||
self.assertEqual(snapshot.status, STATUS_OK)
|
||||
self.assertTrue(snapshot.ready)
|
||||
self.assertTrue(snapshot.readiness_complete)
|
||||
self.assertEqual(snapshot.readiness_reasons, ())
|
||||
self.assertEqual(len(snapshot.dependencies), 3)
|
||||
|
||||
def test_required_dependency_down_blocks_readiness(self):
|
||||
probes = (
|
||||
_probe("control_plane_db", STATUS_DOWN, detail="file missing", kind="sqlite"),
|
||||
_probe("repository", STATUS_OK, kind="git"),
|
||||
_probe("gitea", STATUS_OK, required=False, kind="http"),
|
||||
)
|
||||
snapshot = load_system_health(probes=probes, daemon_head="abc123")
|
||||
self.assertEqual(snapshot.status, STATUS_DOWN)
|
||||
self.assertFalse(snapshot.ready)
|
||||
self.assertTrue(
|
||||
any("control_plane_db" in reason for reason in snapshot.readiness_reasons)
|
||||
)
|
||||
|
||||
def test_optional_dependency_down_degrades_but_stays_ready(self):
|
||||
"""A failing optional probe must not claim the process itself is unready."""
|
||||
probes = (
|
||||
_probe("control_plane_db", STATUS_OK, kind="sqlite"),
|
||||
_probe("repository", STATUS_OK, kind="git"),
|
||||
_probe("gitea", STATUS_DOWN, required=False, detail="timeout", kind="http"),
|
||||
)
|
||||
snapshot = load_system_health(probes=probes, daemon_head="abc123")
|
||||
self.assertEqual(snapshot.status, STATUS_DEGRADED)
|
||||
self.assertTrue(snapshot.ready)
|
||||
self.assertTrue(any("gitea" in reason for reason in snapshot.readiness_reasons))
|
||||
|
||||
def test_unrun_required_probe_leaves_readiness_incomplete(self):
|
||||
"""Not probed is not the same as passing."""
|
||||
probes = (
|
||||
_probe("control_plane_db", STATUS_OK, kind="sqlite"),
|
||||
_probe("repository", STATUS_SKIPPED, detail="offline", kind="git"),
|
||||
)
|
||||
snapshot = load_system_health(probes=probes, daemon_head="abc123")
|
||||
self.assertFalse(snapshot.ready)
|
||||
self.assertFalse(snapshot.readiness_complete)
|
||||
self.assertEqual(snapshot.status, STATUS_DEGRADED)
|
||||
|
||||
def test_skipped_optional_probe_does_not_block_readiness(self):
|
||||
probes = (
|
||||
_probe("control_plane_db", STATUS_OK, kind="sqlite"),
|
||||
_probe("repository", STATUS_OK, kind="git"),
|
||||
_probe("gitea", STATUS_SKIPPED, required=False, kind="http"),
|
||||
)
|
||||
snapshot = load_system_health(probes=probes, daemon_head="abc123")
|
||||
self.assertTrue(snapshot.ready)
|
||||
self.assertTrue(snapshot.readiness_complete)
|
||||
|
||||
|
||||
class TestVersionAndUptime(CleanParityMixin, unittest.TestCase):
|
||||
"""AC2 — version and uptime present when knowable."""
|
||||
|
||||
def test_uptime_and_start_time_present(self):
|
||||
snapshot = load_system_health(probes=_ALL_HEALTHY, daemon_head="abc123")
|
||||
self.assertGreaterEqual(snapshot.uptime_seconds, 0.0)
|
||||
self.assertIn("T", snapshot.started_at)
|
||||
|
||||
def test_process_uptime_helper_matches_shape(self):
|
||||
started_at, uptime = process_uptime()
|
||||
self.assertIn("T", started_at)
|
||||
self.assertGreaterEqual(uptime, 0.0)
|
||||
|
||||
def test_version_reports_python_and_schema_version(self):
|
||||
probes = (
|
||||
DependencyProbe(
|
||||
name="control_plane_db",
|
||||
kind="sqlite",
|
||||
status=STATUS_OK,
|
||||
detail="ok",
|
||||
required=True,
|
||||
latency_ms=1.0,
|
||||
metadata={"schema_version": control_plane_db.SCHEMA_VERSION},
|
||||
),
|
||||
_probe("repository", STATUS_OK, kind="git"),
|
||||
)
|
||||
snapshot = load_system_health(probes=probes, daemon_head="abc123")
|
||||
self.assertEqual(
|
||||
snapshot.version.control_plane_schema_version,
|
||||
control_plane_db.SCHEMA_VERSION,
|
||||
)
|
||||
self.assertTrue(snapshot.version.python_version)
|
||||
|
||||
def test_version_known_flag_false_when_sha_unavailable(self):
|
||||
with mock.patch("webui.system_health._git", return_value=None):
|
||||
snapshot = load_system_health(probes=_ALL_HEALTHY, daemon_head="abc")
|
||||
self.assertIsNone(snapshot.version.git_sha)
|
||||
self.assertFalse(snapshot.version.known)
|
||||
|
||||
|
||||
class TestStaleRuntime(unittest.TestCase):
|
||||
"""AC3 — stale runtime reflected without a false mutation-safe claim."""
|
||||
|
||||
def test_matching_commits_are_mutation_safe(self):
|
||||
assessment = assess_stale_runtime(
|
||||
Path("/tmp"),
|
||||
daemon_head="aaa",
|
||||
git_reader=lambda *args: "aaa",
|
||||
)
|
||||
self.assertFalse(assessment.stale)
|
||||
self.assertTrue(assessment.determinable)
|
||||
self.assertTrue(assessment.mutation_safe)
|
||||
|
||||
def test_diverged_commits_are_stale_and_not_mutation_safe(self):
|
||||
reads = {"HEAD": "aaa", "@{upstream}": "bbb"}
|
||||
assessment = assess_stale_runtime(
|
||||
Path("/tmp"),
|
||||
daemon_head="aaa",
|
||||
git_reader=lambda *args: reads.get(args[-1]),
|
||||
)
|
||||
self.assertTrue(assessment.stale)
|
||||
self.assertFalse(assessment.mutation_safe)
|
||||
self.assertTrue(assessment.reasons)
|
||||
|
||||
def test_unknown_remote_is_not_mutation_safe(self):
|
||||
"""Indeterminate must never read as safe."""
|
||||
reads = {"HEAD": "aaa", "@{upstream}": None}
|
||||
assessment = assess_stale_runtime(
|
||||
Path("/tmp"),
|
||||
daemon_head="aaa",
|
||||
git_reader=lambda *args: reads.get(args[-1]),
|
||||
)
|
||||
self.assertFalse(assessment.determinable)
|
||||
self.assertFalse(assessment.mutation_safe)
|
||||
self.assertFalse(assessment.stale)
|
||||
self.assertTrue(
|
||||
any("indeterminate" in reason for reason in assessment.reasons)
|
||||
)
|
||||
|
||||
def test_unobservable_daemon_head_is_disclosed(self):
|
||||
assessment = assess_stale_runtime(
|
||||
Path("/tmp"),
|
||||
git_reader=lambda *args: "aaa",
|
||||
)
|
||||
self.assertTrue(
|
||||
any("not observable" in reason for reason in assessment.reasons)
|
||||
)
|
||||
|
||||
def test_stale_runtime_degrades_overall_status(self):
|
||||
reads = {"HEAD": "aaa", "@{upstream}": "bbb"}
|
||||
# Pinned rather than inherited: this path uses the default git reader,
|
||||
# so the assertion must hold whether or not the suite runs offline.
|
||||
with mock.patch.dict(os.environ, {"WEBUI_TEST_OFFLINE": ""}), mock.patch(
|
||||
"webui.system_health._git",
|
||||
side_effect=lambda repo, *args: reads.get(args[-1]),
|
||||
):
|
||||
snapshot = load_system_health(probes=_ALL_HEALTHY, daemon_head="aaa")
|
||||
self.assertTrue(snapshot.stale_runtime.stale)
|
||||
self.assertFalse(snapshot.stale_runtime.mutation_safe)
|
||||
self.assertEqual(snapshot.status, STATUS_DEGRADED)
|
||||
|
||||
|
||||
class TestControlPlaneDbProbe(unittest.TestCase):
|
||||
"""The required local dependency, probed read-only."""
|
||||
|
||||
def setUp(self):
|
||||
self.tmp = tempfile.TemporaryDirectory()
|
||||
self.addCleanup(self.tmp.cleanup)
|
||||
self.db_path = str(Path(self.tmp.name) / "control-plane.db")
|
||||
|
||||
def _build_db(self, schema_version):
|
||||
conn = sqlite3.connect(self.db_path)
|
||||
conn.execute("CREATE TABLE schema_meta (key TEXT PRIMARY KEY, value TEXT)")
|
||||
conn.execute("CREATE TABLE leases (lease_id TEXT PRIMARY KEY, status TEXT)")
|
||||
conn.execute(
|
||||
"INSERT INTO schema_meta(key, value) VALUES ('schema_version', ?)",
|
||||
(str(schema_version),),
|
||||
)
|
||||
conn.execute("INSERT INTO leases(lease_id, status) VALUES ('l1', 'active')")
|
||||
conn.commit()
|
||||
conn.close()
|
||||
|
||||
def test_missing_database_is_down(self):
|
||||
probe = probe_control_plane_db(str(Path(self.tmp.name) / "absent.db"))
|
||||
self.assertEqual(probe.status, STATUS_DOWN)
|
||||
self.assertTrue(probe.required)
|
||||
self.assertIsNotNone(probe.latency_ms)
|
||||
|
||||
def test_matching_schema_is_ok(self):
|
||||
self._build_db(control_plane_db.SCHEMA_VERSION)
|
||||
probe = probe_control_plane_db(self.db_path)
|
||||
self.assertEqual(probe.status, STATUS_OK)
|
||||
self.assertEqual(
|
||||
probe.metadata["schema_version"], control_plane_db.SCHEMA_VERSION
|
||||
)
|
||||
self.assertEqual(probe.metadata["active_leases"], 1)
|
||||
|
||||
def test_mismatched_schema_is_degraded(self):
|
||||
self._build_db(control_plane_db.SCHEMA_VERSION + 99)
|
||||
probe = probe_control_plane_db(self.db_path)
|
||||
self.assertEqual(probe.status, STATUS_DEGRADED)
|
||||
|
||||
def test_probe_does_not_create_a_database(self):
|
||||
"""A health check must never initialise the substrate it inspects."""
|
||||
absent = str(Path(self.tmp.name) / "never-created.db")
|
||||
probe_control_plane_db(absent)
|
||||
self.assertFalse(Path(absent).exists())
|
||||
|
||||
def test_unreadable_database_is_down_not_raised(self):
|
||||
Path(self.db_path).write_text("this is not a sqlite database")
|
||||
probe = probe_control_plane_db(self.db_path)
|
||||
self.assertEqual(probe.status, STATUS_DOWN)
|
||||
|
||||
|
||||
class TestRedaction(unittest.TestCase):
|
||||
"""AC4 — redaction. No credential-shaped text crosses the boundary."""
|
||||
|
||||
def test_redacts_token_assignment(self):
|
||||
cleaned = redact("failed with token=ghp_ABCDEFGHIJKLMNOPQRSTUVWXYZ012345")
|
||||
self.assertNotIn("ghp_ABCDEFGHIJKLMNOPQRSTUVWXYZ012345", cleaned)
|
||||
self.assertIn("[redacted]", cleaned)
|
||||
|
||||
def test_redacts_authorization_header_text(self):
|
||||
cleaned = redact("Authorization: Bearer abcdefghijklmnopqrstuvwxyz123456")
|
||||
self.assertNotIn("abcdefghijklmnopqrstuvwxyz123456", cleaned)
|
||||
|
||||
def test_redacts_long_opaque_strings(self):
|
||||
cleaned = redact("value 0123456789abcdef0123456789abcdef here")
|
||||
self.assertNotIn("0123456789abcdef0123456789abcdef", cleaned)
|
||||
|
||||
def test_url_userinfo_and_query_are_stripped(self):
|
||||
cleaned = redact_url("https://user:[email protected]/api/v1?token=xyz")
|
||||
self.assertNotIn("secretpass", cleaned)
|
||||
self.assertNotIn("token=xyz", cleaned)
|
||||
self.assertEqual(cleaned, "https://gitea.example.com/api/v1")
|
||||
|
||||
def test_url_inside_free_text_is_redacted(self):
|
||||
cleaned = redact("GET https://u:[email protected]/x?token=abc failed")
|
||||
self.assertNotIn("u:p@", cleaned)
|
||||
self.assertNotIn("token=abc", cleaned)
|
||||
|
||||
def test_gitea_probe_failure_detail_is_redacted(self):
|
||||
boom = RuntimeError(
|
||||
"connection refused for https://user:[email protected]/api/v1/version"
|
||||
)
|
||||
with mock.patch("webui.system_health.get_auth_header", return_value="token x"), \
|
||||
mock.patch("webui.system_health.api_request", side_effect=boom):
|
||||
probe = probe_gitea("gitea.example.com")
|
||||
self.assertEqual(probe.status, STATUS_DOWN)
|
||||
self.assertNotIn("hunter2", probe.detail)
|
||||
self.assertEqual(scan_text_for_client_secrets(probe.detail), [])
|
||||
|
||||
def test_credential_guard_refusal_is_a_status_not_a_crash(self):
|
||||
with mock.patch(
|
||||
"webui.system_health.get_auth_header",
|
||||
side_effect=RuntimeError("daemon guard refused"),
|
||||
):
|
||||
probe = probe_gitea("gitea.example.com")
|
||||
self.assertEqual(probe.status, STATUS_DEGRADED)
|
||||
self.assertFalse(probe.required)
|
||||
|
||||
|
||||
class TestNamespaceSummaries(unittest.TestCase):
|
||||
"""A web process cannot prove IDE namespace health, and must not claim to."""
|
||||
|
||||
def test_every_namespace_reports_unproven(self):
|
||||
rows = namespace_summaries()
|
||||
self.assertTrue(rows)
|
||||
for row in rows:
|
||||
with self.subTest(namespace=row["namespace"]):
|
||||
self.assertEqual(row["status"], "unproven")
|
||||
self.assertFalse(row["ide_namespace_proven"])
|
||||
self.assertIn("client_namespace", row["reason"])
|
||||
|
||||
|
||||
class TestSystemHealthRoutes(CleanParityMixin, unittest.TestCase):
|
||||
"""The HTTP surface: versioned path, status codes, read-only guard."""
|
||||
|
||||
def setUp(self):
|
||||
super().setUp()
|
||||
clear_probe_cache()
|
||||
self.addCleanup(clear_probe_cache)
|
||||
self.client = TestClient(create_app())
|
||||
|
||||
def _patch_snapshot(self, probes, daemon_head="abc123"):
|
||||
snapshot = load_system_health(probes=probes, daemon_head=daemon_head)
|
||||
patcher = mock.patch(
|
||||
"webui.app.load_system_health",
|
||||
return_value=snapshot,
|
||||
)
|
||||
patcher.start()
|
||||
self.addCleanup(patcher.stop)
|
||||
return snapshot
|
||||
|
||||
def test_versioned_route_is_registered(self):
|
||||
self.assertEqual(API_PATH, "/api/v1/system/health")
|
||||
self._patch_snapshot(_ALL_HEALTHY)
|
||||
response = self.client.get(API_PATH)
|
||||
self.assertEqual(response.status_code, 200)
|
||||
|
||||
def test_healthy_payload_shape(self):
|
||||
self._patch_snapshot(_ALL_HEALTHY)
|
||||
data = self.client.get(API_PATH).json()
|
||||
self.assertEqual(data["status"], STATUS_OK)
|
||||
self.assertTrue(data["readiness"]["ready"])
|
||||
self.assertTrue(data["readiness"]["complete"])
|
||||
self.assertEqual(data["api"], API_PATH)
|
||||
self.assertEqual(len(data["dependencies"]), 3)
|
||||
for key in ("version", "process", "stale_runtime", "mcp_namespaces"):
|
||||
self.assertIn(key, data)
|
||||
self.assertIn("uptime_seconds", data["process"])
|
||||
self.assertIn("mutation_safe", data["stale_runtime"])
|
||||
|
||||
def test_degraded_dependency_returns_503(self):
|
||||
probes = (
|
||||
_probe("control_plane_db", STATUS_DOWN, detail="missing", kind="sqlite"),
|
||||
_probe("repository", STATUS_OK, kind="git"),
|
||||
)
|
||||
self._patch_snapshot(probes)
|
||||
response = self.client.get(API_PATH)
|
||||
self.assertEqual(response.status_code, 503)
|
||||
data = response.json()
|
||||
self.assertFalse(data["readiness"]["ready"])
|
||||
self.assertTrue(data["readiness"]["reasons"])
|
||||
|
||||
def test_dependency_entries_expose_status_and_latency(self):
|
||||
self._patch_snapshot(_ALL_HEALTHY)
|
||||
data = self.client.get(API_PATH).json()
|
||||
names = {entry["name"] for entry in data["dependencies"]}
|
||||
self.assertEqual(names, {"control_plane_db", "repository", "gitea"})
|
||||
for entry in data["dependencies"]:
|
||||
with self.subTest(dependency=entry["name"]):
|
||||
self.assertIn("status", entry)
|
||||
self.assertIn("required", entry)
|
||||
self.assertIn("latency_ms", entry)
|
||||
|
||||
def test_response_body_carries_no_client_secrets(self):
|
||||
self._patch_snapshot(_ALL_HEALTHY)
|
||||
body = self.client.get(API_PATH).text
|
||||
self.assertEqual(scan_text_for_client_secrets(body), [])
|
||||
|
||||
def test_deep_flag_is_forwarded(self):
|
||||
snapshot = load_system_health(probes=_ALL_HEALTHY, daemon_head="abc")
|
||||
with mock.patch(
|
||||
"webui.app.load_system_health", return_value=snapshot
|
||||
) as loader:
|
||||
self.client.get(f"{API_PATH}?deep=1")
|
||||
loader.assert_called_once_with(deep=True)
|
||||
|
||||
def test_shallow_is_the_default(self):
|
||||
snapshot = load_system_health(probes=_ALL_HEALTHY, daemon_head="abc")
|
||||
with mock.patch(
|
||||
"webui.app.load_system_health", return_value=snapshot
|
||||
) as loader:
|
||||
self.client.get(API_PATH)
|
||||
loader.assert_called_once_with(deep=False)
|
||||
|
||||
def test_route_rejects_mutation_methods(self):
|
||||
for method in ("POST", "PUT", "PATCH", "DELETE"):
|
||||
with self.subTest(method=method):
|
||||
response = self.client.request(method, API_PATH)
|
||||
self.assertEqual(response.status_code, 405)
|
||||
self.assertEqual(response.json()["error"], "read-only-mvp")
|
||||
|
||||
def test_default_shallow_call_skips_the_network_probe(self):
|
||||
"""The expensive probe must not run unless it was asked for."""
|
||||
with mock.patch("webui.system_health.probe_gitea") as probe:
|
||||
snapshot = load_system_health(deep=False)
|
||||
probe.assert_not_called()
|
||||
gitea = next(p for p in snapshot.dependencies if p.name == "gitea")
|
||||
self.assertEqual(gitea.status, STATUS_SKIPPED)
|
||||
|
||||
|
||||
class TestHealthRouteBackwardCompatibility(unittest.TestCase):
|
||||
"""`/health` is expanded additively; MVP consumers must keep working."""
|
||||
|
||||
def setUp(self):
|
||||
self.client = TestClient(create_app())
|
||||
|
||||
def test_mvp_keys_are_unchanged(self):
|
||||
data = self.client.get("/health").json()
|
||||
self.assertEqual(data["status"], "ok")
|
||||
self.assertEqual(data["service"], "mcp-control-plane-webui")
|
||||
self.assertEqual(data["mode"], "read-only-mvp")
|
||||
self.assertIn("timestamp", data)
|
||||
self.assertEqual(data["deployment"]["mode"], "internal-operator-console")
|
||||
|
||||
def test_health_points_at_the_versioned_api(self):
|
||||
data = self.client.get("/health").json()
|
||||
self.assertEqual(data["system_health_api"], API_PATH)
|
||||
self.assertIn("uptime_seconds", data)
|
||||
self.assertIn("started_at", data)
|
||||
|
||||
def test_health_runs_no_dependency_probe(self):
|
||||
"""Liveness must stay cheap: no probe, no snapshot assembly."""
|
||||
with mock.patch("webui.app.load_system_health") as loader:
|
||||
response = self.client.get("/health")
|
||||
self.assertEqual(response.status_code, 200)
|
||||
loader.assert_not_called()
|
||||
|
||||
|
||||
class TestSnapshotSerialisation(CleanParityMixin, unittest.TestCase):
|
||||
def test_snapshot_dict_is_json_serialisable(self):
|
||||
snapshot = load_system_health(probes=_ALL_HEALTHY, daemon_head="abc123")
|
||||
encoded = json.dumps(snapshot_to_dict(snapshot))
|
||||
self.assertIn("readiness", encoded)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user