76 lines
2.1 KiB
Python
76 lines
2.1 KiB
Python
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() == []
|