ed
0506c5da63
TIER-2 READ AGENTS.md, conductor/workflow.md, conductor/edit_workflow.md,
conductor/tier2/githooks/forbidden-files.txt,
conductor/tracks/tier2_leak_prevention_20260620/spec.md,
conductor/code_styleguides/data_oriented_design.md,
conductor/code_styleguides/error_handling.md,
conductor/code_styleguides/type_aliases.md before Phase 1.
Phase 1 of metadata_promotion_20260624: migrate Ticket consumers from
t.get('key', default) / t['key'] to direct field access (t.id, t.status, etc.).
Changes:
- self.active_tickets: list[Metadata] -> list[models.Ticket]
- _deserialize_active_track_result populates self.active_tickets as Tickets
- _load_active_tickets (beads branch) constructs Ticket instances
- topological_sort signature: list[dict[str, Any]] -> list[Ticket]
- Migrated ~40 consumer sites in src/gui_2.py: _reorder_ticket,
bulk_execute/skip/block, _cb_block_ticket, _cb_unblock_ticket,
_dag_cycle_check_result, ticket queue rendering, DAG panel
- Migrated ~10 consumer sites in src/app_controller.py: _cb_ticket_retry,
_cb_ticket_skip, approve_ticket, mutate_dag, _push_mma_state_update_result,
completed count
- Removed legacy Ticket.get() compat method (Task 1.5)
- Added tests/test_metadata_promotion_phase1.py with 15 regression-guard tests
- Updated existing tests to construct Ticket instances instead of dicts
Verified: 1885 of 1910 unit tests pass (25 pre-existing failures unrelated
to Ticket migration; many are live_gui/sim tests that need a running GUI).
86 lines
3.6 KiB
Python
86 lines
3.6 KiB
Python
import unittest
|
|
from unittest.mock import patch
|
|
from src import conductor_tech_lead
|
|
from src.models import Ticket
|
|
from src.result_types import Result
|
|
import pytest
|
|
|
|
class TestConductorTechLead(unittest.TestCase):
|
|
def test_generate_tickets_retry_failure(self) -> None:
|
|
with patch('src.ai_client.send') as mock_send:
|
|
mock_send.return_value = Result(data="invalid json")
|
|
# conductor_tech_lead.generate_tickets now raises RuntimeError on error after 3 attempts
|
|
with pytest.raises(RuntimeError):
|
|
conductor_tech_lead.generate_tickets("brief", "skeletons")
|
|
assert mock_send.call_count == 3
|
|
|
|
def test_generate_tickets_retry_success(self) -> None:
|
|
with patch('src.ai_client.send') as mock_send:
|
|
mock_send.side_effect = [Result(data="invalid json"), Result(data='[{"Task": "Test"}]')]
|
|
tickets = conductor_tech_lead.generate_tickets("brief", "skeletons")
|
|
assert tickets == [{"Task": "Test"}]
|
|
assert mock_send.call_count == 2
|
|
|
|
def test_generate_tickets_success(self) -> None:
|
|
with patch('src.ai_client.send') as mock_send:
|
|
mock_send.return_value = Result(data='[{"id": "T1", "description": "desc", "depends_on": []}]')
|
|
tickets = conductor_tech_lead.generate_tickets("brief", "skeletons")
|
|
self.assertEqual(len(tickets), 1)
|
|
self.assertEqual(tickets[0]['id'], "T1")
|
|
|
|
class TestTopologicalSort(unittest.TestCase):
|
|
def test_topological_sort_linear(self) -> None:
|
|
tickets = [
|
|
Ticket(id="t2", description="t2", depends_on=["t1"]),
|
|
Ticket(id="t1", description="t1", depends_on=[]),
|
|
]
|
|
sorted_tickets = conductor_tech_lead.topological_sort(tickets)
|
|
self.assertEqual(sorted_tickets[0].id, "t1")
|
|
self.assertEqual(sorted_tickets[1].id, "t2")
|
|
|
|
def test_topological_sort_complex(self) -> None:
|
|
tickets = [
|
|
Ticket(id="t3", description="t3", depends_on=["t1", "t2"]),
|
|
Ticket(id="t1", description="t1", depends_on=[]),
|
|
Ticket(id="t2", description="t2", depends_on=["t1"]),
|
|
]
|
|
sorted_tickets = conductor_tech_lead.topological_sort(tickets)
|
|
self.assertEqual(sorted_tickets[0].id, "t1")
|
|
self.assertEqual(sorted_tickets[1].id, "t2")
|
|
self.assertEqual(sorted_tickets[2].id, "t3")
|
|
|
|
def test_topological_sort_cycle(self) -> None:
|
|
tickets = [
|
|
Ticket(id="t1", description="t1", depends_on=["t2"]),
|
|
Ticket(id="t2", description="t2", depends_on=["t1"]),
|
|
]
|
|
with self.assertRaises(ValueError) as cm:
|
|
conductor_tech_lead.topological_sort(tickets)
|
|
# Match against our new standard ValueError message
|
|
self.assertIn("Dependency cycle detected", str(cm.exception))
|
|
|
|
def test_topological_sort_empty(self) -> None:
|
|
self.assertEqual(conductor_tech_lead.topological_sort([]), [])
|
|
|
|
def test_topological_sort_missing_dependency(self) -> None:
|
|
# If a ticket depends on something not in the list, we should handle it or let it fail.
|
|
# The TrackDAG silently ignores missing dependencies, causing cycle detection to trigger.
|
|
tickets = [
|
|
Ticket(id="t1", description="t1", depends_on=["missing"]),
|
|
]
|
|
# Currently this raises ValueError due to cycle detection on incomplete sort
|
|
with self.assertRaises(ValueError):
|
|
conductor_tech_lead.topological_sort(tickets)
|
|
|
|
def test_topological_sort_vlog(vlogger) -> None:
|
|
tickets = [
|
|
Ticket(id="t2", description="t2", depends_on=["t1"]),
|
|
Ticket(id="t1", description="t1", depends_on=[]),
|
|
]
|
|
vlogger.log_state("Input Order", ["t2", "t1"], ["t2", "t1"])
|
|
sorted_tickets = conductor_tech_lead.topological_sort(tickets)
|
|
result_ids = [t.id for t in sorted_tickets]
|
|
vlogger.log_state("Sorted Order", "N/A", result_ids)
|
|
assert result_ids == ["t1", "t2"]
|
|
vlogger.finalize("Topological Sort Verification", "PASS", "Linear dependencies correctly ordered.")
|