freemo
92c83ecc7e
CI / benchmark-publish (pull_request) Has been skipped
CI / lint (pull_request) Successful in 16s
CI / build (pull_request) Successful in 17s
CI / quality (pull_request) Successful in 19s
CI / typecheck (pull_request) Successful in 33s
CI / security (pull_request) Successful in 35s
CI / integration_tests (pull_request) Successful in 2m41s
CI / unit_tests (pull_request) Successful in 6m18s
CI / docker (pull_request) Successful in 1m2s
CI / benchmark-regression (pull_request) Successful in 15m41s
CI / coverage (pull_request) Failing after 20m52s
117 lines
3.8 KiB
Python
117 lines
3.8 KiB
Python
"""ASV benchmarks for plan phase transition validation overhead.
|
|
|
|
Measures the cost of:
|
|
- ``can_transition()`` lookups
|
|
- ``Plan.get_next_phase()`` traversals
|
|
- ``Plan.is_terminal`` property checks
|
|
- ``Plan.can_transition_to_next_phase`` property checks
|
|
"""
|
|
|
|
from cleveragents.domain.models.core.plan import (
|
|
NamespacedName,
|
|
Plan,
|
|
PlanIdentity,
|
|
PlanPhase,
|
|
ProcessingState,
|
|
can_transition,
|
|
)
|
|
|
|
|
|
def _make_plan(
|
|
phase: PlanPhase = PlanPhase.STRATEGIZE,
|
|
processing_state: ProcessingState = ProcessingState.QUEUED,
|
|
) -> Plan:
|
|
return Plan(
|
|
identity=PlanIdentity(plan_id="01ARZ3NDEKTSV4RRFFQ69G5FAV"),
|
|
namespaced_name=NamespacedName(server=None, namespace="local", name="bench"),
|
|
description="Benchmark plan",
|
|
action_name="local/bench-action",
|
|
phase=phase,
|
|
processing_state=processing_state,
|
|
)
|
|
|
|
|
|
class TimeCanTransition:
|
|
"""Benchmark ``can_transition()`` function."""
|
|
|
|
def time_valid_transition_strategize_to_execute(self) -> None:
|
|
can_transition(PlanPhase.STRATEGIZE, PlanPhase.EXECUTE)
|
|
|
|
def time_valid_transition_action_to_strategize(self) -> None:
|
|
can_transition(PlanPhase.ACTION, PlanPhase.STRATEGIZE)
|
|
|
|
def time_invalid_transition_strategize_to_apply(self) -> None:
|
|
can_transition(PlanPhase.STRATEGIZE, PlanPhase.APPLY)
|
|
|
|
def time_revert_execute_to_strategize(self) -> None:
|
|
can_transition(PlanPhase.EXECUTE, PlanPhase.STRATEGIZE)
|
|
|
|
def time_revert_apply_to_strategize(self) -> None:
|
|
can_transition(PlanPhase.APPLY, PlanPhase.STRATEGIZE)
|
|
|
|
|
|
class TimeGetNextPhase:
|
|
"""Benchmark ``Plan.get_next_phase()``."""
|
|
|
|
def setup(self) -> None:
|
|
self.plan_action = _make_plan(PlanPhase.ACTION)
|
|
self.plan_strategize = _make_plan(PlanPhase.STRATEGIZE)
|
|
self.plan_execute = _make_plan(PlanPhase.EXECUTE)
|
|
self.plan_apply = _make_plan(PlanPhase.APPLY)
|
|
|
|
def time_next_from_action(self) -> None:
|
|
self.plan_action.get_next_phase()
|
|
|
|
def time_next_from_strategize(self) -> None:
|
|
self.plan_strategize.get_next_phase()
|
|
|
|
def time_next_from_execute(self) -> None:
|
|
self.plan_execute.get_next_phase()
|
|
|
|
def time_next_from_apply(self) -> None:
|
|
self.plan_apply.get_next_phase()
|
|
|
|
|
|
class TimeIsTerminal:
|
|
"""Benchmark ``Plan.is_terminal`` property."""
|
|
|
|
def setup(self) -> None:
|
|
self.plan_queued = _make_plan(PlanPhase.STRATEGIZE, ProcessingState.QUEUED)
|
|
self.plan_applied = _make_plan(PlanPhase.APPLY, ProcessingState.APPLIED)
|
|
self.plan_constrained = _make_plan(PlanPhase.APPLY, ProcessingState.CONSTRAINED)
|
|
self.plan_cancelled = _make_plan(
|
|
PlanPhase.STRATEGIZE, ProcessingState.CANCELLED
|
|
)
|
|
|
|
def time_non_terminal(self) -> None:
|
|
_ = self.plan_queued.is_terminal
|
|
|
|
def time_applied_terminal(self) -> None:
|
|
_ = self.plan_applied.is_terminal
|
|
|
|
def time_constrained_terminal(self) -> None:
|
|
_ = self.plan_constrained.is_terminal
|
|
|
|
def time_cancelled_terminal(self) -> None:
|
|
_ = self.plan_cancelled.is_terminal
|
|
|
|
|
|
class TimeCanTransitionToNextPhase:
|
|
"""Benchmark ``Plan.can_transition_to_next_phase`` property."""
|
|
|
|
def setup(self) -> None:
|
|
self.plan_action = _make_plan(PlanPhase.ACTION, ProcessingState.QUEUED)
|
|
self.plan_strat_complete = _make_plan(
|
|
PlanPhase.STRATEGIZE, ProcessingState.COMPLETE
|
|
)
|
|
self.plan_apply = _make_plan(PlanPhase.APPLY, ProcessingState.QUEUED)
|
|
|
|
def time_action_can_transition(self) -> None:
|
|
_ = self.plan_action.can_transition_to_next_phase
|
|
|
|
def time_strategize_complete_can_transition(self) -> None:
|
|
_ = self.plan_strat_complete.can_transition_to_next_phase
|
|
|
|
def time_apply_cannot_transition(self) -> None:
|
|
_ = self.plan_apply.can_transition_to_next_phase
|