sysadminandClaude Opus 4.8
d50328aea4
Add templates, validators, and documentation for discussion/issue/PR state comments and final-report next-action handoff fields. Wire enforcement into assess_final_report_validator across all workflow task kinds. Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
243 lines
8.9 KiB
Python
243 lines
8.9 KiB
Python
"""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() |