forked from HAL9000/cleveragents-core
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
356 lines
12 KiB
Python
356 lines
12 KiB
Python
"""Step definitions for plan_model_coverage_boost.feature.
|
|
|
|
Targets uncovered lines in src/cleveragents/domain/models/core/plan.py:
|
|
- Line 792: can_transition_to_next_phase returns True for ACTION phase
|
|
- Lines 867, 870: get_project_scope resolves by alias / name
|
|
- Lines 893, 895: can_revert_to returns False for terminal / max-reversion
|
|
- Lines 970-980: as_cli_dict multi-project scope alias, read_only, changeset
|
|
- Lines 985-991: as_cli_dict cross_project_dependencies
|
|
- Lines 995, 997: as_cli_dict last_completed_step / last_checkpoint_id
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from behave import given, then, when
|
|
from behave.runner import Context
|
|
|
|
from cleveragents.domain.models.core.multi_project import (
|
|
ChangeSetSummary,
|
|
CrossProjectDependency,
|
|
MultiProjectMetadata,
|
|
ProjectScope,
|
|
)
|
|
from cleveragents.domain.models.core.plan import (
|
|
NamespacedName,
|
|
Plan,
|
|
PlanIdentity,
|
|
PlanPhase,
|
|
ProcessingState,
|
|
)
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Valid ULIDs for test fixtures (Crockford base32, 26 chars)
|
|
# ---------------------------------------------------------------------------
|
|
_ULID_A = "01HGZ6FE0AQDYTR4BXVQZ6EA01"
|
|
_ULID_CHECKPOINT = "01HGZ6FE0AQDYTR4BXVQZ6EB01"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _make_plan(
|
|
phase: PlanPhase = PlanPhase.STRATEGIZE,
|
|
processing_state: ProcessingState = ProcessingState.QUEUED,
|
|
multi_project_metadata: MultiProjectMetadata | None = None,
|
|
last_completed_step: int = -1,
|
|
last_checkpoint_id: str | None = None,
|
|
reversion_count: int = 0,
|
|
) -> Plan:
|
|
"""Build a minimal Plan for coverage-boost testing."""
|
|
return Plan(
|
|
identity=PlanIdentity(plan_id=_ULID_A),
|
|
namespaced_name=NamespacedName(namespace="local", name="cov-boost"),
|
|
action_name="local/cov-boost-action",
|
|
description="Plan for coverage boost testing",
|
|
phase=phase,
|
|
processing_state=processing_state,
|
|
multi_project_metadata=multi_project_metadata,
|
|
last_completed_step=last_completed_step,
|
|
last_checkpoint_id=last_checkpoint_id,
|
|
reversion_count=reversion_count,
|
|
)
|
|
|
|
|
|
# ===================================================================
|
|
# can_transition_to_next_phase — ACTION phase (line 792)
|
|
# ===================================================================
|
|
|
|
|
|
@given("a coverage-boost plan in the ACTION phase")
|
|
def step_create_action_plan(context: Context) -> None:
|
|
context.plan = _make_plan(
|
|
phase=PlanPhase.ACTION,
|
|
processing_state=ProcessingState.QUEUED,
|
|
)
|
|
|
|
|
|
@when("I check whether the coverage-boost plan can transition to the next phase")
|
|
def step_check_transition(context: Context) -> None:
|
|
context.transition_result = context.plan.can_transition_to_next_phase
|
|
|
|
|
|
@then("the coverage-boost transition check should return True")
|
|
def step_verify_transition_true(context: Context) -> None:
|
|
assert context.transition_result is True, (
|
|
f"Expected True, got {context.transition_result}"
|
|
)
|
|
|
|
|
|
# ===================================================================
|
|
# get_project_scope — alias / name lookup (lines 867, 870)
|
|
# ===================================================================
|
|
|
|
|
|
def _multi_project_with_aliased_scopes() -> MultiProjectMetadata:
|
|
"""Build MultiProjectMetadata with two scopes, one aliased."""
|
|
return MultiProjectMetadata(
|
|
project_scopes=[
|
|
ProjectScope(
|
|
project_name="local/api-service",
|
|
alias="api",
|
|
read_only=False,
|
|
),
|
|
ProjectScope(
|
|
project_name="local/web-app",
|
|
alias=None,
|
|
read_only=True,
|
|
),
|
|
],
|
|
)
|
|
|
|
|
|
@given("a coverage-boost plan with multi-project metadata containing aliased scopes")
|
|
def step_create_plan_with_aliased_scopes(context: Context) -> None:
|
|
context.plan = _make_plan(
|
|
multi_project_metadata=_multi_project_with_aliased_scopes(),
|
|
)
|
|
|
|
|
|
@when('I look up the coverage-boost project scope by alias "{alias}"')
|
|
def step_lookup_scope_by_alias(context: Context, alias: str) -> None:
|
|
context.resolved_scope = context.plan.get_project_scope(alias)
|
|
|
|
|
|
@when('I look up the coverage-boost project scope by name "{name}"')
|
|
def step_lookup_scope_by_name(context: Context, name: str) -> None:
|
|
context.resolved_scope = context.plan.get_project_scope(name)
|
|
|
|
|
|
@then('the coverage-boost resolved scope project name should be "{expected}"')
|
|
def step_verify_scope_name(context: Context, expected: str) -> None:
|
|
assert context.resolved_scope is not None, "Expected a scope, got None"
|
|
assert context.resolved_scope.project_name == expected, (
|
|
f"Expected project_name={expected!r}, "
|
|
f"got {context.resolved_scope.project_name!r}"
|
|
)
|
|
|
|
|
|
@then("the coverage-boost resolved scope should be None")
|
|
def step_verify_scope_none(context: Context) -> None:
|
|
assert context.resolved_scope is None, (
|
|
f"Expected None, got {context.resolved_scope}"
|
|
)
|
|
|
|
|
|
@given("a coverage-boost plan without multi-project metadata")
|
|
def step_create_plan_no_mp(context: Context) -> None:
|
|
context.plan = _make_plan(multi_project_metadata=None)
|
|
|
|
|
|
# ===================================================================
|
|
# can_revert_to — terminal states / max reversions (lines 893, 895)
|
|
# ===================================================================
|
|
|
|
|
|
@given("a coverage-boost plan in APPLY phase with APPLIED state")
|
|
def step_create_applied_plan(context: Context) -> None:
|
|
context.plan = _make_plan(
|
|
phase=PlanPhase.APPLY,
|
|
processing_state=ProcessingState.APPLIED,
|
|
)
|
|
|
|
|
|
@given("a coverage-boost plan in APPLY phase with CANCELLED state")
|
|
def step_create_cancelled_plan(context: Context) -> None:
|
|
context.plan = _make_plan(
|
|
phase=PlanPhase.APPLY,
|
|
processing_state=ProcessingState.CANCELLED,
|
|
)
|
|
|
|
|
|
@given("a coverage-boost plan that has used all allowed reversions")
|
|
def step_create_max_reversion_plan(context: Context) -> None:
|
|
context.plan = _make_plan(
|
|
phase=PlanPhase.EXECUTE,
|
|
processing_state=ProcessingState.ERRORED,
|
|
reversion_count=Plan.MAX_REVERSIONS,
|
|
)
|
|
|
|
|
|
@given("a coverage-boost plan in EXECUTE phase with ERRORED state and zero reversions")
|
|
def step_create_execute_errored_plan(context: Context) -> None:
|
|
context.plan = _make_plan(
|
|
phase=PlanPhase.EXECUTE,
|
|
processing_state=ProcessingState.ERRORED,
|
|
reversion_count=0,
|
|
)
|
|
|
|
|
|
@when("I check whether the coverage-boost plan can revert to STRATEGIZE")
|
|
def step_check_revert(context: Context) -> None:
|
|
context.revert_result = context.plan.can_revert_to(PlanPhase.STRATEGIZE)
|
|
|
|
|
|
@then("the coverage-boost reversion check should return False")
|
|
def step_verify_revert_false(context: Context) -> None:
|
|
assert context.revert_result is False, (
|
|
f"Expected False, got {context.revert_result}"
|
|
)
|
|
|
|
|
|
@then("the coverage-boost reversion check should return True")
|
|
def step_verify_revert_true(context: Context) -> None:
|
|
assert context.revert_result is True, f"Expected True, got {context.revert_result}"
|
|
|
|
|
|
# ===================================================================
|
|
# as_cli_dict — multi-project scope alias, read_only (lines 970, 972)
|
|
# ===================================================================
|
|
|
|
|
|
@given(
|
|
"a coverage-boost plan with multi-project scopes having alias and read-only flags"
|
|
)
|
|
def step_create_plan_mp_alias_readonly(context: Context) -> None:
|
|
mp = MultiProjectMetadata(
|
|
project_scopes=[
|
|
ProjectScope(
|
|
project_name="local/api-service",
|
|
alias="api",
|
|
read_only=True,
|
|
),
|
|
],
|
|
)
|
|
context.plan = _make_plan(multi_project_metadata=mp)
|
|
|
|
|
|
@when("I generate the coverage-boost CLI dict")
|
|
def step_generate_cli_dict(context: Context) -> None:
|
|
context.cli_dict = context.plan.as_cli_dict()
|
|
|
|
|
|
@then('the coverage-boost CLI dict multi_project scope should contain alias "{alias}"')
|
|
def step_verify_mp_alias(context: Context, alias: str) -> None:
|
|
mp = context.cli_dict.get("multi_project")
|
|
assert mp is not None, "multi_project key missing from CLI dict"
|
|
scopes = mp.get("project_scopes", [])
|
|
assert len(scopes) > 0, "No project_scopes in multi_project"
|
|
assert scopes[0].get("alias") == alias, (
|
|
f"Expected alias={alias!r}, got {scopes[0].get('alias')!r}"
|
|
)
|
|
|
|
|
|
@then("the coverage-boost CLI dict multi_project scope should contain read_only True")
|
|
def step_verify_mp_read_only(context: Context) -> None:
|
|
mp = context.cli_dict["multi_project"]
|
|
scopes = mp["project_scopes"]
|
|
assert scopes[0].get("read_only") is True, (
|
|
f"Expected read_only=True, got {scopes[0].get('read_only')}"
|
|
)
|
|
|
|
|
|
# ===================================================================
|
|
# as_cli_dict — changeset summary (lines 974-980)
|
|
# ===================================================================
|
|
|
|
|
|
@given("a coverage-boost plan with a multi-project scope that has a changeset summary")
|
|
def step_create_plan_mp_changeset(context: Context) -> None:
|
|
mp = MultiProjectMetadata(
|
|
project_scopes=[
|
|
ProjectScope(
|
|
project_name="local/api-service",
|
|
changeset_summary=ChangeSetSummary(
|
|
project_name="local/api-service",
|
|
files_changed=3,
|
|
files_added=1,
|
|
files_deleted=0,
|
|
total_lines_changed=42,
|
|
validation_passed=True,
|
|
),
|
|
),
|
|
],
|
|
)
|
|
context.plan = _make_plan(multi_project_metadata=mp)
|
|
|
|
|
|
@then(
|
|
"the coverage-boost CLI dict multi_project scope should contain changeset details"
|
|
)
|
|
def step_verify_mp_changeset(context: Context) -> None:
|
|
mp = context.cli_dict["multi_project"]
|
|
scope = mp["project_scopes"][0]
|
|
cs = scope.get("changeset")
|
|
assert cs is not None, "changeset key missing from scope"
|
|
assert cs["files_changed"] == 3
|
|
assert cs["files_added"] == 1
|
|
assert cs["files_deleted"] == 0
|
|
assert cs["total_lines_changed"] == 42
|
|
assert cs["validation_passed"] is True
|
|
|
|
|
|
# ===================================================================
|
|
# as_cli_dict — cross_project_dependencies (lines 985-991)
|
|
# ===================================================================
|
|
|
|
|
|
@given("a coverage-boost plan with cross-project dependencies in metadata")
|
|
def step_create_plan_mp_deps(context: Context) -> None:
|
|
mp = MultiProjectMetadata(
|
|
project_scopes=[
|
|
ProjectScope(project_name="local/api-service"),
|
|
ProjectScope(project_name="local/web-app"),
|
|
],
|
|
cross_project_dependencies=[
|
|
CrossProjectDependency(
|
|
source_project="local/web-app",
|
|
target_project="local/api-service",
|
|
dependency_type="api-consumer",
|
|
),
|
|
],
|
|
)
|
|
context.plan = _make_plan(multi_project_metadata=mp)
|
|
|
|
|
|
@then(
|
|
"the coverage-boost CLI dict multi_project should contain cross_project_dependencies"
|
|
)
|
|
def step_verify_mp_deps(context: Context) -> None:
|
|
mp = context.cli_dict["multi_project"]
|
|
deps = mp.get("cross_project_dependencies")
|
|
assert deps is not None, "cross_project_dependencies key missing"
|
|
assert len(deps) == 1
|
|
assert deps[0]["source"] == "local/web-app"
|
|
assert deps[0]["target"] == "local/api-service"
|
|
assert deps[0]["type"] == "api-consumer"
|
|
|
|
|
|
# ===================================================================
|
|
# as_cli_dict — last_completed_step / last_checkpoint_id (lines 995, 997)
|
|
# ===================================================================
|
|
|
|
|
|
@given("a coverage-boost plan with last_completed_step 2 and a checkpoint ID")
|
|
def step_create_plan_resume_metadata(context: Context) -> None:
|
|
context.plan = _make_plan(
|
|
last_completed_step=2,
|
|
last_checkpoint_id=_ULID_CHECKPOINT,
|
|
)
|
|
|
|
|
|
@then("the coverage-boost CLI dict should contain last_completed_step 2")
|
|
def step_verify_last_step(context: Context) -> None:
|
|
assert context.cli_dict.get("last_completed_step") == 2, (
|
|
f"Expected last_completed_step=2, got {context.cli_dict.get('last_completed_step')}"
|
|
)
|
|
|
|
|
|
@then("the coverage-boost CLI dict should contain last_checkpoint_id")
|
|
def step_verify_checkpoint(context: Context) -> None:
|
|
assert context.cli_dict.get("last_checkpoint_id") == _ULID_CHECKPOINT, (
|
|
f"Expected last_checkpoint_id={_ULID_CHECKPOINT}, "
|
|
f"got {context.cli_dict.get('last_checkpoint_id')}"
|
|
)
|