import pytest from src.models import Ticket from src.dag_engine import TrackDAG from src.performance_monitor import get_monitor def test_dag_performance(): perf = get_monitor() perf.enabled = True num_tickets = 500 tickets = [] # Create a complex DAG: linear dependencies + branching for i in range(num_tickets): deps = [] if i > 0: # Linear dependency deps.append(f"ticket_{i-1}") if i > 10: # Branching dependency (binary tree-ish) deps.append(f"ticket_{i//2}") if i > 50: # More cross-links deps.append(f"ticket_{i-10}") tickets.append(Ticket( id=f"ticket_{i}", description=f"Ticket number {i}", depends_on=deps, status="todo" )) dag = TrackDAG(tickets[::-1]) # 1. Measure has_cycle() cycle_found = dag.has_cycle() assert cycle_found is False # 2. Measure topological_sort() sorted_ids = dag.topological_sort() assert len(sorted_ids) == num_tickets # 3. Measure cascade_blocks() tickets[0].status = "blocked" dag.cascade_blocks() # Verify all are blocked for t in tickets: assert t.status == "blocked", f"Ticket {t.id} failed to cascade block" # 4. Verify metrics were recorded metrics = perf.get_metrics() assert "time_dag_has_cycle_ms" in metrics assert "time_dag_topological_sort_ms" in metrics assert "time_dag_cascade_blocks_ms" in metrics # Output results for visibility when running with -s print(f"\n[PERF] DAG Performance (n={num_tickets}):") print(f" has_cycle: {metrics['time_dag_has_cycle_ms']:.4f} ms") print(f" topological_sort: {metrics['time_dag_topological_sort_ms']:.4f} ms") print(f" cascade_blocks: {metrics['time_dag_cascade_blocks_ms']:.4f} ms") def test_dag_edge_cases(): # Test cycle detection correctness t1 = Ticket(id="a", description="a", depends_on=["b"]) t2 = Ticket(id="b", description="b", depends_on=["c"]) t3 = Ticket(id="c", description="c", depends_on=["a"]) dag = TrackDAG([t1, t2, t3]) assert dag.has_cycle() is True with pytest.raises(ValueError, match="Dependency cycle detected"): dag.topological_sort() # Test empty DAG dag_empty = TrackDAG([]) assert dag_empty.has_cycle() is False assert dag_empty.topological_sort() == []