cleveragents-core/benchmarks/bench_audit_service.py

"""ASV benchmarks for audit service write throughput.

Measures the overhead of the AuditEventSubscriber pipeline:
EventBus.emit() -> redact_dict() -> AuditService.record() -> SQLite.
"""

from __future__ import annotations

import importlib
import sys
from pathlib import Path
from typing import ClassVar

_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 sqlalchemy import create_engine  # noqa: E402
from sqlalchemy.orm import sessionmaker  # noqa: E402

from cleveragents.application.services.audit_event_subscriber import (  # noqa: E402
    AuditEventSubscriber,
)
from cleveragents.application.services.audit_service import AuditService  # noqa: E402
from cleveragents.config.settings import Settings  # noqa: E402
from cleveragents.infrastructure.database.models import Base  # noqa: E402
from cleveragents.infrastructure.events.models import DomainEvent  # noqa: E402
from cleveragents.infrastructure.events.reactive import (  # noqa: E402
    ReactiveEventBus,
)
from cleveragents.infrastructure.events.types import EventType  # noqa: E402


def _make_audit_service() -> AuditService:
    """Create an in-memory AuditService for benchmarking."""
    Settings._instance = None
    settings = Settings(database_url="sqlite:///:memory:")
    engine = create_engine("sqlite:///:memory:", echo=False)
    Base.metadata.create_all(engine)
    session = sessionmaker(bind=engine)()
    return AuditService(settings=settings, session=session)


class AuditServiceBenchmarks:
    """Benchmarks for audit event recording throughput."""

    timeout = 120
    params: ClassVar[list[int]] = [1, 10, 100, 1000]
    param_names: ClassVar[list[str]] = ["num_events"]

    def setup(self, num_events: int) -> None:
        # Recreate fresh DB/bus per ASV timing round to prevent row
        # accumulation across repetitions (PERF-4).
        self.svc = _make_audit_service()
        self.bus = ReactiveEventBus()
        self.subscriber = AuditEventSubscriber(
            audit_service=self.svc,
            event_bus=self.bus,
        )
        # Pre-generate unique events so each iteration has a distinct
        # timestamp and correlation_id, avoiding benchmark skew from
        # reusing a single frozen object (PERF-3).
        self.events = [
            DomainEvent(
                event_type=EventType.PLAN_APPLIED,
                plan_id=f"BENCH-{i:04d}",
                details={"action_name": "local/benchmark"},
            )
            for i in range(num_events)
        ]

    def time_record_events(self, num_events: int) -> None:
        for event in self.events:
            self.bus.emit(event)


class AuditDirectRecordBenchmarks:
    """Benchmarks for direct AuditService.record() calls."""

    timeout = 120
    params: ClassVar[list[int]] = [1, 10, 100, 1000]
    param_names: ClassVar[list[str]] = ["num_events"]

    def setup(self, num_events: int) -> None:
        self.svc = _make_audit_service()

    def time_direct_record(self, num_events: int) -> None:
        for _ in range(num_events):
            self.svc.record(
                event_type="plan_applied",
                plan_id="BENCH-001",
                details={"action_name": "local/benchmark"},
            )


class AuditRedactionBenchmarks:
    """Benchmarks for the redaction overhead in audit recording."""

    timeout = 120

    def setup(self) -> None:
        self.svc = _make_audit_service()
        self.bus = ReactiveEventBus()
        self.subscriber = AuditEventSubscriber(
            audit_service=self.svc,
            event_bus=self.bus,
        )
        self.event_with_secrets = DomainEvent(
            event_type=EventType.CONFIG_CHANGED,
            details={
                "api_key": "sk-proj-ABCDEFGHIJ1234567890",
                "token": "tok_ABCDEF1234567890",
                "normal_field": "safe-value",
            },
        )

    def time_record_with_redaction(self) -> None:
        self.bus.emit(self.event_with_secrets)