import pytest from src.models import Ticket from src.dag_engine import TrackDAG def test_get_ready_tasks_linear(): t1 = Ticket(id="T1", description="desc", status="todo") t2 = Ticket(id="T2", description="desc", status="todo", depends_on=["T1"]) dag = TrackDAG([t1, t2]) ready = dag.get_ready_tasks() assert len(ready) == 1 assert ready[0].id == "T1" def test_get_ready_tasks_branching(): t1 = Ticket(id="T1", description="desc", status="completed") t2 = Ticket(id="T2", description="desc", status="todo", depends_on=["T1"]) t3 = Ticket(id="T3", description="desc", status="todo", depends_on=["T1"]) dag = TrackDAG([t1, t2, t3]) ready = dag.get_ready_tasks() assert len(ready) == 2 ids = [t.id for t in ready] assert "T2" in ids assert "T3" in ids def test_has_cycle_no_cycle(): t1 = Ticket(id="T1", description="desc", status="todo") t2 = Ticket(id="T2", description="desc", status="todo", depends_on=["T1"]) dag = TrackDAG([t1, t2]) assert dag.has_cycle() is False def test_has_cycle_direct_cycle(): t1 = Ticket(id="T1", description="desc", status="todo", depends_on=["T2"]) t2 = Ticket(id="T2", description="desc", status="todo", depends_on=["T1"]) dag = TrackDAG([t1, t2]) assert dag.has_cycle() is True def test_has_cycle_indirect_cycle(): t1 = Ticket(id="T1", description="desc", status="todo", depends_on=["T3"]) t2 = Ticket(id="T2", description="desc", status="todo", depends_on=["T1"]) t3 = Ticket(id="T3", description="desc", status="todo", depends_on=["T2"]) dag = TrackDAG([t1, t2, t3]) assert dag.has_cycle() is True def test_has_cycle_complex_no_cycle(): t1 = Ticket(id="T1", description="desc", status="todo") t2 = Ticket(id="T2", description="desc", status="todo", depends_on=["T1"]) t3 = Ticket(id="T3", description="desc", status="todo", depends_on=["T1"]) t4 = Ticket(id="T4", description="desc", status="todo", depends_on=["T2", "T3"]) dag = TrackDAG([t1, t2, t3, t4]) assert dag.has_cycle() is False def test_get_ready_tasks_multiple_deps(): t1 = Ticket(id="T1", description="desc", status="completed") t2 = Ticket(id="T2", description="desc", status="todo") t3 = Ticket(id="T3", description="desc", status="todo", depends_on=["T1", "T2"]) dag = TrackDAG([t1, t2, t3]) # Only T2 is ready because T3 depends on T2 (todo) ready = dag.get_ready_tasks() assert len(ready) == 1 assert ready[0].id == "T2" def test_topological_sort(): t1 = Ticket(id="T1", description="desc", status="todo") t2 = Ticket(id="T2", description="desc", status="todo", depends_on=["T1"]) dag = TrackDAG([t2, t1]) # Out of order input sorted_tasks = dag.topological_sort() assert [t.id for t in sorted_tasks] == ["T1", "T2"] def test_topological_sort_cycle(): t1 = Ticket(id="T1", description="desc", status="todo", depends_on=["T2"]) t2 = Ticket(id="T2", description="desc", status="todo", depends_on=["T1"]) dag = TrackDAG([t1, t2]) with pytest.raises(ValueError, match="DAG Validation Error: Cycle detected"): dag.topological_sort()