CoreRasurae
165 lines
5.1 KiB
Python
165 lines
5.1 KiB
Python
"""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()
|