"""ASV benchmarks for validation apply gate throughput. Measures the performance of: - ValidationAttachment creation - DefaultValidationRunner execution - ApplyValidationGate.run() with multiple attachments - ApplyValidationSummary.to_plan_metadata() serialization """ from __future__ import annotations import importlib import sys from pathlib import Path # Ensure the local *source* tree is importable. _SRC = str(Path(__file__).resolve().parents[1] / "src") if _SRC not in sys.path: sys.path.insert(0, _SRC) import cleveragents # noqa: E402 importlib.reload(cleveragents) from cleveragents.application.services.validation_apply import ( # noqa: E402 ApplyValidationGate, ApplyValidationResult, ApplyValidationSummary, DefaultValidationRunner, ValidationAttachment, ) from cleveragents.domain.models.core.tool import ValidationMode # noqa: E402 # --------------------------------------------------------------------------- # Benchmark: ValidationAttachment creation # --------------------------------------------------------------------------- class AttachmentCreation: """Benchmark suite for ``ValidationAttachment`` model creation.""" def time_create_required(self) -> None: """Create a required validation attachment.""" ValidationAttachment( attachment_id="01ATT000000000000000000000", validation_name="lint-check", resource_id="res-001", mode=ValidationMode.REQUIRED, ) def time_create_informational(self) -> None: """Create an informational validation attachment.""" ValidationAttachment( attachment_id="01ATT000000000000000000001", validation_name="style-check", resource_id="res-002", mode=ValidationMode.INFORMATIONAL, ) # --------------------------------------------------------------------------- # Benchmark: DefaultValidationRunner # --------------------------------------------------------------------------- _ATTACHMENT = ValidationAttachment( attachment_id="01ATT000000000000000000000", validation_name="lint-check", resource_id="res-001", mode=ValidationMode.REQUIRED, ) _CTX_PASS = {"lint-check": "passed"} _CTX_FAIL = {"status": "no matches"} class RunnerExecution: """Benchmark suite for ``DefaultValidationRunner``.""" def setup(self) -> None: """Prepare runner.""" self.runner = DefaultValidationRunner() def time_run_pass(self) -> None: """Run validation that passes.""" self.runner.run_validation(_ATTACHMENT, _CTX_PASS) def time_run_fail(self) -> None: """Run validation that fails.""" self.runner.run_validation(_ATTACHMENT, _CTX_FAIL) # --------------------------------------------------------------------------- # Benchmark: ApplyValidationGate # --------------------------------------------------------------------------- def _make_attachments(n: int) -> list[ValidationAttachment]: """Create N test attachments.""" return [ ValidationAttachment( attachment_id=f"01ATT{i:021d}", validation_name=f"val-{i}", resource_id=f"res-{i}", mode=ValidationMode.REQUIRED, ) for i in range(n) ] class GateExecution: """Benchmark suite for ``ApplyValidationGate.run``.""" def setup(self) -> None: """Prepare gate and attachments.""" self.runner = DefaultValidationRunner() self.gate = ApplyValidationGate(runner=self.runner) self.attachments_5 = _make_attachments(5) self.attachments_20 = _make_attachments(20) self.context = {f"val-{i}": "ok" for i in range(20)} def time_gate_5_attachments(self) -> None: """Run gate with 5 attachments.""" self.gate.run("01PLAN0001", self.attachments_5, self.context) def time_gate_20_attachments(self) -> None: """Run gate with 20 attachments.""" self.gate.run("01PLAN0001", self.attachments_20, self.context) # --------------------------------------------------------------------------- # Benchmark: Summary serialization # --------------------------------------------------------------------------- class SummarySerialization: """Benchmark suite for ``ApplyValidationSummary`` serialization.""" def setup(self) -> None: """Prepare summary with mixed results.""" results = [] for i in range(10): results.append( ApplyValidationResult( attachment_id=f"01ATT{i:021d}", validation_name=f"val-{i}", resource_id=f"res-{i}", mode=ValidationMode.REQUIRED, passed=i % 3 != 0, message="ok" if i % 3 != 0 else "failed", ) ) self.summary = ApplyValidationSummary( plan_id="01PLAN0001", results=results, ) def time_to_plan_metadata(self) -> None: """Serialize summary to plan metadata dict.""" self.summary.to_plan_metadata() def time_format_cli_output(self) -> None: """Format summary for CLI display.""" self.summary.format_cli_output()