diff --git a/tests/test_issue_628_orchestration.py b/tests/test_issue_628_orchestration.py new file mode 100644 index 0000000..91e9437 --- /dev/null +++ b/tests/test_issue_628_orchestration.py @@ -0,0 +1,199 @@ +"""Regression tests for #628 building blocks (child scope only). + +Honest scope: unit coverage of pre-existing APIs used by umbrella #628. +This module does **not** implement or verify all 21 umbrella acceptance +criteria, automatic handoff store/retrieve, multi-worker product wiring, +or end-to-end orchestration. + +Covered building blocks: +- CTH format / parse / assess (`format_cth_body`, `parse_cth_comment`, + `assess_cth_comment`) +- Exclusive-ownership skip classification (`classify_skip` with + OWNERSHIP_FOREIGN vs OWNERSHIP_OWN) +- Durable dependency edges (`upsert_dependency_edge` / list) and skip + when dependency_unmet +- Edge state transition UNMET -> MET + +Parent umbrella remains #628; this slice is a scoped child issue only. +""" + +import unittest +from unittest.mock import MagicMock, patch +import os +import json +import tempfile + +from canonical_thread_handoff import ( + format_cth_body, + parse_cth_comment, + assess_cth_comment, +) +import dependency_graph +from control_plane_db import ControlPlaneDB +from allocator_service import ( + WorkCandidate, + classify_skip, + ROLE_AUTHOR, + ROLE_REVIEWER, + ROLE_MERGER, + ROLE_RECONCILER, + OWNERSHIP_OWN, + OWNERSHIP_FOREIGN, +) + + +class TestIssue628Orchestration(unittest.TestCase): + def setUp(self): + self._tmp = tempfile.TemporaryDirectory() + self.db_path = os.path.join(self._tmp.name, "cp.sqlite3") + self.db = ControlPlaneDB(self.db_path) + + def tearDown(self): + self._tmp.cleanup() + + def test_canonical_handoff_serialization_and_retrieval(self): + """Building block: format/parse/assess a CTH body (not full AC1/AC2 product path).""" + handoff = format_cth_body( + cth_type="Author Handoff", + status="completed", + next_owner="reviewer", + current_blocker="none", + decision="Implementation complete, tests passing", + proof="pytest tests/test_issue_628_orchestration.py passed", + next_action="Review PR and run reviewer pre-flight", + ready_to_paste_prompt="Review PR for child issue #878 (parent #628)", + ) + self.assertIn("CTH: Author Handoff", handoff) + + parsed = parse_cth_comment(handoff) + self.assertIsNotNone(parsed) + self.assertEqual(parsed["cth_type"], "Author Handoff") + + assessment = assess_cth_comment(handoff) + self.assertFalse(assessment["block"]) + + def test_exclusive_task_unit_single_owner(self): + """Building block: classify_skip foreign vs own ownership.""" + candidate = WorkCandidate( + kind="issue", + number=878, + title="Child #878 ownership classify candidate", + state="open", + labels=["status:in-progress"], + blocked=False, + dependency_unmet=False, + ) + # Foreign ownership MUST be skipped + skip_foreign = classify_skip( + c=candidate, + role=ROLE_AUTHOR, + terminal_pr=None, + claim_ownership=OWNERSHIP_FOREIGN, + ) + self.assertIsNotNone(skip_foreign) + self.assertIn("active lease", skip_foreign) + + # Own/Self claim remains selectable for session resumption + skip_self = classify_skip( + c=candidate, + role=ROLE_AUTHOR, + terminal_pr=None, + claim_ownership=OWNERSHIP_OWN, + ) + self.assertIsNone(skip_self) + + def test_durable_dependency_graph_blocking(self): + """Building block: unmet dependency edge + classify_skip on dependency_unmet.""" + # Upsert a blocking dependency edge between issue 878 and blocker 601 + self.db.upsert_dependency_edge( + remote="prgs", + org="Scaled-Tech-Consulting", + repo="Gitea-Tools", + source_kind="issue", + source_number=878, + target_kind="issue", + target_number=601, + edge_type=dependency_graph.EDGE_ISSUE_BLOCKED_BY_ISSUE, + state=dependency_graph.STATE_UNMET, + blocking_condition="Target issue #601 is not closed", + completion_condition="Target issue #601 is closed", + evidence={"source": "unit_test"}, + ) + + edges = self.db.list_dependency_edges( + remote="prgs", + org="Scaled-Tech-Consulting", + repo="Gitea-Tools", + source_kind="issue", + source_number=878, + ) + self.assertEqual(len(edges), 1) + self.assertEqual(edges[0]["state"], "unmet") + self.assertEqual(edges[0]["target_number"], 601) + + # When dependency is unmet, candidate is blocked from selection + candidate = WorkCandidate( + kind="issue", + number=878, + title="Blocked candidate", + state="open", + labels=[], + blocked=False, + dependency_unmet=True, + dependency_reason="issue#878 is blocked by unmet dependency issue#601", + ) + skip_reason = classify_skip( + c=candidate, + role=ROLE_AUTHOR, + terminal_pr=None, + claim_ownership=OWNERSHIP_OWN, + ) + self.assertIsNotNone(skip_reason) + self.assertIn("issue#601", skip_reason) + + def test_dependency_completion_reevaluation(self): + """Building block: dependency edge state can transition UNMET -> MET.""" + self.db.upsert_dependency_edge( + remote="prgs", + org="Scaled-Tech-Consulting", + repo="Gitea-Tools", + source_kind="issue", + source_number=878, + target_kind="issue", + target_number=601, + edge_type=dependency_graph.EDGE_ISSUE_BLOCKED_BY_ISSUE, + state=dependency_graph.STATE_UNMET, + blocking_condition="Target issue #601 is open", + completion_condition="Target issue #601 is closed", + evidence={"source": "unit_test"}, + ) + + # Mark edge as met upon target issue closure + self.db.upsert_dependency_edge( + remote="prgs", + org="Scaled-Tech-Consulting", + repo="Gitea-Tools", + source_kind="issue", + source_number=878, + target_kind="issue", + target_number=601, + edge_type=dependency_graph.EDGE_ISSUE_BLOCKED_BY_ISSUE, + state=dependency_graph.STATE_MET, + blocking_condition="Target issue #601 is open", + completion_condition="Target issue #601 is closed", + evidence={"source": "target_closed_event"}, + ) + + edges = self.db.list_dependency_edges( + remote="prgs", + org="Scaled-Tech-Consulting", + repo="Gitea-Tools", + source_kind="issue", + source_number=878, + ) + self.assertEqual(len(edges), 1) + self.assertEqual(edges[0]["state"], "met") + + +if __name__ == "__main__": + unittest.main()