tegwick
bd1d05ba79
Add change detection, structural diff-based impact analysis, configurable-depth incremental recomputation with circular suppression, and impact debt tracking. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
365 lines
12 KiB
Python
365 lines
12 KiB
Python
"""
|
|
Unit tests for IncrementalExecutionEngine.
|
|
|
|
Tests recompute flow, depth control, circular suppression, and budget limits.
|
|
"""
|
|
|
|
import pytest
|
|
import tempfile
|
|
from pathlib import Path
|
|
from unittest.mock import MagicMock
|
|
|
|
from markitect.prompts.dependencies.models import DependencyEdge, EdgeType
|
|
from markitect.prompts.dependencies.repository import SQLiteDependencyRepository
|
|
from markitect.prompts.dependencies.queries import DependencyQueryService
|
|
from markitect.prompts.execution.models import PromptRun, RunConfig, RunStatus
|
|
from markitect.prompts.incremental.engine import IncrementalExecutionEngine
|
|
from markitect.prompts.incremental.models import (
|
|
ArtifactChange,
|
|
ChangeType,
|
|
ImpactDebt,
|
|
RecomputeConfig,
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def temp_db():
|
|
"""Create temporary database for testing."""
|
|
with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as f:
|
|
db_path = f.name
|
|
yield db_path
|
|
Path(db_path).unlink(missing_ok=True)
|
|
|
|
|
|
@pytest.fixture
|
|
def dep_repo(temp_db):
|
|
"""Create dependency repository."""
|
|
return SQLiteDependencyRepository(temp_db)
|
|
|
|
|
|
@pytest.fixture
|
|
def query_service(dep_repo):
|
|
"""Create DependencyQueryService."""
|
|
return DependencyQueryService(dep_repo)
|
|
|
|
|
|
@pytest.fixture
|
|
def engine(temp_db, query_service):
|
|
"""Create IncrementalExecutionEngine."""
|
|
return IncrementalExecutionEngine(temp_db, query_service)
|
|
|
|
|
|
def _create_edge(repo, src, tgt, run_id="run-1", edge_type=EdgeType.REQUIRES):
|
|
"""Helper to create and persist a dependency edge."""
|
|
edge = DependencyEdge.create(
|
|
source_artifact_id=src,
|
|
target_artifact_id=tgt,
|
|
run_id=run_id,
|
|
edge_type=edge_type,
|
|
)
|
|
return repo.create(edge)
|
|
|
|
|
|
def _make_change(artifact_id="art-1"):
|
|
"""Helper to create a test ArtifactChange."""
|
|
return ArtifactChange.create(
|
|
artifact_id=artifact_id,
|
|
old_digest="old-digest",
|
|
new_digest="new-digest",
|
|
change_type=ChangeType.MODIFIED,
|
|
)
|
|
|
|
|
|
class TestFindDependentsAtDepth:
|
|
"""Tests for BFS depth-controlled dependent finding."""
|
|
|
|
def test_depth_1_direct_only(self, dep_repo, engine):
|
|
"""Test depth=1 finds only direct dependents."""
|
|
# A -> B -> C (A depends on B, B depends on C)
|
|
_create_edge(dep_repo, "A", "B")
|
|
_create_edge(dep_repo, "B", "C")
|
|
|
|
# Dependents of C at depth 1: only B
|
|
dependents = engine.find_dependents_at_depth("C", max_depth=1)
|
|
assert dependents == {"B"}
|
|
|
|
def test_depth_2_transitive(self, dep_repo, engine):
|
|
"""Test depth=2 finds two levels of dependents."""
|
|
# A -> B -> C
|
|
_create_edge(dep_repo, "A", "B")
|
|
_create_edge(dep_repo, "B", "C")
|
|
|
|
# Dependents of C at depth 2: B and A
|
|
dependents = engine.find_dependents_at_depth("C", max_depth=2)
|
|
assert dependents == {"A", "B"}
|
|
|
|
def test_depth_0_returns_empty(self, dep_repo, engine):
|
|
"""Test depth=0 returns no dependents."""
|
|
_create_edge(dep_repo, "A", "B")
|
|
|
|
dependents = engine.find_dependents_at_depth("B", max_depth=0)
|
|
assert dependents == set()
|
|
|
|
def test_no_dependents(self, engine):
|
|
"""Test artifact with no dependents."""
|
|
dependents = engine.find_dependents_at_depth("isolated", max_depth=5)
|
|
assert dependents == set()
|
|
|
|
def test_diamond_dependents(self, dep_repo, engine):
|
|
"""Test diamond-shaped dependency graph."""
|
|
# A -> C, B -> C, D -> A, D -> B
|
|
_create_edge(dep_repo, "A", "C")
|
|
_create_edge(dep_repo, "B", "C")
|
|
_create_edge(dep_repo, "D", "A")
|
|
_create_edge(dep_repo, "D", "B")
|
|
|
|
dependents = engine.find_dependents_at_depth("C", max_depth=2)
|
|
assert dependents == {"A", "B", "D"}
|
|
|
|
|
|
class TestRecompute:
|
|
"""Tests for the recompute orchestration flow."""
|
|
|
|
def test_basic_recompute(self, dep_repo, engine):
|
|
"""Test basic recompute with execution callback."""
|
|
_create_edge(dep_repo, "A", "B")
|
|
|
|
change = _make_change("B")
|
|
mock_run = PromptRun.create(
|
|
template_id="template-1",
|
|
input_bundle_hash="hash-1",
|
|
)
|
|
|
|
def callback(run_id):
|
|
return mock_run
|
|
|
|
result = engine.recompute(
|
|
change,
|
|
config=RecomputeConfig(max_depth=1),
|
|
execution_callback=callback,
|
|
old_content="old",
|
|
new_content="new",
|
|
)
|
|
|
|
assert result.changed_artifact_id == "B"
|
|
assert result.total_dependents == 1
|
|
assert result.recomputed_count == 1
|
|
assert result.suppressed_count == 0
|
|
assert len(result.executed_run_ids) == 1
|
|
|
|
def test_dry_run_no_callback(self, dep_repo, engine):
|
|
"""Test recompute without callback records what would be recomputed."""
|
|
_create_edge(dep_repo, "A", "B")
|
|
|
|
change = _make_change("B")
|
|
result = engine.recompute(
|
|
change,
|
|
config=RecomputeConfig(max_depth=1),
|
|
old_content="old",
|
|
new_content="new",
|
|
)
|
|
|
|
assert result.recomputed_count == 1
|
|
assert result.executed_run_ids == ["A"]
|
|
|
|
def test_no_dependents(self, engine):
|
|
"""Test recompute with no dependents."""
|
|
change = _make_change("isolated")
|
|
result = engine.recompute(change)
|
|
|
|
assert result.total_dependents == 0
|
|
assert result.recomputed_count == 0
|
|
assert result.suppressed_count == 0
|
|
|
|
def test_depth_control(self, dep_repo, engine):
|
|
"""Test depth limiting controls recompute scope."""
|
|
# A -> B -> C
|
|
_create_edge(dep_repo, "A", "B")
|
|
_create_edge(dep_repo, "B", "C")
|
|
|
|
change = _make_change("C")
|
|
|
|
# Depth 1: only B
|
|
result1 = engine.recompute(
|
|
change,
|
|
config=RecomputeConfig(max_depth=1),
|
|
old_content="old",
|
|
new_content="new",
|
|
)
|
|
assert result1.total_dependents == 1
|
|
assert result1.recomputed_count == 1
|
|
|
|
# Depth 2: B and A
|
|
result2 = engine.recompute(
|
|
change,
|
|
config=RecomputeConfig(max_depth=2),
|
|
old_content="old",
|
|
new_content="new",
|
|
)
|
|
assert result2.total_dependents == 2
|
|
assert result2.recomputed_count == 2
|
|
|
|
|
|
class TestBudgetLimits:
|
|
"""Tests for recompute budget exhaustion."""
|
|
|
|
def test_budget_exhaustion(self, dep_repo, engine):
|
|
"""Test budget limit suppresses excess recomputes."""
|
|
# Create 5 dependents of C
|
|
for i in range(5):
|
|
_create_edge(dep_repo, f"dep-{i}", "C")
|
|
|
|
change = _make_change("C")
|
|
result = engine.recompute(
|
|
change,
|
|
config=RecomputeConfig(max_depth=1, max_recomputes=3),
|
|
old_content="old",
|
|
new_content="new",
|
|
)
|
|
|
|
assert result.total_dependents == 5
|
|
assert result.recomputed_count == 3
|
|
assert result.suppressed_count == 2
|
|
assert all(
|
|
d.suppression_reason == "budget_exhausted"
|
|
for d in result.suppressed
|
|
)
|
|
|
|
def test_budget_zero_suppresses_all(self, dep_repo, engine):
|
|
"""Test zero budget suppresses all recomputes."""
|
|
_create_edge(dep_repo, "A", "B")
|
|
|
|
change = _make_change("B")
|
|
result = engine.recompute(
|
|
change,
|
|
config=RecomputeConfig(max_depth=1, max_recomputes=0),
|
|
old_content="old",
|
|
new_content="new",
|
|
)
|
|
|
|
assert result.recomputed_count == 0
|
|
assert result.suppressed_count == 1
|
|
|
|
|
|
class TestCircularSuppression:
|
|
"""Tests for circular dependency suppression."""
|
|
|
|
def test_circular_dependency_suppressed(self, dep_repo, engine):
|
|
"""Test circular dependency is suppressed."""
|
|
# A -> B and B -> A (circular)
|
|
_create_edge(dep_repo, "A", "B")
|
|
_create_edge(dep_repo, "B", "A")
|
|
|
|
change = _make_change("B")
|
|
result = engine.recompute(
|
|
change,
|
|
config=RecomputeConfig(max_depth=1, suppress_circular=True),
|
|
old_content="old",
|
|
new_content="new",
|
|
)
|
|
|
|
assert result.total_dependents == 1 # A is a dependent of B
|
|
# A depends on B, and B depends on A — would_create_cycle(A, B) is True
|
|
assert result.suppressed_count == 1
|
|
assert result.suppressed[0].suppression_reason == "circular_dependency"
|
|
|
|
def test_circular_suppression_disabled(self, dep_repo, engine):
|
|
"""Test circular suppression can be disabled."""
|
|
_create_edge(dep_repo, "A", "B")
|
|
_create_edge(dep_repo, "B", "A")
|
|
|
|
change = _make_change("B")
|
|
result = engine.recompute(
|
|
change,
|
|
config=RecomputeConfig(max_depth=1, suppress_circular=False),
|
|
old_content="old",
|
|
new_content="new",
|
|
)
|
|
|
|
# With suppression disabled, circular deps are still recomputed
|
|
assert result.recomputed_count == 1
|
|
assert result.suppressed_count == 0
|
|
|
|
|
|
class TestThresholdSuppression:
|
|
"""Tests for impact threshold suppression."""
|
|
|
|
def test_below_threshold_suppressed(self, dep_repo, engine):
|
|
"""Test below-threshold changes are suppressed."""
|
|
_create_edge(dep_repo, "A", "B")
|
|
|
|
change = _make_change("B")
|
|
# High threshold, small change
|
|
result = engine.recompute(
|
|
change,
|
|
config=RecomputeConfig(max_depth=1, impact_threshold=0.9),
|
|
old_content="hello world",
|
|
new_content="hello World", # small change
|
|
)
|
|
|
|
assert result.suppressed_count == 1
|
|
assert result.suppressed[0].suppression_reason == "below_threshold"
|
|
|
|
def test_above_threshold_recomputed(self, dep_repo, engine):
|
|
"""Test above-threshold changes trigger recompute."""
|
|
_create_edge(dep_repo, "A", "B")
|
|
|
|
change = _make_change("B")
|
|
result = engine.recompute(
|
|
change,
|
|
config=RecomputeConfig(max_depth=1, impact_threshold=0.1),
|
|
old_content="completely old content here",
|
|
new_content="entirely new different stuff",
|
|
)
|
|
|
|
assert result.recomputed_count == 1
|
|
assert result.suppressed_count == 0
|
|
|
|
|
|
class TestDebtPersistence:
|
|
"""Tests for impact debt persistence."""
|
|
|
|
def test_debt_recorded_in_db(self, dep_repo, engine):
|
|
"""Test suppressed recomputes are persisted as debt."""
|
|
_create_edge(dep_repo, "A", "B")
|
|
_create_edge(dep_repo, "B", "A")
|
|
|
|
change = _make_change("B")
|
|
engine.recompute(
|
|
change,
|
|
config=RecomputeConfig(max_depth=1, suppress_circular=True),
|
|
old_content="old",
|
|
new_content="new",
|
|
)
|
|
|
|
debt = engine.get_debt_for_artifact("B")
|
|
assert len(debt) == 1
|
|
assert debt[0].suppression_reason == "circular_dependency"
|
|
|
|
def test_get_all_debt(self, dep_repo, engine):
|
|
"""Test retrieving all debt records."""
|
|
# Create two separate suppressed recomputes
|
|
_create_edge(dep_repo, "A", "B")
|
|
_create_edge(dep_repo, "B", "A")
|
|
_create_edge(dep_repo, "C", "D")
|
|
_create_edge(dep_repo, "D", "C")
|
|
|
|
change1 = _make_change("B")
|
|
engine.recompute(
|
|
change1,
|
|
config=RecomputeConfig(max_depth=1, suppress_circular=True),
|
|
old_content="old",
|
|
new_content="new",
|
|
)
|
|
|
|
change2 = _make_change("D")
|
|
engine.recompute(
|
|
change2,
|
|
config=RecomputeConfig(max_depth=1, suppress_circular=True),
|
|
old_content="old",
|
|
new_content="new",
|
|
)
|
|
|
|
all_debt = engine.get_all_debt()
|
|
assert len(all_debt) == 2
|