Merge remote-tracking branch 'prgs/master' into feat/issue-551-codex-gitea-workflow
# Conflicts: # docs/llm-workflow-runbooks.md
This commit is contained in:
@@ -284,7 +284,7 @@ class TestTaskCapabilityMap(unittest.TestCase):
|
||||
required_permission("reconciliation_cleanup"),
|
||||
"gitea.branch.delete",
|
||||
)
|
||||
self.assertEqual(required_role("reconciliation_cleanup"), "author")
|
||||
self.assertEqual(required_role("reconciliation_cleanup"), "reconciler")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
|
||||
@@ -0,0 +1,81 @@
|
||||
"""Doc-contract checks for the bootstrap review path (#557).
|
||||
|
||||
Self-hosted MCP workflow fixes can deadlock live review daemons. These tests
|
||||
pin the narrow controller-authorized bootstrap path so it cannot silently
|
||||
regress into a general gate bypass.
|
||||
"""
|
||||
|
||||
from pathlib import Path
|
||||
|
||||
REPO_ROOT = Path(__file__).resolve().parent.parent
|
||||
|
||||
DOC = REPO_ROOT / "docs" / "bootstrap-review-path.md"
|
||||
RUNBOOK = REPO_ROOT / "docs" / "llm-workflow-runbooks.md"
|
||||
SKILL = REPO_ROOT / "skills" / "llm-project-workflow" / "SKILL.md"
|
||||
|
||||
|
||||
def _normalize(text: str) -> str:
|
||||
return " ".join(text.split())
|
||||
|
||||
|
||||
def test_bootstrap_doc_exists_and_names_authorization_marker():
|
||||
text = DOC.read_text(encoding="utf-8")
|
||||
assert "BOOTSTRAP REVIEW AUTHORIZATION (#557)" in text
|
||||
assert "bootstrap deadlock" in text.lower() or "Bootstrap deadlock" in text
|
||||
|
||||
|
||||
def test_bootstrap_doc_requires_validation_and_audits():
|
||||
text = _normalize(DOC.read_text(encoding="utf-8"))
|
||||
for phrase in (
|
||||
"git diff --check",
|
||||
"Duplicate-PR audit",
|
||||
"Mutation ledger audit",
|
||||
"Contamination audit",
|
||||
"workflow-contaminated",
|
||||
):
|
||||
assert phrase in text, f"bootstrap doc missing: {phrase!r}"
|
||||
|
||||
|
||||
def test_bootstrap_doc_lists_allowed_and_forbidden_actions():
|
||||
text = _normalize(DOC.read_text(encoding="utf-8"))
|
||||
lower = text.lower()
|
||||
for phrase in (
|
||||
"Allowed verification actions",
|
||||
"Forbidden actions",
|
||||
"project root",
|
||||
"direct Python import",
|
||||
"branches/",
|
||||
):
|
||||
assert phrase in text, f"bootstrap doc missing: {phrase!r}"
|
||||
assert "curl" in lower
|
||||
assert "raw" in lower
|
||||
|
||||
|
||||
def test_bootstrap_doc_forbids_general_gate_weakening():
|
||||
text = _normalize(DOC.read_text(encoding="utf-8")).lower()
|
||||
assert "never weakens normal" in text or "does not weaken normal" in text
|
||||
assert "general bypass" in text or "general gate" in text
|
||||
|
||||
|
||||
def test_bootstrap_doc_post_land_restart_and_verify():
|
||||
text = _normalize(DOC.read_text(encoding="utf-8"))
|
||||
for phrase in (
|
||||
"Restart MCP daemons",
|
||||
"canonical tools",
|
||||
"gitea_whoami",
|
||||
):
|
||||
assert phrase in text, f"post-bootstrap guidance missing: {phrase!r}"
|
||||
|
||||
|
||||
def test_runbook_links_bootstrap_path():
|
||||
text = _normalize(RUNBOOK.read_text(encoding="utf-8"))
|
||||
assert "Bootstrap Review Path for self-hosted MCP fixes (#557)" in text
|
||||
assert "bootstrap-review-path.md" in text
|
||||
assert "BOOTSTRAP REVIEW AUTHORIZATION (#557)" in text
|
||||
|
||||
|
||||
def test_skill_links_bootstrap_path():
|
||||
text = _normalize(SKILL.read_text(encoding="utf-8"))
|
||||
assert "Bootstrap Review Path (#557)" in text
|
||||
assert "bootstrap-review-path.md" in text
|
||||
assert "BLOCKED + DIAGNOSE" in text or "BLOCKED" in text
|
||||
@@ -35,6 +35,9 @@ def _review_handoff(**overrides):
|
||||
"- External-state mutations: none",
|
||||
"- Read-only diagnostics: issue and PR metadata inspected",
|
||||
"- Current status: PR open",
|
||||
"- Next actor: author",
|
||||
"- Next action: address review feedback",
|
||||
"- Next prompt: Fix PR #203 per reviewer request_changes and push updated head",
|
||||
"- Blockers: none",
|
||||
"- Next: await author",
|
||||
"- Safety: no self-review; no self-merge",
|
||||
@@ -92,6 +95,9 @@ def _reconcile_handoff(drop=(), **overrides):
|
||||
"- Read-only diagnostics: gitea_view_pr, gitea_view_issue",
|
||||
"- Reconciliation mutations: PR comment posted",
|
||||
"- Current status: reconciliation complete",
|
||||
"- Next actor: reconciler",
|
||||
"- Next action: close PR #99 after capability proof",
|
||||
"- Next prompt: Reconcile already-landed PR #99 with prgs-reconciler profile",
|
||||
"- Blocker: missing gitea.pr.close capability",
|
||||
"- Safe next action: hand off to a profile with gitea.pr.close",
|
||||
"- No review/merge confirmation: confirmed",
|
||||
|
||||
@@ -0,0 +1,152 @@
|
||||
"""Tests for the master-parity staleness gate (#420).
|
||||
|
||||
Covers the pure assessment logic, the mutation-block helper, the env escape
|
||||
hatches, and the server-side wiring: reads are never blocked by staleness, a
|
||||
mutation is refused when the on-disk master has advanced, and the read-only
|
||||
gitea_assess_master_parity tool reports the stale state.
|
||||
"""
|
||||
import os
|
||||
import sys
|
||||
import unittest
|
||||
from unittest.mock import patch
|
||||
|
||||
sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
|
||||
|
||||
import master_parity_gate as mp # noqa: E402
|
||||
|
||||
|
||||
SHA_A = "a" * 40
|
||||
SHA_B = "b" * 40
|
||||
|
||||
|
||||
class TestAssessMasterParity(unittest.TestCase):
|
||||
def test_in_parity_when_heads_match(self):
|
||||
res = mp.assess_master_parity({"startup_head": SHA_A}, SHA_A)
|
||||
self.assertTrue(res["in_parity"])
|
||||
self.assertFalse(res["stale"])
|
||||
self.assertFalse(res["restart_required"])
|
||||
self.assertTrue(res["determinable"])
|
||||
|
||||
def test_stale_when_master_advanced(self):
|
||||
res = mp.assess_master_parity({"startup_head": SHA_A}, SHA_B)
|
||||
self.assertFalse(res["in_parity"])
|
||||
self.assertTrue(res["stale"])
|
||||
self.assertTrue(res["restart_required"])
|
||||
self.assertTrue(res["determinable"])
|
||||
self.assertTrue(any("restart" in r for r in res["reasons"]))
|
||||
|
||||
def test_missing_startup_head_not_determinable(self):
|
||||
res = mp.assess_master_parity({"startup_head": None}, SHA_B)
|
||||
self.assertFalse(res["determinable"])
|
||||
self.assertFalse(res["stale"])
|
||||
self.assertTrue(res["in_parity"])
|
||||
|
||||
def test_missing_current_head_not_determinable(self):
|
||||
res = mp.assess_master_parity({"startup_head": SHA_A}, None)
|
||||
self.assertFalse(res["determinable"])
|
||||
self.assertFalse(res["stale"])
|
||||
self.assertTrue(res["in_parity"])
|
||||
|
||||
def test_none_startup_baseline(self):
|
||||
res = mp.assess_master_parity(None, SHA_A)
|
||||
self.assertFalse(res["determinable"])
|
||||
self.assertFalse(res["stale"])
|
||||
|
||||
|
||||
class TestBlockReasonsAndReport(unittest.TestCase):
|
||||
def test_stale_produces_block_reasons(self):
|
||||
res = mp.assess_master_parity({"startup_head": SHA_A}, SHA_B)
|
||||
self.assertTrue(mp.parity_block_reasons(res))
|
||||
|
||||
def test_in_parity_no_block_reasons(self):
|
||||
res = mp.assess_master_parity({"startup_head": SHA_A}, SHA_A)
|
||||
self.assertEqual(mp.parity_block_reasons(res), [])
|
||||
|
||||
def test_disable_env_suppresses_block(self):
|
||||
res = mp.assess_master_parity({"startup_head": SHA_A}, SHA_B)
|
||||
with patch.dict(os.environ, {mp.ENV_DISABLE: "1"}):
|
||||
self.assertTrue(mp.gate_disabled())
|
||||
self.assertEqual(mp.parity_block_reasons(res), [])
|
||||
|
||||
def test_report_shape_when_stale(self):
|
||||
res = mp.assess_master_parity({"startup_head": SHA_A}, SHA_B)
|
||||
report = mp.parity_report(res)
|
||||
self.assertEqual(report["kind"], "server_stale")
|
||||
self.assertTrue(report["restart_required"])
|
||||
self.assertEqual(report["startup_head"], SHA_A)
|
||||
self.assertEqual(report["current_head"], SHA_B)
|
||||
self.assertTrue(report["recovery"])
|
||||
|
||||
|
||||
class TestReadGitHead(unittest.TestCase):
|
||||
def test_test_override_takes_precedence(self):
|
||||
with patch.dict(os.environ, {mp.ENV_TEST_CURRENT_HEAD: SHA_B}):
|
||||
self.assertEqual(mp.read_git_head("/nonexistent"), SHA_B)
|
||||
|
||||
def test_blank_override_is_none(self):
|
||||
with patch.dict(os.environ, {mp.ENV_TEST_CURRENT_HEAD: " "}):
|
||||
self.assertIsNone(mp.read_git_head("/nonexistent"))
|
||||
|
||||
def test_empty_root_is_none(self):
|
||||
# No override set; empty root cannot resolve a HEAD.
|
||||
env = {k: v for k, v in os.environ.items()
|
||||
if k != mp.ENV_TEST_CURRENT_HEAD}
|
||||
with patch.dict(os.environ, env, clear=True):
|
||||
self.assertIsNone(mp.read_git_head(""))
|
||||
|
||||
|
||||
class TestServerWiring(unittest.TestCase):
|
||||
"""Integration with the gate choke point in the server namespace."""
|
||||
|
||||
def setUp(self):
|
||||
import mcp_server
|
||||
self.srv = mcp_server
|
||||
# Pin a known startup baseline so the current-HEAD override can diverge.
|
||||
self._saved = self.srv._STARTUP_PARITY
|
||||
self.srv._STARTUP_PARITY = {"root": self.srv.PROJECT_ROOT,
|
||||
"startup_head": SHA_A}
|
||||
|
||||
def tearDown(self):
|
||||
self.srv._STARTUP_PARITY = self._saved
|
||||
|
||||
def test_reads_not_blocked_when_stale(self):
|
||||
with patch.dict(os.environ, {mp.ENV_TEST_CURRENT_HEAD: SHA_B}):
|
||||
self.assertEqual(self.srv._master_parity_block("gitea.read"), [])
|
||||
|
||||
def test_mutation_blocked_when_stale(self):
|
||||
with patch.dict(os.environ, {mp.ENV_TEST_CURRENT_HEAD: SHA_B}):
|
||||
reasons = self.srv._master_parity_block("gitea.pr.create")
|
||||
self.assertTrue(reasons)
|
||||
# The profile-operation choke point surfaces the same block.
|
||||
self.assertTrue(self.srv._profile_operation_gate("gitea.pr.create"))
|
||||
|
||||
def test_mutation_allowed_when_in_parity(self):
|
||||
with patch.dict(os.environ, {mp.ENV_TEST_CURRENT_HEAD: SHA_A}):
|
||||
self.assertEqual(
|
||||
self.srv._master_parity_block("gitea.pr.create"), [])
|
||||
|
||||
def test_disable_env_lets_mutation_through(self):
|
||||
with patch.dict(os.environ, {mp.ENV_TEST_CURRENT_HEAD: SHA_B,
|
||||
mp.ENV_DISABLE: "1"}):
|
||||
self.assertEqual(
|
||||
self.srv._master_parity_block("gitea.pr.create"), [])
|
||||
|
||||
def test_assess_tool_reports_stale(self):
|
||||
with patch.dict(os.environ, {mp.ENV_TEST_CURRENT_HEAD: SHA_B}):
|
||||
out = self.srv.gitea_assess_master_parity(remote="prgs")
|
||||
self.assertTrue(out["stale"])
|
||||
self.assertTrue(out["restart_required"])
|
||||
self.assertEqual(out["startup_head"], SHA_A)
|
||||
self.assertEqual(out["current_head"], SHA_B)
|
||||
self.assertIn("report", out)
|
||||
|
||||
def test_assess_tool_reports_in_parity(self):
|
||||
with patch.dict(os.environ, {mp.ENV_TEST_CURRENT_HEAD: SHA_A}):
|
||||
out = self.srv.gitea_assess_master_parity(remote="prgs")
|
||||
self.assertFalse(out["stale"])
|
||||
self.assertTrue(out["in_parity"])
|
||||
self.assertNotIn("report", out)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -105,6 +105,19 @@ class TestMcpMenuScript(unittest.TestCase):
|
||||
self.assertIn("./mcp-menu.sh", docs_text)
|
||||
self.assertIn("placeholder", docs_text.lower())
|
||||
|
||||
def test_reviewer_skip_stale_request_changes_prompt_discoverable(self):
|
||||
# #482: the skip-already-reviewed-stale-REQUEST_CHANGES reviewer prompt
|
||||
# must be reachable from the reviewer menu and documented.
|
||||
label = "Skip already-reviewed stale REQUEST_CHANGES PR and hand off to author"
|
||||
reviewer_fn = self._extract_function("show_reviewer_prompts")
|
||||
self.assertIn(label, reviewer_fn, "new reviewer prompt label must appear in the reviewer menu")
|
||||
# The prompt must instruct: no duplicate terminal review mutation for a
|
||||
# non-stale REQUEST_CHANGES head, and must carry the author handoff.
|
||||
self.assertIn("do NOT submit another terminal review mutation", reviewer_fn)
|
||||
self.assertIn("author-ready fix prompt", reviewer_fn)
|
||||
docs_text = DOCS.read_text(encoding="utf-8")
|
||||
self.assertIn("REQUEST_CHANGES", docs_text)
|
||||
|
||||
def _extract_function(self, name: str) -> str:
|
||||
marker = f"{name}() {{"
|
||||
start = self.content.index(marker)
|
||||
|
||||
@@ -4,7 +4,7 @@ import os
|
||||
import sys
|
||||
import tempfile
|
||||
import unittest
|
||||
from unittest.mock import patch
|
||||
from unittest.mock import patch, MagicMock
|
||||
|
||||
sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
|
||||
|
||||
@@ -154,6 +154,290 @@ class TestMergedCleanupReport(unittest.TestCase):
|
||||
finally:
|
||||
os.remove(lock_path)
|
||||
|
||||
@patch("subprocess.run")
|
||||
def test_discover_local_worktrees(self, mock_run):
|
||||
mock_output = (
|
||||
"worktree /repo/Gitea-Tools\n"
|
||||
"bare\n\n"
|
||||
"worktree /repo/Gitea-Tools/branches/custom-name\n"
|
||||
"branch refs/heads/feat/issue-11-x\n"
|
||||
"HEAD cafebabe\n\n"
|
||||
)
|
||||
mock_run.return_value = MagicMock(returncode=0, stdout=mock_output)
|
||||
res = mcr.discover_local_worktrees("/repo/Gitea-Tools")
|
||||
self.assertEqual(res, {"feat/issue-11-x": "/repo/Gitea-Tools/branches/custom-name"})
|
||||
|
||||
@patch("subprocess.run")
|
||||
@patch("merged_cleanup_reconcile.read_local_worktree_state")
|
||||
def test_build_report_discovers_non_canonical_worktree_by_branch(self, mock_state, mock_run):
|
||||
mock_output = (
|
||||
"worktree /repo/Gitea-Tools\n"
|
||||
"bare\n\n"
|
||||
"worktree /repo/Gitea-Tools/branches/custom-name\n"
|
||||
"branch refs/heads/feat/issue-11-x\n"
|
||||
"HEAD cafebabe\n\n"
|
||||
)
|
||||
mock_run.return_value = MagicMock(returncode=0, stdout=mock_output)
|
||||
mock_state.return_value = {
|
||||
"exists": True,
|
||||
"clean": True,
|
||||
"current_branch": "feat/issue-11-x",
|
||||
"head_sha": "cafebabe",
|
||||
"dirty_files": [],
|
||||
}
|
||||
|
||||
report = mcr.build_reconciliation_report(
|
||||
project_root="/repo/Gitea-Tools",
|
||||
closed_prs=[
|
||||
{
|
||||
"number": 11,
|
||||
"title": "merged",
|
||||
"body": "Closes #11",
|
||||
"head": {"ref": "feat/issue-11-x"},
|
||||
"merged_at": "2026-07-06T12:00:00Z",
|
||||
"merge_commit_sha": "abc123",
|
||||
}
|
||||
],
|
||||
open_prs=[],
|
||||
remote_branch_exists={"feat/issue-11-x": True},
|
||||
head_on_master={11: True},
|
||||
delete_capability_allowed=True,
|
||||
)
|
||||
self.assertEqual(report["merged_pr_count"], 1)
|
||||
entry = report["entries"][0]
|
||||
local = entry["local_worktree"]
|
||||
self.assertEqual(local["worktree_path"], "/repo/Gitea-Tools/branches/custom-name")
|
||||
self.assertEqual(local["match_type"], "branch")
|
||||
|
||||
@patch("subprocess.run")
|
||||
def test_remove_local_worktree_uses_discovered_path(self, mock_run):
|
||||
mock_output = (
|
||||
"worktree /repo/Gitea-Tools\n"
|
||||
"bare\n\n"
|
||||
"worktree /repo/Gitea-Tools/branches/custom-name\n"
|
||||
"branch refs/heads/feat/issue-11-x\n"
|
||||
"HEAD cafebabe\n\n"
|
||||
)
|
||||
# First call is discover_local_worktrees, second is git worktree remove
|
||||
mock_run.side_effect = [
|
||||
MagicMock(returncode=0, stdout=mock_output),
|
||||
MagicMock(returncode=0, stdout="", stderr=""),
|
||||
]
|
||||
|
||||
with patch("os.path.isdir", return_value=True):
|
||||
result = mcr.remove_local_worktree("/repo/Gitea-Tools", "feat/issue-11-x")
|
||||
self.assertTrue(result["success"])
|
||||
self.assertTrue(result["performed"])
|
||||
self.assertEqual(result["worktree_path"], "/repo/Gitea-Tools/branches/custom-name")
|
||||
# Assert git worktree remove was called with the custom path
|
||||
mock_run.assert_any_call(
|
||||
["git", "-C", "/repo/Gitea-Tools", "worktree", "remove", "/repo/Gitea-Tools/branches/custom-name"],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
check=False,
|
||||
)
|
||||
|
||||
|
||||
class TestReviewerScratchCleanup(unittest.TestCase):
|
||||
"""#534: authorized cleanup path for reviewer scratch worktrees."""
|
||||
|
||||
def test_parse_reviewer_scratch_folder(self):
|
||||
self.assertEqual(mcr.parse_reviewer_scratch_folder("review-pr512-submit"), 512)
|
||||
self.assertEqual(mcr.parse_reviewer_scratch_folder("review-pr99"), 99)
|
||||
self.assertIsNone(mcr.parse_reviewer_scratch_folder("feat-issue-11-x"))
|
||||
self.assertIsNone(mcr.parse_reviewer_scratch_folder("review-feat-issue-11"))
|
||||
self.assertIsNone(mcr.parse_reviewer_scratch_folder("review-pr512-extra-tail"))
|
||||
|
||||
def test_assess_safe_clean_detached_after_merge(self):
|
||||
assessment = mcr.assess_reviewer_scratch_cleanup(
|
||||
pr_number=512,
|
||||
worktree_path="/repo/Gitea-Tools/branches/review-pr512-submit",
|
||||
folder_name="review-pr512-submit",
|
||||
pr_merged=True,
|
||||
pr_closed=True,
|
||||
worktree_state={
|
||||
"exists": True,
|
||||
"clean": True,
|
||||
"current_branch": None,
|
||||
"head_sha": "abc",
|
||||
"dirty_files": [],
|
||||
},
|
||||
active_reviewer_lease=False,
|
||||
)
|
||||
self.assertTrue(assessment["safe_to_remove_worktree"])
|
||||
self.assertFalse(assessment["author_worktree_cleanup"])
|
||||
self.assertEqual(
|
||||
assessment["recommended_action"], "remove_reviewer_scratch_worktree"
|
||||
)
|
||||
self.assertEqual(assessment["worktree_kind"], "reviewer_scratch")
|
||||
|
||||
def test_assess_blocks_dirty_scratch(self):
|
||||
assessment = mcr.assess_reviewer_scratch_cleanup(
|
||||
pr_number=512,
|
||||
worktree_path="/repo/Gitea-Tools/branches/review-pr512-submit",
|
||||
folder_name="review-pr512-submit",
|
||||
pr_merged=True,
|
||||
pr_closed=True,
|
||||
worktree_state={
|
||||
"exists": True,
|
||||
"clean": False,
|
||||
"current_branch": None,
|
||||
"head_sha": "abc",
|
||||
"dirty_files": ["foo.py"],
|
||||
},
|
||||
active_reviewer_lease=False,
|
||||
)
|
||||
self.assertFalse(assessment["safe_to_remove_worktree"])
|
||||
self.assertTrue(
|
||||
any("tracked edits" in r for r in assessment["block_reasons"])
|
||||
)
|
||||
|
||||
def test_assess_blocks_active_lease(self):
|
||||
assessment = mcr.assess_reviewer_scratch_cleanup(
|
||||
pr_number=512,
|
||||
worktree_path="/repo/Gitea-Tools/branches/review-pr512-submit",
|
||||
folder_name="review-pr512-submit",
|
||||
pr_merged=True,
|
||||
pr_closed=True,
|
||||
worktree_state={
|
||||
"exists": True,
|
||||
"clean": True,
|
||||
"current_branch": None,
|
||||
"head_sha": "abc",
|
||||
"dirty_files": [],
|
||||
},
|
||||
active_reviewer_lease=True,
|
||||
)
|
||||
self.assertFalse(assessment["safe_to_remove_worktree"])
|
||||
self.assertTrue(
|
||||
any("active reviewer lease" in r for r in assessment["block_reasons"])
|
||||
)
|
||||
|
||||
def test_assess_blocks_open_pr(self):
|
||||
assessment = mcr.assess_reviewer_scratch_cleanup(
|
||||
pr_number=512,
|
||||
worktree_path="/repo/Gitea-Tools/branches/review-pr512-submit",
|
||||
folder_name="review-pr512-submit",
|
||||
pr_merged=False,
|
||||
pr_closed=False,
|
||||
worktree_state={
|
||||
"exists": True,
|
||||
"clean": True,
|
||||
"current_branch": None,
|
||||
"head_sha": "abc",
|
||||
"dirty_files": [],
|
||||
},
|
||||
active_reviewer_lease=False,
|
||||
)
|
||||
self.assertFalse(assessment["safe_to_remove_worktree"])
|
||||
self.assertTrue(any("still open" in r for r in assessment["block_reasons"]))
|
||||
|
||||
@patch("subprocess.run")
|
||||
@patch("merged_cleanup_reconcile.read_local_worktree_state")
|
||||
def test_report_lists_scratch_separately_from_author(self, mock_state, mock_run):
|
||||
mock_output = (
|
||||
"worktree /repo/Gitea-Tools\n"
|
||||
"bare\n\n"
|
||||
"worktree /repo/Gitea-Tools/branches/feat-issue-11-x\n"
|
||||
"branch refs/heads/feat/issue-11-x\n"
|
||||
"HEAD deadbeef\n\n"
|
||||
"worktree /repo/Gitea-Tools/branches/review-pr512-submit\n"
|
||||
"HEAD abcdef0\n"
|
||||
"detached\n\n"
|
||||
)
|
||||
mock_run.return_value = MagicMock(returncode=0, stdout=mock_output)
|
||||
|
||||
def _state(path):
|
||||
if "review-pr512" in path:
|
||||
return {
|
||||
"exists": True,
|
||||
"clean": True,
|
||||
"current_branch": None,
|
||||
"head_sha": "abcdef0",
|
||||
"dirty_files": [],
|
||||
}
|
||||
return {
|
||||
"exists": True,
|
||||
"clean": True,
|
||||
"current_branch": "feat/issue-11-x",
|
||||
"head_sha": "deadbeef",
|
||||
"dirty_files": [],
|
||||
}
|
||||
|
||||
mock_state.side_effect = _state
|
||||
report = mcr.build_reconciliation_report(
|
||||
project_root="/repo/Gitea-Tools",
|
||||
closed_prs=[
|
||||
{
|
||||
"number": 11,
|
||||
"title": "merged author",
|
||||
"body": "Closes #11",
|
||||
"head": {"ref": "feat/issue-11-x"},
|
||||
"merged_at": "2026-07-06T12:00:00Z",
|
||||
},
|
||||
{
|
||||
"number": 512,
|
||||
"title": "merged reviewed",
|
||||
"body": "Closes #512",
|
||||
"head": {"ref": "feat/issue-512-x"},
|
||||
"merged_at": "2026-07-06T12:00:00Z",
|
||||
},
|
||||
],
|
||||
open_prs=[],
|
||||
remote_branch_exists={
|
||||
"feat/issue-11-x": False,
|
||||
"feat/issue-512-x": False,
|
||||
},
|
||||
head_on_master={11: True, 512: True},
|
||||
delete_capability_allowed=True,
|
||||
active_reviewer_leases={512: False},
|
||||
)
|
||||
self.assertEqual(report["merged_pr_count"], 2)
|
||||
self.assertEqual(report["reviewer_scratch_count"], 1)
|
||||
scratch = report["reviewer_scratch_entries"][0]
|
||||
self.assertEqual(scratch["pr_number"], 512)
|
||||
self.assertEqual(scratch["worktree_kind"], "reviewer_scratch")
|
||||
self.assertFalse(scratch["author_worktree_cleanup"])
|
||||
self.assertTrue(scratch["safe_to_remove_worktree"])
|
||||
# Author entries must not swallow the scratch path as local_worktree.
|
||||
author_paths = {
|
||||
(e.get("local_worktree") or {}).get("worktree_path")
|
||||
for e in report["entries"]
|
||||
}
|
||||
self.assertNotIn(
|
||||
"/repo/Gitea-Tools/branches/review-pr512-submit", author_paths
|
||||
)
|
||||
|
||||
@patch("subprocess.run")
|
||||
def test_remove_reviewer_scratch_worktree_path_guard(self, mock_run):
|
||||
mock_run.return_value = MagicMock(returncode=0, stdout="", stderr="")
|
||||
with patch("os.path.isdir", return_value=True):
|
||||
bad = mcr.remove_reviewer_scratch_worktree(
|
||||
"/repo/Gitea-Tools",
|
||||
"/repo/Gitea-Tools/branches/feat-issue-11-x",
|
||||
)
|
||||
self.assertFalse(bad["performed"])
|
||||
good = mcr.remove_reviewer_scratch_worktree(
|
||||
"/repo/Gitea-Tools",
|
||||
"/repo/Gitea-Tools/branches/review-pr512-submit",
|
||||
)
|
||||
self.assertTrue(good["success"])
|
||||
self.assertTrue(good["performed"])
|
||||
self.assertFalse(good["author_worktree_cleanup"])
|
||||
mock_run.assert_any_call(
|
||||
[
|
||||
"git",
|
||||
"-C",
|
||||
"/repo/Gitea-Tools",
|
||||
"worktree",
|
||||
"remove",
|
||||
"/repo/Gitea-Tools/branches/review-pr512-submit",
|
||||
],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
check=False,
|
||||
)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -265,5 +265,38 @@ class TestCleanupReportVerifier(unittest.TestCase):
|
||||
self.assertEqual(direct["proven"], wrapped["proven"])
|
||||
|
||||
|
||||
class TestCleanupEmptyQueue(unittest.TestCase):
|
||||
def test_empty_queue_report_passes(self):
|
||||
report = _clean_report(
|
||||
selected="Selected PR: none",
|
||||
decision="Review decision: comment",
|
||||
next="Next suggested PR: approvals (queue empty)",
|
||||
pagination="PR inventory pagination proof: inventory_complete=true, total_count 0",
|
||||
)
|
||||
report += "\npr_inventory_trust_gate.status: trusted_empty"
|
||||
result = assess_pr_queue_cleanup_report(report)
|
||||
self.assertTrue(result["proven"], result["reasons"])
|
||||
|
||||
def test_empty_queue_without_evidence_fails(self):
|
||||
report = _clean_report(
|
||||
selected="Selected PR: none",
|
||||
)
|
||||
result = assess_pr_queue_cleanup_report(report)
|
||||
self.assertFalse(result["proven"])
|
||||
self.assertTrue(
|
||||
any("no valid empty-queue claim" in r for r in result["reasons"]),
|
||||
result["reasons"]
|
||||
)
|
||||
|
||||
def test_contradictory_selected_pr_fails(self):
|
||||
report = _clean_report() + "\nSelected PR: none"
|
||||
result = assess_pr_queue_cleanup_report(report)
|
||||
self.assertFalse(result["proven"])
|
||||
self.assertTrue(
|
||||
any("contains both a selected PR and a claim of none selected" in r for r in result["reasons"]),
|
||||
result["reasons"]
|
||||
)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
|
||||
@@ -0,0 +1,112 @@
|
||||
import unittest
|
||||
|
||||
from premerge_baseline_proof import (
|
||||
CLEAN_PASS,
|
||||
PREMERGE_BASELINE_PROVEN_FAILURE,
|
||||
UNRESOLVED_REGRESSION_RISK,
|
||||
assess_premerge_baseline_proof,
|
||||
)
|
||||
from final_report_validator import assess_final_report_validator
|
||||
|
||||
_FIELDS = (
|
||||
"Pre-merge base commit: 1a2b3c4d5e6f7a8b9c0d1e2f3a4b5c6d7e8f9a0b\n"
|
||||
"Tested commit: 1a2b3c4d5e6f7a8b9c0d1e2f3a4b5c6d7e8f9a0b\n"
|
||||
"Command: venv/bin/python -m pytest tests/test_foo.py -q\n"
|
||||
"Exit status: 1\n"
|
||||
"Failure signature: AssertionError: None is not true\n"
|
||||
)
|
||||
|
||||
# Current-master-only reproduction carries no pre-merge base proof (that is the
|
||||
# whole defect #533 guards against): a command/exit/signature but no base commit.
|
||||
_CURRENT_ONLY_FIELDS = (
|
||||
"Command: venv/bin/python -m pytest tests/test_foo.py -q\n"
|
||||
"Exit status: 1\n"
|
||||
"Failure signature: AssertionError: None is not true\n"
|
||||
)
|
||||
|
||||
|
||||
class TestPremergeBaselineProof(unittest.TestCase):
|
||||
def test_clean_pass_not_blocked(self):
|
||||
result = assess_premerge_baseline_proof(
|
||||
"Validation: full suite passed, exit status 0. Clean pass."
|
||||
)
|
||||
self.assertFalse(result["block"])
|
||||
self.assertTrue(result["skipped"])
|
||||
self.assertEqual(result["label"], CLEAN_PASS)
|
||||
|
||||
def test_nonzero_exit_not_claimed_baseline_not_blocked(self):
|
||||
result = assess_premerge_baseline_proof(
|
||||
"Full suite: 1 failed. Investigating the regression; not yet classified."
|
||||
)
|
||||
self.assertFalse(result["block"])
|
||||
self.assertEqual(result["label"], UNRESOLVED_REGRESSION_RISK)
|
||||
|
||||
def test_current_master_only_reproduction_rejected(self):
|
||||
report = (
|
||||
"Full suite: 1 failed. This is a pre-existing baseline failure — "
|
||||
"reproduced on current master after the PR merged.\n" + _CURRENT_ONLY_FIELDS
|
||||
)
|
||||
result = assess_premerge_baseline_proof(report)
|
||||
self.assertTrue(result["block"])
|
||||
self.assertTrue(
|
||||
any("current-master reproduction only" in r for r in result["reasons"])
|
||||
)
|
||||
|
||||
def test_baseline_claim_missing_fields_blocked(self):
|
||||
report = (
|
||||
"Full suite: 1 failed. This failure is baseline / pre-existing. "
|
||||
"Verified against the pre-merge base commit."
|
||||
)
|
||||
result = assess_premerge_baseline_proof(report)
|
||||
self.assertTrue(result["block"])
|
||||
self.assertTrue(
|
||||
any("missing required proof field" in r for r in result["reasons"])
|
||||
)
|
||||
|
||||
def test_premerge_base_proof_accepted(self):
|
||||
report = (
|
||||
"Full suite: 1 failed. This is a pre-existing baseline failure, proven "
|
||||
"on the PR pre-merge base commit.\n" + _FIELDS
|
||||
)
|
||||
result = assess_premerge_baseline_proof(report)
|
||||
self.assertFalse(result["block"])
|
||||
self.assertTrue(result["proven"])
|
||||
self.assertEqual(result["label"], PREMERGE_BASELINE_PROVEN_FAILURE)
|
||||
|
||||
def test_documented_known_failure_record_accepted(self):
|
||||
report = (
|
||||
"Full suite: 1 failed. Baseline failure: cited known-failure record #529 "
|
||||
"which predates the PR.\n" + _FIELDS
|
||||
)
|
||||
result = assess_premerge_baseline_proof(report)
|
||||
self.assertFalse(result["block"])
|
||||
self.assertEqual(result["label"], PREMERGE_BASELINE_PROVEN_FAILURE)
|
||||
|
||||
|
||||
class TestPremergeBaselineProofValidatorIntegration(unittest.TestCase):
|
||||
def _has_rule(self, report):
|
||||
result = assess_final_report_validator(report, "review_pr")
|
||||
return any(
|
||||
f["rule_id"] == "reviewer.premerge_baseline_proof"
|
||||
for f in result["findings"]
|
||||
)
|
||||
|
||||
def test_review_report_current_master_only_flagged(self):
|
||||
report = (
|
||||
"## Reviewer final report\n"
|
||||
"Full suite: 1 failed. Pre-existing baseline failure — reproduced on "
|
||||
"post-merge master.\n" + _CURRENT_ONLY_FIELDS
|
||||
)
|
||||
self.assertTrue(self._has_rule(report))
|
||||
|
||||
def test_review_report_premerge_proof_clean(self):
|
||||
report = (
|
||||
"## Reviewer final report\n"
|
||||
"Full suite: 1 failed. Pre-existing baseline failure proven on the "
|
||||
"pre-merge base commit.\n" + _FIELDS
|
||||
)
|
||||
self.assertFalse(self._has_rule(report))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,263 @@
|
||||
"""Integration tests for reconciler merged cleanup (#523)."""
|
||||
import os
|
||||
import sys
|
||||
import unittest
|
||||
from unittest.mock import patch
|
||||
|
||||
sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
|
||||
|
||||
import mcp_server
|
||||
import task_capability_map
|
||||
from audit_reconciliation_mode import clear_phase
|
||||
from task_capability_map import required_role
|
||||
|
||||
RECONCILER_PROFILE = {
|
||||
"profile_name": "prgs-reconciler",
|
||||
"context": "prgs",
|
||||
"role": "reconciler",
|
||||
"username": "sysadmin",
|
||||
"base_url": "https://gitea.prgs.cc",
|
||||
"allowed_operations": [
|
||||
"gitea.read",
|
||||
"gitea.pr.close",
|
||||
"gitea.pr.comment",
|
||||
"gitea.issue.comment",
|
||||
],
|
||||
"forbidden_operations": [],
|
||||
}
|
||||
|
||||
AUTHOR_PROFILE = {
|
||||
"profile_name": "prgs-author",
|
||||
"context": "prgs",
|
||||
"role": "author",
|
||||
"username": "jcwalker3",
|
||||
"base_url": "https://gitea.prgs.cc",
|
||||
"allowed_operations": [
|
||||
"gitea.read",
|
||||
"gitea.pr.create",
|
||||
"gitea.branch.create",
|
||||
"gitea.branch.push",
|
||||
"gitea.repo.commit",
|
||||
],
|
||||
"forbidden_operations": [],
|
||||
}
|
||||
|
||||
CONTROL_ROOT = "/Users/jasonwalker/Development/Gitea-Tools"
|
||||
|
||||
|
||||
class TestReconcilerCleanupIntegration(unittest.TestCase):
|
||||
def setUp(self):
|
||||
clear_phase()
|
||||
mcp_server._preflight_whoami_called = False
|
||||
mcp_server._preflight_capability_called = False
|
||||
mcp_server._preflight_resolved_role = None
|
||||
mcp_server._preflight_resolved_task = None
|
||||
mcp_server._preflight_whoami_violation = False
|
||||
mcp_server._preflight_capability_violation = False
|
||||
mcp_server._preflight_whoami_violation_files = []
|
||||
mcp_server._preflight_capability_violation_files = []
|
||||
mcp_server._preflight_capability_baseline_porcelain = ""
|
||||
self.mock_api = patch("gitea_auth.api_request").start()
|
||||
self.mock_auth = patch(
|
||||
"mcp_server.get_auth_header", return_value="token test"
|
||||
).start()
|
||||
|
||||
def tearDown(self):
|
||||
patch.stopall()
|
||||
clear_phase()
|
||||
mcp_server._IDENTITY_CACHE.clear()
|
||||
|
||||
def test_capability_map_routes_merged_cleanup_to_reconciler(self):
|
||||
self.assertEqual(required_role("reconcile_merged_cleanups"), "reconciler")
|
||||
self.assertEqual(required_role("reconciliation_cleanup"), "reconciler")
|
||||
self.assertEqual(
|
||||
task_capability_map.required_permission("reconcile_merged_cleanups"),
|
||||
"gitea.read",
|
||||
)
|
||||
|
||||
@patch.dict(os.environ, {"GITEA_PROFILE_NAME": "prgs-reconciler"}, clear=True)
|
||||
@patch("mcp_server.get_profile", return_value=RECONCILER_PROFILE)
|
||||
def test_reconciler_can_run_reconcile_merged_cleanups_directly(self, _profile):
|
||||
mcp_server.record_preflight_check("whoami")
|
||||
mcp_server.record_preflight_check(
|
||||
"capability",
|
||||
resolved_role="reconciler",
|
||||
resolved_task="reconcile_merged_cleanups",
|
||||
)
|
||||
|
||||
self.mock_api.side_effect = [
|
||||
[ # closed pulls page
|
||||
{
|
||||
"number": 100,
|
||||
"title": "merged feature",
|
||||
"body": "Closes #100",
|
||||
"merged": True,
|
||||
"merged_at": "2026-07-06T12:00:00Z",
|
||||
"merge_commit_sha": "deadbeef",
|
||||
"head": {"ref": "feat/issue-100", "sha": "cafebabe"},
|
||||
},
|
||||
{
|
||||
# closed but NOT merged — must not become a cleanup entry
|
||||
"number": 101,
|
||||
"title": "abandoned",
|
||||
"body": "Closes #101",
|
||||
"merged": False,
|
||||
"merged_at": None,
|
||||
"head": {"ref": "feat/issue-101", "sha": "badcafe"},
|
||||
},
|
||||
],
|
||||
[], # open pulls
|
||||
]
|
||||
|
||||
with patch("mcp_server._remote_branch_exists", return_value=True):
|
||||
with patch(
|
||||
"mcp_server.merged_cleanup_reconcile.is_head_ancestor_of_ref",
|
||||
return_value=True,
|
||||
):
|
||||
result = mcp_server.gitea_reconcile_merged_cleanups(
|
||||
dry_run=True,
|
||||
remote="prgs",
|
||||
)
|
||||
self.assertTrue(result["success"])
|
||||
self.assertTrue(result["dry_run"])
|
||||
entry_numbers = [e["issue_number"] for e in result["entries"]]
|
||||
self.assertEqual(entry_numbers, [100])
|
||||
self.assertNotIn(101, entry_numbers)
|
||||
|
||||
@patch.dict(
|
||||
os.environ,
|
||||
{
|
||||
"GITEA_PROFILE_NAME": "prgs-reconciler",
|
||||
# Force preflight purity to evaluate (disable test short-circuit).
|
||||
"GITEA_TEST_PORCELAIN": "",
|
||||
},
|
||||
clear=True,
|
||||
)
|
||||
@patch("mcp_server.get_profile", return_value=RECONCILER_PROFILE)
|
||||
def test_reconciler_role_bypasses_branches_only_mutation_guard(self, _profile):
|
||||
mcp_server.record_preflight_check("whoami")
|
||||
mcp_server.record_preflight_check(
|
||||
"capability",
|
||||
resolved_role="reconciler",
|
||||
resolved_task="reconcile_merged_cleanups",
|
||||
)
|
||||
|
||||
with patch("mcp_server.PROJECT_ROOT", CONTROL_ROOT):
|
||||
with patch(
|
||||
"mcp_server.root_checkout_guard.resolve_remote_master_sha",
|
||||
return_value="abc123",
|
||||
):
|
||||
with patch(
|
||||
"mcp_server.issue_lock_worktree.read_worktree_git_state",
|
||||
return_value={
|
||||
"current_branch": "master",
|
||||
"head_sha": "abc123",
|
||||
"porcelain_status": "",
|
||||
},
|
||||
):
|
||||
try:
|
||||
mcp_server.verify_preflight_purity(
|
||||
remote="prgs",
|
||||
task="reconcile_merged_cleanups",
|
||||
)
|
||||
except RuntimeError as e:
|
||||
self.fail(
|
||||
"verify_preflight_purity raised RuntimeError "
|
||||
f"unexpectedly for reconciler: {e}"
|
||||
)
|
||||
|
||||
@patch.dict(
|
||||
os.environ,
|
||||
{"GITEA_PROFILE_NAME": "prgs-author", "GITEA_TEST_PORCELAIN": ""},
|
||||
clear=True,
|
||||
)
|
||||
@patch("mcp_server.get_profile", return_value=AUTHOR_PROFILE)
|
||||
def test_author_role_remains_blocked_on_root_checkout(self, _profile):
|
||||
mcp_server.record_preflight_check("whoami")
|
||||
mcp_server.record_preflight_check(
|
||||
"capability",
|
||||
resolved_role="author",
|
||||
resolved_task="reconcile_merged_cleanups",
|
||||
)
|
||||
|
||||
with patch("mcp_server.PROJECT_ROOT", CONTROL_ROOT):
|
||||
with patch(
|
||||
"mcp_server.root_checkout_guard.resolve_remote_master_sha",
|
||||
return_value="abc123",
|
||||
):
|
||||
with patch(
|
||||
"mcp_server.issue_lock_worktree.read_worktree_git_state",
|
||||
return_value={
|
||||
"current_branch": "master",
|
||||
"head_sha": "abc123",
|
||||
"porcelain_status": "",
|
||||
},
|
||||
):
|
||||
with self.assertRaises(RuntimeError) as ctx:
|
||||
mcp_server.verify_preflight_purity(
|
||||
remote="prgs",
|
||||
task="reconcile_merged_cleanups",
|
||||
)
|
||||
message = str(ctx.exception)
|
||||
self.assertTrue(
|
||||
"stable control checkout" in message
|
||||
or "author mutation blocked" in message
|
||||
or "branches/" in message,
|
||||
msg=message,
|
||||
)
|
||||
|
||||
@patch.dict(os.environ, {"GITEA_PROFILE_NAME": "prgs-reconciler"}, clear=True)
|
||||
@patch("mcp_server.get_profile", return_value=RECONCILER_PROFILE)
|
||||
def test_open_unmerged_pr_heads_are_not_safe_cleanup_targets(self, _profile):
|
||||
"""AC5: active/unmerged author work must remain blocked from cleanup."""
|
||||
mcp_server.record_preflight_check("whoami")
|
||||
mcp_server.record_preflight_check(
|
||||
"capability",
|
||||
resolved_role="reconciler",
|
||||
resolved_task="reconcile_merged_cleanups",
|
||||
)
|
||||
|
||||
open_pr = {
|
||||
"number": 200,
|
||||
"title": "active work",
|
||||
"body": "Closes #200",
|
||||
"merged": False,
|
||||
"state": "open",
|
||||
"head": {"ref": "feat/issue-200", "sha": "livehead1"},
|
||||
}
|
||||
# Same head branch still open on another PR while a closed/merged PR
|
||||
# used the same branch name historically — open head must block delete.
|
||||
closed_merged = {
|
||||
"number": 199,
|
||||
"title": "older merge same branch name",
|
||||
"body": "Closes #199",
|
||||
"merged": True,
|
||||
"merged_at": "2026-07-01T12:00:00Z",
|
||||
"merge_commit_sha": "deadbeef",
|
||||
"head": {"ref": "feat/issue-200", "sha": "oldhead99"},
|
||||
}
|
||||
self.mock_api.side_effect = [
|
||||
[closed_merged], # closed
|
||||
[open_pr], # open
|
||||
]
|
||||
|
||||
with patch("mcp_server._remote_branch_exists", return_value=True):
|
||||
with patch(
|
||||
"mcp_server.merged_cleanup_reconcile.is_head_ancestor_of_ref",
|
||||
return_value=True,
|
||||
):
|
||||
result = mcp_server.gitea_reconcile_merged_cleanups(
|
||||
dry_run=True,
|
||||
remote="prgs",
|
||||
)
|
||||
self.assertTrue(result["success"])
|
||||
self.assertEqual(len(result["entries"]), 1)
|
||||
remote_assessment = result["entries"][0].get("remote_branch") or {}
|
||||
self.assertFalse(
|
||||
remote_assessment.get("safe_to_delete_remote"),
|
||||
msg=f"open head must block remote delete: {remote_assessment}",
|
||||
)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,243 @@
|
||||
"""Tests for canonical state handoff ledger (#494)."""
|
||||
import sys
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
||||
|
||||
from final_report_validator import assess_final_report_validator # noqa: E402
|
||||
from state_handoff_ledger import ( # noqa: E402
|
||||
ISSUE_STATE_FIELDS,
|
||||
assess_contradictory_state_handoff,
|
||||
assess_final_report_next_action_handoff,
|
||||
assess_state_comment,
|
||||
assess_state_handoff_ledger_report,
|
||||
parse_state_comment,
|
||||
render_issue_state_comment,
|
||||
render_pr_state_comment,
|
||||
render_queue_controller_report,
|
||||
)
|
||||
|
||||
|
||||
def _work_issue_handoff(**overrides):
|
||||
lines = [
|
||||
"## Controller Handoff",
|
||||
"",
|
||||
"- Task: work-issue",
|
||||
"- Repo: Scaled-Tech-Consulting/Gitea-Tools",
|
||||
"- Role: author",
|
||||
"- Identity: jcwalker3 / prgs-author",
|
||||
"- Selected issue: #494",
|
||||
"- Current status: implementation complete; PR opened awaiting review",
|
||||
"- Next actor: reviewer",
|
||||
"- Next action: review PR at pinned head",
|
||||
"- Next prompt: Review PR #500 for issue #494 at current head in reviewer worktree",
|
||||
"- Safe next action: switch to reviewer session",
|
||||
"- External-state mutations: none",
|
||||
]
|
||||
text = "\n".join(lines)
|
||||
for key, value in overrides.items():
|
||||
needle = f"- {key}:"
|
||||
if needle in text:
|
||||
prefix = text.split(needle)[0]
|
||||
rest = text.split(needle, 1)[1]
|
||||
suffix = rest.split("\n", 1)[1] if "\n" in rest else ""
|
||||
text = f"{prefix}{needle} {value}\n{suffix}".rstrip()
|
||||
else:
|
||||
text += f"\n- {key}: {value}"
|
||||
return text
|
||||
|
||||
|
||||
class TestStateCommentTemplates(unittest.TestCase):
|
||||
def test_render_issue_state_includes_all_fields(self):
|
||||
body = render_issue_state_comment(
|
||||
STATE="issue_ready_for_author",
|
||||
NEXT_ACTOR="author",
|
||||
NEXT_ACTION="lock and implement",
|
||||
NEXT_PROMPT="Find issue #494 and open a PR",
|
||||
LAST_UPDATED_BY="jcwalker3",
|
||||
)
|
||||
parsed = parse_state_comment(body)
|
||||
for field in ISSUE_STATE_FIELDS:
|
||||
self.assertIn(field, parsed)
|
||||
|
||||
def test_render_pr_state_parses(self):
|
||||
body = render_pr_state_comment(
|
||||
STATE="needs_review",
|
||||
NEXT_ACTOR="reviewer",
|
||||
NEXT_ACTION="review at head",
|
||||
NEXT_PROMPT="Review PR #500",
|
||||
ISSUE="#494",
|
||||
HEAD_SHA="a" * 40,
|
||||
LAST_UPDATED_BY="sysadmin",
|
||||
)
|
||||
parsed = parse_state_comment(body)
|
||||
self.assertEqual(parsed["STATE"], "needs_review")
|
||||
self.assertEqual(parsed["NEXT_ACTOR"], "reviewer")
|
||||
|
||||
def test_queue_controller_template(self):
|
||||
body = render_queue_controller_report(
|
||||
QUEUE_STATE="open_prs_need_review",
|
||||
SELECTED_OBJECT="PR #500",
|
||||
SELECTED_OBJECT_TYPE="pr",
|
||||
NEXT_ACTOR="reviewer",
|
||||
NEXT_ACTION="start review session",
|
||||
NEXT_PROMPT="Review PR #500",
|
||||
LAST_UPDATED_BY="controller",
|
||||
)
|
||||
result = assess_state_comment(body, kind="queue_controller")
|
||||
self.assertTrue(result["complete"])
|
||||
|
||||
|
||||
class TestStateCommentValidation(unittest.TestCase):
|
||||
def test_complete_issue_state_comment(self):
|
||||
body = render_issue_state_comment(
|
||||
STATE="in_progress",
|
||||
NEXT_ACTOR="author",
|
||||
NEXT_ACTION="push branch",
|
||||
NEXT_PROMPT="Continue issue #494",
|
||||
STATUS="open",
|
||||
RELATED_PRS="none",
|
||||
BRANCH="feat/issue-494-state-handoff-ledger",
|
||||
HEAD_SHA="b" * 40,
|
||||
VALIDATION="pytest passed",
|
||||
BLOCKERS="none",
|
||||
DUPLICATES_OR_SUPERSEDES="none",
|
||||
LAST_UPDATED_BY="jcwalker3",
|
||||
)
|
||||
result = assess_state_comment(body, kind="issue_state")
|
||||
self.assertTrue(result["complete"])
|
||||
self.assertFalse(result["block"])
|
||||
|
||||
def test_missing_fields_blocked(self):
|
||||
body = "STATE: open\nNEXT_ACTOR: author"
|
||||
result = assess_state_comment(body, kind="issue_state")
|
||||
self.assertFalse(result["complete"])
|
||||
self.assertIn("NEXT_ACTION", result["missing_fields"])
|
||||
|
||||
def test_discussion_requires_five_comments_by_default(self):
|
||||
body = render_issue_state_comment(
|
||||
STATE="collecting_feedback",
|
||||
NEXT_ACTOR="controller",
|
||||
NEXT_ACTION="facilitate discussion",
|
||||
NEXT_PROMPT="Add acceptance criteria",
|
||||
STATUS="open",
|
||||
RELATED_PRS="none",
|
||||
BRANCH="none",
|
||||
HEAD_SHA="none",
|
||||
VALIDATION="none",
|
||||
BLOCKERS="none",
|
||||
DUPLICATES_OR_SUPERSEDES="none",
|
||||
LAST_UPDATED_BY="controller",
|
||||
)
|
||||
# Re-map to discussion fields manually for comment-count test
|
||||
from state_handoff_ledger import render_discussion_state_comment
|
||||
|
||||
disc = render_discussion_state_comment(
|
||||
STATE="collecting_feedback",
|
||||
NEXT_ACTOR="controller",
|
||||
NEXT_ACTION="wait for comments",
|
||||
NEXT_PROMPT="Continue discussion",
|
||||
COMMENT_COUNT="2",
|
||||
SUBSTANTIVE_COMMENTS="yes",
|
||||
URGENCY="normal",
|
||||
BLOCKERS="none",
|
||||
LAST_UPDATED_BY="controller",
|
||||
)
|
||||
result = assess_state_comment(disc, kind="discussion_state")
|
||||
self.assertFalse(result["complete"])
|
||||
self.assertTrue(any("5" in reason for reason in result["reasons"]))
|
||||
|
||||
def test_discussion_urgent_exception(self):
|
||||
from state_handoff_ledger import render_discussion_state_comment
|
||||
|
||||
disc = render_discussion_state_comment(
|
||||
STATE="ready_for_summary",
|
||||
NEXT_ACTOR="controller",
|
||||
NEXT_ACTION="summarize",
|
||||
NEXT_PROMPT="Post summary",
|
||||
COMMENT_COUNT="2",
|
||||
SUBSTANTIVE_COMMENTS="yes",
|
||||
URGENCY="urgent",
|
||||
BLOCKERS="none",
|
||||
LAST_UPDATED_BY="controller",
|
||||
)
|
||||
result = assess_state_comment(disc, kind="discussion_state")
|
||||
self.assertTrue(result["complete"])
|
||||
|
||||
|
||||
class TestFinalReportNextAction(unittest.TestCase):
|
||||
def test_complete_next_action_handoff(self):
|
||||
report = _work_issue_handoff()
|
||||
result = assess_final_report_next_action_handoff(report)
|
||||
self.assertTrue(result["complete"])
|
||||
self.assertFalse(result["block"])
|
||||
|
||||
def test_missing_next_prompt_blocked(self):
|
||||
report = _work_issue_handoff(**{"Next prompt": "none"})
|
||||
result = assess_final_report_next_action_handoff(report)
|
||||
self.assertFalse(result["complete"])
|
||||
self.assertIn("Next prompt", result["missing_fields"])
|
||||
|
||||
def test_missing_handoff_section_blocked(self):
|
||||
result = assess_final_report_next_action_handoff("no handoff here")
|
||||
self.assertTrue(result["block"])
|
||||
|
||||
|
||||
class TestContradictoryState(unittest.TestCase):
|
||||
def test_ready_to_merge_without_approval_blocked(self):
|
||||
report = _work_issue_handoff(
|
||||
**{
|
||||
"Current status": "PR ready to merge",
|
||||
"Review decision": "not attempted",
|
||||
}
|
||||
)
|
||||
result = assess_contradictory_state_handoff(report)
|
||||
self.assertTrue(result["block"])
|
||||
|
||||
def test_ready_to_merge_with_approval_allowed(self):
|
||||
report = _work_issue_handoff(
|
||||
**{
|
||||
"Current status": "PR ready to merge",
|
||||
"Review decision": "approve",
|
||||
}
|
||||
)
|
||||
result = assess_contradictory_state_handoff(report)
|
||||
self.assertFalse(result["block"])
|
||||
|
||||
def test_issue_done_without_pr_proof_blocked(self):
|
||||
report = _work_issue_handoff(**{"Current status": "issue complete"})
|
||||
result = assess_contradictory_state_handoff(report)
|
||||
self.assertTrue(result["block"])
|
||||
|
||||
def test_external_none_with_lock_activity_blocked(self):
|
||||
report = _work_issue_handoff(
|
||||
**{
|
||||
"External-state mutations": "none",
|
||||
}
|
||||
)
|
||||
report += "\n- Issue mutations: gitea_lock_issue adopted branch"
|
||||
result = assess_contradictory_state_handoff(report)
|
||||
self.assertTrue(result["block"])
|
||||
|
||||
def test_discussion_complete_without_summary_blocked(self):
|
||||
report = _work_issue_handoff(**{"Current status": "discussion complete"})
|
||||
result = assess_contradictory_state_handoff(report)
|
||||
self.assertTrue(result["block"])
|
||||
|
||||
|
||||
class TestFinalReportValidatorIntegration(unittest.TestCase):
|
||||
def test_work_issue_validator_flags_missing_next_prompt(self):
|
||||
report = _work_issue_handoff(**{"Next prompt": ""})
|
||||
result = assess_final_report_validator(report, "work_issue")
|
||||
rule_ids = [f["rule_id"] for f in result["findings"]]
|
||||
self.assertIn("shared.state_handoff_next_action", rule_ids)
|
||||
|
||||
def test_work_issue_validator_accepts_complete_handoff(self):
|
||||
report = _work_issue_handoff()
|
||||
result = assess_state_handoff_ledger_report(report)
|
||||
self.assertTrue(result["complete"])
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user