"""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()