test(benchmarks): add ASV benchmarks for domain module
CI / benchmark-publish (pull_request) Has been skipped
CI / benchmark-regression (pull_request) Failing after 40s
CI / lint (pull_request) Failing after 59s
CI / quality (pull_request) Successful in 1m5s
CI / push-validation (pull_request) Successful in 26s
CI / helm (pull_request) Successful in 30s
CI / build (pull_request) Successful in 48s
CI / security (pull_request) Successful in 1m35s
CI / typecheck (pull_request) Successful in 2m4s
CI / e2e_tests (pull_request) Failing after 5m6s
CI / integration_tests (pull_request) Successful in 8m19s
CI / unit_tests (pull_request) Successful in 10m25s
CI / coverage (pull_request) Has been skipped
CI / docker (pull_request) Has been skipped
CI / status-check (pull_request) Failing after 4s

Added comprehensive ASV (Airspeed Velocity) benchmarks for the domain layer,
measuring performance of:
- DomainBaseModel validation and serialization
- Context model operations
- Change and ChangeSet operations
- Plan model creation and serialization
- Project model operations
- High-volume domain model validation throughput

Benchmarks cover instantiation, serialization (model_dump and JSON), and
throughput testing with 100+ instances to identify performance regressions.

ISSUES CLOSED: #1925
This commit is contained in:
2026-05-03 00:55:54 +00:00
parent 3d7f576243
commit 82b3094ed7
+390
View File
@@ -0,0 +1,390 @@
"""ASV benchmarks for domain layer models and operations.
Measures the performance of:
- Domain model instantiation and validation
- Context model creation and updates
- Change and ChangeSet operations
- Plan model creation and serialization
- Project model operations
- DomainBaseModel validation and serialization
"""
from __future__ import annotations
import importlib
import sys
from pathlib import Path
# Ensure the local *source* tree is importable even when ASV has an
# older build of the package installed.
_SRC = str(Path(__file__).resolve().parents[1] / "src")
if _SRC not in sys.path:
sys.path.insert(0, _SRC)
# Force-reload so ASV picks up the source tree version.
import cleveragents # noqa: E402
importlib.reload(cleveragents)
from cleveragents.domain.models.base import DomainBaseModel # noqa: E402
from cleveragents.domain.models.core import ( # noqa: E402
Change,
ChangeSet,
ChangeType,
Context,
ContextFile,
ContextType,
Operation,
OperationType,
Plan,
PlanStatus,
Project,
)
# ---------------------------------------------------------------------------
# DomainBaseModel validation benchmarks
# ---------------------------------------------------------------------------
class DomainBaseModelValidationSuite:
"""Benchmark DomainBaseModel validation and instantiation."""
timeout = 60
def setup(self) -> None:
"""Set up test data for validation benchmarks."""
# Create a simple test model class
class TestModel(DomainBaseModel):
name: str
value: int
description: str | None = None
self.TestModel = TestModel
self.test_data = {
"name": " test_name ", # Will be stripped
"value": 42,
"description": " test description ",
}
def time_instantiate_simple_model(self) -> None:
"""Benchmark simple model instantiation."""
self.TestModel(**self.test_data)
def time_instantiate_with_whitespace_stripping(self) -> None:
"""Benchmark model instantiation with whitespace stripping."""
self.TestModel(
name=" padded_name ",
value=100,
description=" padded description ",
)
def time_model_dump(self) -> None:
"""Benchmark model_dump() serialization."""
model = self.TestModel(**self.test_data)
model.model_dump()
def time_model_dump_json(self) -> None:
"""Benchmark model_dump_json() JSON serialization."""
model = self.TestModel(**self.test_data)
model.model_dump_json()
# ---------------------------------------------------------------------------
# Context model benchmarks
# ---------------------------------------------------------------------------
class ContextModelSuite:
"""Benchmark Context model operations."""
timeout = 60
def setup(self) -> None:
"""Set up test data for context benchmarks."""
self.context_data = {
"id": "ctx-001",
"type": ContextType.REPOSITORY,
"path": "/path/to/repo",
"content": "sample content",
}
def time_create_context(self) -> None:
"""Benchmark Context instantiation."""
Context(**self.context_data)
def time_create_context_with_file(self) -> None:
"""Benchmark Context with ContextFile."""
ctx = Context(**self.context_data)
file = ContextFile(
path="/path/to/file.py",
content="def hello(): pass",
language="python",
)
# Simulate adding file to context
_ = (ctx, file)
def time_context_serialization(self) -> None:
"""Benchmark Context serialization."""
ctx = Context(**self.context_data)
ctx.model_dump()
def time_context_json_serialization(self) -> None:
"""Benchmark Context JSON serialization."""
ctx = Context(**self.context_data)
ctx.model_dump_json()
# ---------------------------------------------------------------------------
# Change and ChangeSet benchmarks
# ---------------------------------------------------------------------------
class ChangeSetOperationsSuite:
"""Benchmark Change and ChangeSet operations."""
timeout = 60
def setup(self) -> None:
"""Set up test data for change operations."""
self.change_data = {
"path": "/path/to/file.py",
"type": ChangeType.MODIFIED,
"content": "def new_function(): pass",
}
def time_create_change(self) -> None:
"""Benchmark Change instantiation."""
Change(**self.change_data)
def time_create_changeset(self) -> None:
"""Benchmark ChangeSet instantiation."""
changes = [
Change(
path=f"/path/to/file{i}.py",
type=ChangeType.MODIFIED,
content=f"content {i}",
)
for i in range(10)
]
ChangeSet(changes=changes)
def time_changeset_with_operations(self) -> None:
"""Benchmark ChangeSet with operations."""
changes = [
Change(
path=f"/path/to/file{i}.py",
type=ChangeType.MODIFIED,
content=f"content {i}",
)
for i in range(10)
]
operations = [
Operation(
type=OperationType.CREATE,
path=f"/path/to/new{i}.py",
description=f"Create file {i}",
)
for i in range(5)
]
changeset = ChangeSet(changes=changes)
_ = (changeset, operations)
def time_changeset_serialization(self) -> None:
"""Benchmark ChangeSet serialization."""
changes = [
Change(
path=f"/path/to/file{i}.py",
type=ChangeType.MODIFIED,
content=f"content {i}",
)
for i in range(10)
]
changeset = ChangeSet(changes=changes)
changeset.model_dump()
def time_changeset_json_serialization(self) -> None:
"""Benchmark ChangeSet JSON serialization."""
changes = [
Change(
path=f"/path/to/file{i}.py",
type=ChangeType.MODIFIED,
content=f"content {i}",
)
for i in range(10)
]
changeset = ChangeSet(changes=changes)
changeset.model_dump_json()
# ---------------------------------------------------------------------------
# Plan model benchmarks
# ---------------------------------------------------------------------------
class PlanModelSuite:
"""Benchmark Plan model operations."""
timeout = 60
def setup(self) -> None:
"""Set up test data for plan benchmarks."""
self.plan_data = {
"id": "plan-001",
"status": PlanStatus.PENDING,
"description": "Test plan",
}
def time_create_plan(self) -> None:
"""Benchmark Plan instantiation."""
Plan(**self.plan_data)
def time_plan_serialization(self) -> None:
"""Benchmark Plan serialization."""
plan = Plan(**self.plan_data)
plan.model_dump()
def time_plan_json_serialization(self) -> None:
"""Benchmark Plan JSON serialization."""
plan = Plan(**self.plan_data)
plan.model_dump_json()
def time_plan_with_changes(self) -> None:
"""Benchmark Plan with associated changes."""
plan = Plan(**self.plan_data)
changes = [
Change(
path=f"/path/to/file{i}.py",
type=ChangeType.MODIFIED,
content=f"content {i}",
)
for i in range(5)
]
changeset = ChangeSet(changes=changes)
_ = (plan, changeset)
# ---------------------------------------------------------------------------
# Project model benchmarks
# ---------------------------------------------------------------------------
class ProjectModelSuite:
"""Benchmark Project model operations."""
timeout = 60
def setup(self) -> None:
"""Set up test data for project benchmarks."""
self.project_data = {
"id": "proj-001",
"name": "Test Project",
"description": "A test project for benchmarking",
}
def time_create_project(self) -> None:
"""Benchmark Project instantiation."""
Project(**self.project_data)
def time_project_serialization(self) -> None:
"""Benchmark Project serialization."""
project = Project(**self.project_data)
project.model_dump()
def time_project_json_serialization(self) -> None:
"""Benchmark Project JSON serialization."""
project = Project(**self.project_data)
project.model_dump_json()
def time_project_with_contexts(self) -> None:
"""Benchmark Project with multiple contexts."""
project = Project(**self.project_data)
contexts = [
Context(
id=f"ctx-{i:03d}",
type=ContextType.REPOSITORY,
path=f"/path/to/repo{i}",
content=f"content {i}",
)
for i in range(5)
]
_ = (project, contexts)
# ---------------------------------------------------------------------------
# Domain model validation throughput
# ---------------------------------------------------------------------------
class DomainModelValidationThroughputSuite:
"""Benchmark high-volume domain model validation."""
timeout = 60
def time_create_100_contexts(self) -> None:
"""Benchmark creating 100 Context instances."""
for i in range(100):
Context(
id=f"ctx-{i:03d}",
type=ContextType.REPOSITORY,
path=f"/path/to/repo{i}",
content=f"content {i}",
)
def time_create_100_changes(self) -> None:
"""Benchmark creating 100 Change instances."""
for i in range(100):
Change(
path=f"/path/to/file{i}.py",
type=ChangeType.MODIFIED,
content=f"content {i}",
)
def time_create_100_plans(self) -> None:
"""Benchmark creating 100 Plan instances."""
for i in range(100):
Plan(
id=f"plan-{i:03d}",
status=PlanStatus.PENDING,
description=f"Plan {i}",
)
def time_create_100_projects(self) -> None:
"""Benchmark creating 100 Project instances."""
for i in range(100):
Project(
id=f"proj-{i:03d}",
name=f"Project {i}",
description=f"Project {i} description",
)
def time_serialize_100_contexts(self) -> None:
"""Benchmark serializing 100 Context instances."""
contexts = [
Context(
id=f"ctx-{i:03d}",
type=ContextType.REPOSITORY,
path=f"/path/to/repo{i}",
content=f"content {i}",
)
for i in range(100)
]
for ctx in contexts:
ctx.model_dump()
def time_serialize_100_changesets(self) -> None:
"""Benchmark serializing 100 ChangeSet instances."""
changesets = [
ChangeSet(
changes=[
Change(
path=f"/path/to/file{j}.py",
type=ChangeType.MODIFIED,
content=f"content {j}",
)
for j in range(5)
]
)
for i in range(100)
]
for cs in changesets:
cs.model_dump()