cleveragents-core/benchmarks/db_migration_plans_bench.py

"""ASV benchmarks for plan table ORM round-trip performance.

Measures the performance of:
- LifecyclePlanModel.from_domain() construction from Plan domain objects
- LifecyclePlanModel.to_domain() conversion back to domain objects
- Bulk plan model construction for migration-volume workloads
"""

from __future__ import annotations

import sys
from pathlib import Path

try:
    from cleveragents.domain.models.core.plan import (
        AutomationProfileProvenance,
        AutomationProfileRef,
        InvariantSource,
        NamespacedName,
        Plan,
        PlanIdentity,
        PlanInvariant,
        PlanPhase,
        ProcessingState,
        ProjectLink,
    )
    from cleveragents.infrastructure.database.models import LifecyclePlanModel
except ModuleNotFoundError:
    sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
    from cleveragents.domain.models.core.plan import (
        AutomationProfileProvenance,
        AutomationProfileRef,
        InvariantSource,
        NamespacedName,
        Plan,
        PlanIdentity,
        PlanInvariant,
        PlanPhase,
        ProcessingState,
        ProjectLink,
    )
    from cleveragents.infrastructure.database.models import LifecyclePlanModel


def _make_plan(suffix: str = "bench") -> Plan:
    """Create a fully-populated Plan for benchmarking."""
    return Plan(
        identity=PlanIdentity(
            plan_id=f"01ARZ3NDEKTSV4RRFFQ69G5F{suffix[:2].upper().ljust(2, 'A')}"
        ),
        namespaced_name=NamespacedName(
            server=None, namespace="local", name=f"plan-{suffix}"
        ),
        description="Benchmark plan for ORM round-trip testing",
        action_name="local/bench-action",
        definition_of_done="All benchmarks pass within time budget",
        phase=PlanPhase.STRATEGIZE,
        processing_state=ProcessingState.QUEUED,
        automation_profile=AutomationProfileRef(
            profile_name="org/bench-profile",
            provenance=AutomationProfileProvenance.ACTION,
        ),
        project_links=[
            ProjectLink(project_name="local/project-a", alias="main"),
            ProjectLink(project_name="local/project-b", alias="lib", read_only=True),
        ],
        invariants=[
            PlanInvariant(
                text="Global invariant",
                source=InvariantSource.GLOBAL,
            ),
            PlanInvariant(
                text="Action invariant",
                source=InvariantSource.ACTION,
            ),
            PlanInvariant(
                text="Plan invariant",
                source=InvariantSource.PLAN,
            ),
        ],
        strategy_actor="openai/gpt-4",
        execution_actor="openai/gpt-4",
        estimation_actor="openai/gpt-4",
        invariant_actor="openai/gpt-4",
        arguments={"target_coverage": 90, "framework": "pytest"},
        arguments_order=["target_coverage", "framework"],
        created_by="bench-user",
        tags=["benchmark", "test"],
        reusable=True,
        read_only=False,
    )


class PlanModelFromDomainSuite:
    """Benchmark LifecyclePlanModel.from_domain() construction."""

    def setup(self) -> None:
        self.plan = _make_plan()
        self.minimal_plan = Plan(
            identity=PlanIdentity(plan_id="01ARZ3NDEKTSV4RRFFQ69G5FAV"),
            namespaced_name=NamespacedName(
                server=None, namespace="local", name="min-plan"
            ),
            description="Minimal plan",
            action_name="local/min-action",
            phase=PlanPhase.STRATEGIZE,
            processing_state=ProcessingState.QUEUED,
        )

    def time_from_domain_full(self) -> None:
        """Convert a fully-populated Plan to ORM model."""
        LifecyclePlanModel.from_domain(self.plan)

    def time_from_domain_minimal(self) -> None:
        """Convert a minimal Plan to ORM model."""
        LifecyclePlanModel.from_domain(self.minimal_plan)


class PlanModelToDomainSuite:
    """Benchmark LifecyclePlanModel.to_domain() conversion."""

    def setup(self) -> None:
        self.full_model = LifecyclePlanModel.from_domain(_make_plan())
        self.minimal_model = LifecyclePlanModel.from_domain(
            Plan(
                identity=PlanIdentity(plan_id="01ARZ3NDEKTSV4RRFFQ69G5FAV"),
                namespaced_name=NamespacedName(
                    server=None, namespace="local", name="minimal"
                ),
                description="Minimal plan",
                action_name="local/min-action",
                phase=PlanPhase.STRATEGIZE,
                processing_state=ProcessingState.QUEUED,
            )
        )

    def time_to_domain_full(self) -> None:
        """Convert a fully-populated ORM model back to domain."""
        self.full_model.to_domain()

    def time_to_domain_minimal(self) -> None:
        """Convert a minimal ORM model back to domain."""
        self.minimal_model.to_domain()


class PlanModelBulkSuite:
    """Benchmark bulk plan model construction for migration volumes."""

    def setup(self) -> None:
        self.plans = [_make_plan(str(i)) for i in range(50)]

    def time_bulk_from_domain_50(self) -> None:
        """Convert 50 Plan domain objects to ORM models."""
        for plan in self.plans:
            LifecyclePlanModel.from_domain(plan)

    def time_bulk_round_trip_50(self) -> None:
        """Full round-trip: domain -> ORM -> domain for 50 plans."""
        for plan in self.plans:
            model = LifecyclePlanModel.from_domain(plan)
            model.to_domain()