"""ASV benchmarks for invariant merge operations. Measures the performance of: - Invariant model construction (Pydantic validation) - merge_invariants() with varying list sizes - InvariantSet.merge() class method - InvariantService.get_effective_invariants() """ from __future__ import annotations import sys from pathlib import Path try: from cleveragents.application.services.invariant_service import InvariantService from cleveragents.domain.models.core.invariant import ( Invariant, InvariantScope, InvariantSet, merge_invariants, ) except ModuleNotFoundError: sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src")) from cleveragents.application.services.invariant_service import InvariantService from cleveragents.domain.models.core.invariant import ( Invariant, InvariantScope, InvariantSet, merge_invariants, ) def _make_invariants(scope: InvariantScope, count: int) -> list[Invariant]: """Create a list of invariants for benchmarking.""" return [ Invariant( text=f"Constraint {scope.value} {i}", scope=scope, source_name=f"source-{scope.value}", ) for i in range(count) ] class InvariantConstructionSuite: """Benchmark Invariant model construction.""" def time_invariant_construction(self) -> None: """Benchmark single invariant creation.""" Invariant( text="Never delete production data", scope=InvariantScope.GLOBAL, source_name="system", ) def time_invariant_construction_plan(self) -> None: """Benchmark plan-scoped invariant creation.""" Invariant( text="Use Python 3.13 only", scope=InvariantScope.PLAN, source_name="plan-001", ) class MergeSmallSuite: """Benchmark merge with small invariant lists.""" def setup(self) -> None: """Create small invariant lists.""" self.plan = _make_invariants(InvariantScope.PLAN, 3) self.action = _make_invariants(InvariantScope.ACTION, 2) self.project = _make_invariants(InvariantScope.PROJECT, 5) self.global_invs = _make_invariants(InvariantScope.GLOBAL, 5) def time_merge_small(self) -> None: """Benchmark merge with ~15 invariants.""" merge_invariants(self.plan, self.action, self.project, self.global_invs) class MergeMediumSuite: """Benchmark merge with medium invariant lists.""" def setup(self) -> None: """Create medium invariant lists.""" self.plan = _make_invariants(InvariantScope.PLAN, 10) self.action = _make_invariants(InvariantScope.ACTION, 5) self.project = _make_invariants(InvariantScope.PROJECT, 20) self.global_invs = _make_invariants(InvariantScope.GLOBAL, 20) def time_merge_medium(self) -> None: """Benchmark merge with ~55 invariants.""" merge_invariants(self.plan, self.action, self.project, self.global_invs) class MergeLargeSuite: """Benchmark merge with large invariant lists.""" def setup(self) -> None: """Create large invariant lists.""" self.plan = _make_invariants(InvariantScope.PLAN, 50) self.action = _make_invariants(InvariantScope.ACTION, 25) self.project = _make_invariants(InvariantScope.PROJECT, 100) self.global_invs = _make_invariants(InvariantScope.GLOBAL, 100) def time_merge_large(self) -> None: """Benchmark merge with ~275 invariants.""" merge_invariants(self.plan, self.action, self.project, self.global_invs) class MergeDeduplicationSuite: """Benchmark merge with heavy de-duplication.""" def setup(self) -> None: """Create invariant lists with many duplicates.""" self.plan = [ Invariant( text=f"Shared constraint {i}", scope=InvariantScope.PLAN, source_name="plan-001", ) for i in range(20) ] self.action = [ Invariant( text=f"Shared constraint {i}", scope=InvariantScope.ACTION, source_name="action-001", ) for i in range(20) ] self.project = [ Invariant( text=f"Shared constraint {i}", scope=InvariantScope.PROJECT, source_name="proj-001", ) for i in range(20) ] self.global_invs = [ Invariant( text=f"Shared constraint {i}", scope=InvariantScope.GLOBAL, source_name="system", ) for i in range(20) ] def time_merge_dedup(self) -> None: """Benchmark merge with 80 invariants, all duplicates.""" merge_invariants(self.plan, self.action, self.project, self.global_invs) class InvariantSetMergeSuite: """Benchmark InvariantSet.merge() class method.""" def setup(self) -> None: """Create invariant lists.""" self.plan = _make_invariants(InvariantScope.PLAN, 5) self.action = _make_invariants(InvariantScope.ACTION, 3) self.project = _make_invariants(InvariantScope.PROJECT, 10) self.global_invs = _make_invariants(InvariantScope.GLOBAL, 10) def time_invariant_set_merge(self) -> None: """Benchmark InvariantSet.merge().""" InvariantSet.merge(self.plan, self.action, self.project, self.global_invs) class ServiceEffectiveSuite: """Benchmark InvariantService.get_effective_invariants().""" def setup(self) -> None: """Populate service with invariants.""" self.service = InvariantService() for i in range(10): self.service.add_invariant(f"Global {i}", InvariantScope.GLOBAL, "system") for i in range(10): self.service.add_invariant(f"Project {i}", InvariantScope.PROJECT, "myapp") for i in range(5): self.service.add_invariant( f"Action {i}", InvariantScope.ACTION, "action-001" ) for i in range(5): self.service.add_invariant(f"Plan {i}", InvariantScope.PLAN, "plan-001") def time_get_effective(self) -> None: """Benchmark get_effective_invariants().""" self.service.get_effective_invariants( plan_id="plan-001", project_name="myapp", action_name="action-001" )