Scaled-Tech-Consulting/Gitea-Tools
2
0
Fork 0
Code Issues 74 Pull Requests 9 Actions Packages Projects Releases Wiki Activity

Merge branch 'master' into feat/issue-523-reconciler-cleanup

Browse Source
This commit is contained in:
sysadmin
2026-07-08 22:44:45 -05:00
parent 32fc1719aa 066eec8477
commit 72ae320400
18 changed files with 1993 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_bootstrap_review_path_docs.py
+81
View File
@@ -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
tests/test_final_report_validator.py
+6
View File
@@ -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",
tests/test_merged_cleanup_reconcile.py
+285 -1
View File
@@ -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()
tests/test_premerge_baseline_proof.py
+112
View File
@@ -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()
tests/test_state_handoff_ledger.py
+243
View File
@@ -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()