forked from HAL9000/cleveragents-core
freemo
a808c395f9
Add 53 new .feature files and corresponding step definition files targeting uncovered lines identified in build/coverage.xml. Fix AmbiguousStep conflicts in 7 pre-existing step files by disambiguating step text. New tests cover: ACP clients/facade, actor CLI/config, application container, ACMS service/strategies, async worker, automation profile CLI, autonomy guardrail, bridge, change model, config CLI/service, context service, cross-plan correction, database models, decision service, decomposition clustering/service, discovery handler, langchain chat provider, langgraph nodes, materializers, multi-project service, plan apply/CLI/lifecycle/model/ preflight/resume/service, PostgreSQL analyzer, project CLI/context CLI, provider registry, reactive application/route, repositories, resolver handler, resource registry service, resume model, retry patterns, sandbox protocol, server CLI, skill CLI/service, skills registry, subplan execution/service, system CLI, UKO loader, UoW, and YAML template engine. Closes #645
433 lines
15 KiB
Python
433 lines
15 KiB
Python
"""Step definitions for plan_apply_service_coverage_boost feature.
|
|
|
|
Targets uncovered lines in plan_apply_service.py:
|
|
- Lines 674-688: _try_checkpoint success and exception paths
|
|
- Lines 697-707: _try_rollback success, empty, and exception paths
|
|
- Lines 777, 782, 787: _ApplyPhaseProxy constructor, sandbox_id, context
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
from unittest.mock import MagicMock
|
|
|
|
from behave import given, then, when # type: ignore[import-untyped]
|
|
from behave.runner import Context # type: ignore[import-untyped]
|
|
|
|
from cleveragents.application.services.plan_apply_service import (
|
|
ApplyOutcome,
|
|
PlanApplyService,
|
|
_ApplyPhaseProxy,
|
|
)
|
|
from cleveragents.core.exceptions import PlanError
|
|
from cleveragents.domain.models.core.change import (
|
|
ChangeEntry,
|
|
ChangeOperation,
|
|
SpecChangeSet,
|
|
)
|
|
from cleveragents.domain.models.core.plan import (
|
|
PlanPhase,
|
|
PlanTimestamps,
|
|
ProcessingState,
|
|
)
|
|
|
|
__all__: list[str] = []
|
|
|
|
_PLAN_ID = "01COVERAGEBOOST0000001"
|
|
_CHANGESET_ID = "01CSBOOST00000000000001"
|
|
|
|
|
|
# ----------------------------------------------------------------------
|
|
# Helpers
|
|
# ----------------------------------------------------------------------
|
|
|
|
|
|
class _StubPhase:
|
|
"""Stub for PlanPhase with a .value attribute."""
|
|
|
|
def __init__(self, phase: PlanPhase) -> None:
|
|
self._phase = phase
|
|
self.value = phase.value
|
|
|
|
def __eq__(self, other: object) -> bool:
|
|
if isinstance(other, PlanPhase):
|
|
return self._phase == other
|
|
if isinstance(other, _StubPhase):
|
|
return self._phase == other._phase
|
|
return NotImplemented
|
|
|
|
|
|
class _StubState:
|
|
"""Stub for ProcessingState with a .value attribute."""
|
|
|
|
def __init__(self, state: ProcessingState) -> None:
|
|
self._state = state
|
|
self.value = state.value
|
|
|
|
def __eq__(self, other: object) -> bool:
|
|
if isinstance(other, ProcessingState):
|
|
return self._state == other
|
|
if isinstance(other, _StubState):
|
|
return self._state == other._state
|
|
return NotImplemented
|
|
|
|
|
|
def _make_mock_plan(
|
|
*,
|
|
plan_id: str = _PLAN_ID,
|
|
phase: PlanPhase = PlanPhase.EXECUTE,
|
|
state: ProcessingState = ProcessingState.COMPLETE,
|
|
changeset_id: str | None = _CHANGESET_ID,
|
|
validation_summary: dict[str, Any] | None = None,
|
|
is_terminal: bool = False,
|
|
error_details: dict[str, str] | None = None,
|
|
) -> MagicMock:
|
|
"""Create a mock Plan with sensible defaults."""
|
|
plan = MagicMock()
|
|
plan.identity.plan_id = plan_id
|
|
plan.phase = _StubPhase(phase)
|
|
plan.processing_state = _StubState(state)
|
|
plan.changeset_id = changeset_id
|
|
plan.validation_summary = validation_summary
|
|
plan.is_terminal = is_terminal
|
|
plan.error_details = error_details
|
|
plan.timestamps = PlanTimestamps()
|
|
plan.sandbox_refs = ["sandbox-ref-boost"]
|
|
return plan
|
|
|
|
|
|
def _make_lifecycle_mock(plan: MagicMock | None = None) -> MagicMock:
|
|
"""Create a mock PlanLifecycleService."""
|
|
lifecycle = MagicMock()
|
|
if plan is not None:
|
|
lifecycle.get_plan.return_value = plan
|
|
lifecycle._commit_plan = MagicMock()
|
|
lifecycle.complete_apply = MagicMock()
|
|
lifecycle.constrain_apply = MagicMock()
|
|
lifecycle.fail_apply = MagicMock()
|
|
return lifecycle
|
|
|
|
|
|
def _make_changeset_with_entry(
|
|
plan_id: str = _PLAN_ID,
|
|
changeset_id: str = _CHANGESET_ID,
|
|
) -> SpecChangeSet:
|
|
"""Create a SpecChangeSet with a single modify entry."""
|
|
cs = SpecChangeSet(changeset_id=changeset_id, plan_id=plan_id)
|
|
entry = ChangeEntry(
|
|
plan_id=plan_id,
|
|
resource_id="RES001",
|
|
tool_name="builtin/test-tool",
|
|
operation=ChangeOperation.MODIFY,
|
|
path="src/app.py",
|
|
before_hash="abcdef0123456789",
|
|
after_hash="123456abcdefghij",
|
|
)
|
|
cs.add_change(entry)
|
|
return cs
|
|
|
|
|
|
# ======================================================================
|
|
# _ApplyPhaseProxy scenarios
|
|
# ======================================================================
|
|
|
|
|
|
@given('pacb an ApplyPhaseProxy with plan_id "{plan_id}"')
|
|
def step_create_proxy(context: Context, plan_id: str) -> None:
|
|
"""Create an _ApplyPhaseProxy instance."""
|
|
context.pacb_proxy = _ApplyPhaseProxy(plan_id)
|
|
|
|
|
|
@then('pacb the proxy sandbox_id should be "{expected}"')
|
|
def step_proxy_sandbox_id(context: Context, expected: str) -> None:
|
|
"""Assert the proxy sandbox_id matches."""
|
|
actual = context.pacb_proxy.sandbox_id
|
|
assert actual == expected, f"Expected sandbox_id '{expected}', got '{actual}'"
|
|
|
|
|
|
@then("pacb the proxy context should be None")
|
|
def step_proxy_context_none(context: Context) -> None:
|
|
"""Assert the proxy context returns None."""
|
|
assert context.pacb_proxy.context is None, (
|
|
f"Expected context to be None, got {context.pacb_proxy.context}"
|
|
)
|
|
|
|
|
|
# ======================================================================
|
|
# _try_checkpoint scenarios
|
|
# ======================================================================
|
|
|
|
|
|
@given("pacb a service with a configured checkpoint manager")
|
|
def step_service_with_checkpoint_manager(context: Context) -> None:
|
|
"""Create a service with a mock checkpoint manager."""
|
|
lifecycle = _make_lifecycle_mock()
|
|
checkpoint_mgr = MagicMock()
|
|
checkpoint_mgr.create_checkpoint = MagicMock()
|
|
context.pacb_checkpoint_mgr = checkpoint_mgr
|
|
context.pacb_service = PlanApplyService(
|
|
lifecycle_service=lifecycle,
|
|
changeset_store=None,
|
|
checkpoint_manager=checkpoint_mgr,
|
|
)
|
|
context.pacb_error = None
|
|
|
|
|
|
@when('pacb I call _try_checkpoint with plan_id "{plan_id}" and phase "{phase}"')
|
|
def step_call_try_checkpoint(context: Context, plan_id: str, phase: str) -> None:
|
|
"""Call _try_checkpoint on the service."""
|
|
context.pacb_error = None
|
|
try:
|
|
context.pacb_service._try_checkpoint(plan_id, phase)
|
|
except Exception as exc:
|
|
context.pacb_error = exc
|
|
|
|
|
|
@when(
|
|
'pacb I call _try_checkpoint with plan_id "{plan_id}" and phase "{phase}" and metadata'
|
|
)
|
|
def step_call_try_checkpoint_with_metadata(
|
|
context: Context, plan_id: str, phase: str
|
|
) -> None:
|
|
"""Call _try_checkpoint on the service with metadata."""
|
|
context.pacb_error = None
|
|
try:
|
|
context.pacb_service._try_checkpoint(
|
|
plan_id, phase, metadata={"key1": "val1", "key2": "val2"}
|
|
)
|
|
except Exception as exc:
|
|
context.pacb_error = exc
|
|
|
|
|
|
@then("pacb the checkpoint manager create_checkpoint should have been called")
|
|
def step_checkpoint_create_called(context: Context) -> None:
|
|
"""Assert create_checkpoint was called on the checkpoint manager."""
|
|
context.pacb_checkpoint_mgr.create_checkpoint.assert_called()
|
|
|
|
|
|
@then('pacb the checkpoint sandbox should have sandbox_id starting with "apply-"')
|
|
def step_checkpoint_sandbox_has_prefix(context: Context) -> None:
|
|
"""Assert the sandbox arg passed to create_checkpoint has correct sandbox_id."""
|
|
call_args = context.pacb_checkpoint_mgr.create_checkpoint.call_args
|
|
sandbox_arg = call_args.kwargs.get("sandbox") or call_args[0][0]
|
|
assert sandbox_arg.sandbox_id.startswith("apply-"), (
|
|
f"Expected sandbox_id starting with 'apply-', got '{sandbox_arg.sandbox_id}'"
|
|
)
|
|
|
|
|
|
@then("pacb the checkpoint manager create_checkpoint should have received metadata")
|
|
def step_checkpoint_received_metadata(context: Context) -> None:
|
|
"""Assert create_checkpoint was called with metadata dict."""
|
|
call_args = context.pacb_checkpoint_mgr.create_checkpoint.call_args
|
|
metadata_arg = call_args.kwargs.get("metadata") or call_args[1].get(
|
|
"metadata", call_args[0][3] if len(call_args[0]) > 3 else {}
|
|
)
|
|
assert isinstance(metadata_arg, dict), (
|
|
f"Expected metadata dict, got {type(metadata_arg)}"
|
|
)
|
|
assert len(metadata_arg) > 0, "Expected non-empty metadata"
|
|
|
|
|
|
# -- _try_checkpoint exception path ----------------------------------------
|
|
|
|
|
|
@given("pacb a service with a checkpoint manager that raises on create")
|
|
def step_service_checkpoint_raises_on_create(context: Context) -> None:
|
|
"""Create a service with a checkpoint manager that raises on create."""
|
|
lifecycle = _make_lifecycle_mock()
|
|
checkpoint_mgr = MagicMock()
|
|
checkpoint_mgr.create_checkpoint.side_effect = RuntimeError(
|
|
"simulated checkpoint failure"
|
|
)
|
|
context.pacb_checkpoint_mgr = checkpoint_mgr
|
|
context.pacb_service = PlanApplyService(
|
|
lifecycle_service=lifecycle,
|
|
changeset_store=None,
|
|
checkpoint_manager=checkpoint_mgr,
|
|
)
|
|
context.pacb_error = None
|
|
|
|
|
|
@then("pacb no exception should have been raised")
|
|
def step_no_exception_raised(context: Context) -> None:
|
|
"""Assert that no exception was raised."""
|
|
assert context.pacb_error is None, (
|
|
f"Expected no exception, but got {type(context.pacb_error).__name__}: "
|
|
f"{context.pacb_error}"
|
|
)
|
|
|
|
|
|
# ======================================================================
|
|
# _try_rollback scenarios
|
|
# ======================================================================
|
|
|
|
|
|
@given("pacb a service with a checkpoint manager that has existing checkpoints")
|
|
def step_service_checkpoint_has_checkpoints(context: Context) -> None:
|
|
"""Create a service with a checkpoint manager that returns checkpoints."""
|
|
lifecycle = _make_lifecycle_mock()
|
|
checkpoint_mgr = MagicMock()
|
|
mock_checkpoint = MagicMock()
|
|
mock_checkpoint.checkpoint_id = "01CKPT00000000000001"
|
|
checkpoint_mgr.list_checkpoints.return_value = [mock_checkpoint]
|
|
checkpoint_mgr.rollback_to = MagicMock(return_value=True)
|
|
context.pacb_checkpoint_mgr = checkpoint_mgr
|
|
context.pacb_mock_checkpoint = mock_checkpoint
|
|
context.pacb_service = PlanApplyService(
|
|
lifecycle_service=lifecycle,
|
|
changeset_store=None,
|
|
checkpoint_manager=checkpoint_mgr,
|
|
)
|
|
context.pacb_error = None
|
|
|
|
|
|
@given("pacb a service with a checkpoint manager that returns no checkpoints")
|
|
def step_service_checkpoint_empty_list(context: Context) -> None:
|
|
"""Create a service with a checkpoint manager that returns empty list."""
|
|
lifecycle = _make_lifecycle_mock()
|
|
checkpoint_mgr = MagicMock()
|
|
checkpoint_mgr.list_checkpoints.return_value = []
|
|
checkpoint_mgr.rollback_to = MagicMock()
|
|
context.pacb_checkpoint_mgr = checkpoint_mgr
|
|
context.pacb_service = PlanApplyService(
|
|
lifecycle_service=lifecycle,
|
|
changeset_store=None,
|
|
checkpoint_manager=checkpoint_mgr,
|
|
)
|
|
context.pacb_error = None
|
|
|
|
|
|
@given("pacb a service with a checkpoint manager that raises on list")
|
|
def step_service_checkpoint_raises_on_list(context: Context) -> None:
|
|
"""Create a service with a checkpoint manager that raises on list."""
|
|
lifecycle = _make_lifecycle_mock()
|
|
checkpoint_mgr = MagicMock()
|
|
checkpoint_mgr.list_checkpoints.side_effect = RuntimeError("simulated list failure")
|
|
context.pacb_checkpoint_mgr = checkpoint_mgr
|
|
context.pacb_service = PlanApplyService(
|
|
lifecycle_service=lifecycle,
|
|
changeset_store=None,
|
|
checkpoint_manager=checkpoint_mgr,
|
|
)
|
|
context.pacb_error = None
|
|
|
|
|
|
@when('pacb I call _try_rollback with plan_id "{plan_id}"')
|
|
def step_call_try_rollback(context: Context, plan_id: str) -> None:
|
|
"""Call _try_rollback on the service."""
|
|
context.pacb_error = None
|
|
try:
|
|
context.pacb_service._try_rollback(plan_id)
|
|
except Exception as exc:
|
|
context.pacb_error = exc
|
|
|
|
|
|
@then("pacb the checkpoint manager list_checkpoints should have been called")
|
|
def step_list_checkpoints_called(context: Context) -> None:
|
|
"""Assert list_checkpoints was called."""
|
|
context.pacb_checkpoint_mgr.list_checkpoints.assert_called()
|
|
|
|
|
|
@then(
|
|
"pacb the checkpoint manager rollback_to should have been called with the last checkpoint"
|
|
)
|
|
def step_rollback_called_with_last(context: Context) -> None:
|
|
"""Assert rollback_to was called with the last checkpoint."""
|
|
context.pacb_checkpoint_mgr.rollback_to.assert_called_once_with(
|
|
context.pacb_mock_checkpoint
|
|
)
|
|
|
|
|
|
@then("pacb the checkpoint manager rollback_to should not have been called")
|
|
def step_rollback_not_called(context: Context) -> None:
|
|
"""Assert rollback_to was NOT called."""
|
|
context.pacb_checkpoint_mgr.rollback_to.assert_not_called()
|
|
|
|
|
|
# ======================================================================
|
|
# Integration: apply_with_validation_gate + checkpoint manager
|
|
# ======================================================================
|
|
|
|
|
|
@given("pacb a service with checkpoint manager and a plan with passing validations")
|
|
def step_service_checkpoint_plan_passing(context: Context) -> None:
|
|
"""Create a service with checkpoint_manager and a plan that passes validation."""
|
|
plan = _make_mock_plan(
|
|
changeset_id=_CHANGESET_ID,
|
|
is_terminal=False,
|
|
validation_summary={
|
|
"total": 2,
|
|
"required_passed": 2,
|
|
"required_failed": 0,
|
|
},
|
|
)
|
|
cs = _make_changeset_with_entry()
|
|
store = MagicMock()
|
|
store.get.return_value = cs
|
|
|
|
lifecycle = _make_lifecycle_mock(plan)
|
|
|
|
checkpoint_mgr = MagicMock()
|
|
checkpoint_mgr.create_checkpoint = MagicMock()
|
|
context.pacb_checkpoint_mgr = checkpoint_mgr
|
|
|
|
context.pacb_service = PlanApplyService(
|
|
lifecycle_service=lifecycle,
|
|
changeset_store=store,
|
|
checkpoint_manager=checkpoint_mgr,
|
|
)
|
|
context.pacb_error = None
|
|
|
|
|
|
@given("pacb a service with checkpoint manager and complete_apply raises PlanError")
|
|
def step_service_checkpoint_complete_raises(context: Context) -> None:
|
|
"""Create a service where complete_apply raises, triggering rollback."""
|
|
plan = _make_mock_plan(
|
|
changeset_id=_CHANGESET_ID,
|
|
is_terminal=False,
|
|
validation_summary={
|
|
"total": 2,
|
|
"required_passed": 2,
|
|
"required_failed": 0,
|
|
},
|
|
)
|
|
cs = _make_changeset_with_entry()
|
|
store = MagicMock()
|
|
store.get.return_value = cs
|
|
|
|
lifecycle = _make_lifecycle_mock(plan)
|
|
lifecycle.complete_apply.side_effect = PlanError("Not in Apply phase")
|
|
|
|
checkpoint_mgr = MagicMock()
|
|
checkpoint_mgr.create_checkpoint = MagicMock()
|
|
checkpoint_mgr.list_checkpoints.return_value = []
|
|
context.pacb_checkpoint_mgr = checkpoint_mgr
|
|
|
|
context.pacb_service = PlanApplyService(
|
|
lifecycle_service=lifecycle,
|
|
changeset_store=store,
|
|
checkpoint_manager=checkpoint_mgr,
|
|
)
|
|
context.pacb_error = None
|
|
|
|
|
|
@when("pacb I call apply_with_validation_gate")
|
|
def step_call_apply_gate(context: Context) -> None:
|
|
"""Call apply_with_validation_gate on the service."""
|
|
context.pacb_error = None
|
|
try:
|
|
context.pacb_apply_result = context.pacb_service.apply_with_validation_gate(
|
|
plan_id=_PLAN_ID,
|
|
)
|
|
except Exception as exc:
|
|
context.pacb_error = exc
|
|
|
|
|
|
@then('pacb the apply result outcome should be "{outcome}"')
|
|
def step_apply_result_outcome(context: Context, outcome: str) -> None:
|
|
"""Assert the apply result outcome matches."""
|
|
expected = ApplyOutcome(outcome)
|
|
assert context.pacb_apply_result.outcome == expected, (
|
|
f"Expected outcome '{outcome}', got '{context.pacb_apply_result.outcome}'"
|
|
)
|