forked from HAL9000/cleveragents-core
freemo
a0df5a4cd0
Implements the spec-required JSON/YAML output envelope for all CLI commands
that use format_output(). The envelope structure is:
{
"command": "<command that was run>",
"status": "ok" | "warn" | "error",
"exit_code": 0,
"data": { ... command-specific payload ... },
"timing": { "duration_ms": 123 },
"messages": [{ "level": "ok", "text": "..." }]
}
Changes:
- Add _build_envelope() helper to construct the spec-required envelope
- Add optional command, status, exit_code, messages parameters to format_output()
- Wrap json/yaml output in the envelope; plain/table/rich/color unchanged
- Add timing measurement (duration_ms) to all json/yaml outputs
- Add new BDD feature file (cli_json_envelope.feature) with 14 scenarios
testing envelope field presence, values, and data payload
- Update 14 existing step files to unwrap the envelope when checking
specific data keys (backward-compatible via _unwrap_envelope() helper)
Closes #3431
831 lines
28 KiB
Python
831 lines
28 KiB
Python
"""Step definitions for plan_explain_cli_coverage.feature.
|
|
|
|
Exercises the CLI-level code paths for plan explain, plan tree,
|
|
plan correct, plan resume, plan revert, and the read-only plan guards,
|
|
using the Typer CliRunner with mocked services.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from typing import Any
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
from behave import given, then, when
|
|
from behave.runner import Context
|
|
from typer.testing import CliRunner
|
|
from ulid import ULID
|
|
|
|
from cleveragents.application.services.plan_lifecycle_service import (
|
|
InvalidPhaseTransitionError,
|
|
)
|
|
from cleveragents.cli.commands.plan import app as plan_app
|
|
from cleveragents.core.exceptions import (
|
|
CleverAgentsError,
|
|
PlanError,
|
|
ValidationError,
|
|
)
|
|
from cleveragents.domain.models.core.decision import (
|
|
ContextSnapshot,
|
|
Decision,
|
|
DecisionType,
|
|
ResourceRef,
|
|
)
|
|
|
|
runner = CliRunner()
|
|
|
|
_ENVELOPE_KEYS = {"command", "status", "exit_code", "data", "timing", "messages"}
|
|
|
|
|
|
def _unwrap_envelope(parsed: Any) -> Any:
|
|
"""Return the ``data`` field if *parsed* is a spec envelope, else *parsed* as-is."""
|
|
if isinstance(parsed, dict) and _ENVELOPE_KEYS.issubset(parsed.keys()):
|
|
return parsed["data"]
|
|
return parsed
|
|
|
|
|
|
_PATCH_CONTAINER = "cleveragents.application.container.get_container"
|
|
_PATCH_LIFECYCLE = "cleveragents.cli.commands.plan._get_lifecycle_service"
|
|
|
|
_PATCH_RESUME_SVC_MOD = (
|
|
"cleveragents.application.services.plan_resume_service.PlanResumeService"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _make_decision(
|
|
decision_id: str | None = None,
|
|
plan_id: str | None = None,
|
|
sequence: int = 0,
|
|
parent_id: str | None = None,
|
|
dtype: DecisionType = DecisionType.PROMPT_DEFINITION,
|
|
question: str = "What should we build?",
|
|
chosen: str = "A REST API",
|
|
superseded_by: str | None = None,
|
|
is_correction: bool = False,
|
|
corrects_decision_id: str | None = None,
|
|
correction_reason: str | None = None,
|
|
confidence_score: float | None = 0.85,
|
|
rationale: str = "",
|
|
actor_reasoning: str | None = None,
|
|
alternatives: list[str] | None = None,
|
|
context_snapshot: ContextSnapshot | None = None,
|
|
) -> Decision:
|
|
did = decision_id or str(ULID())
|
|
pid = plan_id or str(ULID())
|
|
kwargs: dict = {
|
|
"decision_id": did,
|
|
"plan_id": pid,
|
|
"sequence_number": sequence,
|
|
"decision_type": dtype,
|
|
"question": question,
|
|
"chosen_option": chosen,
|
|
"confidence_score": confidence_score,
|
|
"rationale": rationale,
|
|
"superseded_by": superseded_by,
|
|
"is_correction": is_correction,
|
|
}
|
|
if parent_id is not None:
|
|
kwargs["parent_decision_id"] = parent_id
|
|
if corrects_decision_id is not None:
|
|
kwargs["corrects_decision_id"] = corrects_decision_id
|
|
if correction_reason is not None:
|
|
kwargs["correction_reason"] = correction_reason
|
|
if actor_reasoning is not None:
|
|
kwargs["actor_reasoning"] = actor_reasoning
|
|
if alternatives is not None:
|
|
kwargs["alternatives_considered"] = alternatives
|
|
if context_snapshot is not None:
|
|
kwargs["context_snapshot"] = context_snapshot
|
|
return Decision(**kwargs)
|
|
|
|
|
|
def _mock_container_with_decision_svc(svc_mock: MagicMock) -> MagicMock:
|
|
container = MagicMock()
|
|
container.decision_service.return_value = svc_mock
|
|
return container
|
|
|
|
|
|
def _make_tree_decisions() -> list[Decision]:
|
|
root_id = str(ULID())
|
|
child_id = str(ULID())
|
|
return [
|
|
_make_decision(decision_id=root_id, sequence=0),
|
|
_make_decision(
|
|
decision_id=child_id,
|
|
parent_id=root_id,
|
|
sequence=1,
|
|
dtype=DecisionType.STRATEGY_CHOICE,
|
|
question="Which framework?",
|
|
chosen="FastAPI",
|
|
),
|
|
]
|
|
|
|
|
|
def _make_deep_decisions() -> list[Decision]:
|
|
root_id = str(ULID())
|
|
child_id = str(ULID())
|
|
grandchild_id = str(ULID())
|
|
return [
|
|
_make_decision(decision_id=root_id, sequence=0),
|
|
_make_decision(
|
|
decision_id=child_id,
|
|
parent_id=root_id,
|
|
sequence=1,
|
|
dtype=DecisionType.STRATEGY_CHOICE,
|
|
question="Child question?",
|
|
chosen="Child answer",
|
|
),
|
|
_make_decision(
|
|
decision_id=grandchild_id,
|
|
parent_id=child_id,
|
|
sequence=2,
|
|
dtype=DecisionType.IMPLEMENTATION_CHOICE,
|
|
question="Grandchild question?",
|
|
chosen="Grandchild answer",
|
|
),
|
|
]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# GIVEN steps - explain
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("pec a mock DecisionService returning a valid decision")
|
|
def step_pec_mock_decision_svc(context: Context) -> None:
|
|
context.pec_decision_id = str(ULID())
|
|
context.pec_decision = _make_decision(decision_id=context.pec_decision_id)
|
|
svc = MagicMock()
|
|
svc.get_decision.return_value = context.pec_decision
|
|
context.pec_container = _mock_container_with_decision_svc(svc)
|
|
|
|
|
|
@given("pec a mock DecisionService returning None for get_decision")
|
|
def step_pec_mock_decision_none(context: Context) -> None:
|
|
context.pec_decision_id = str(ULID())
|
|
svc = MagicMock()
|
|
svc.get_decision.return_value = None
|
|
svc.list_decisions.return_value = []
|
|
context.pec_container = _mock_container_with_decision_svc(svc)
|
|
|
|
|
|
@given("pec a mock DecisionService returning a decision with context snapshot")
|
|
def step_pec_mock_decision_ctx(context: Context) -> None:
|
|
context.pec_decision_id = str(ULID())
|
|
snap = ContextSnapshot(
|
|
hot_context_hash="sha256:abc123",
|
|
hot_context_ref="store://ctx/1",
|
|
relevant_resources=[
|
|
ResourceRef(resource_id=str(ULID()), path="src/main.py"),
|
|
],
|
|
actor_state_ref="checkpoint://state/1",
|
|
)
|
|
context.pec_decision = _make_decision(
|
|
decision_id=context.pec_decision_id,
|
|
context_snapshot=snap,
|
|
)
|
|
svc = MagicMock()
|
|
svc.get_decision.return_value = context.pec_decision
|
|
context.pec_container = _mock_container_with_decision_svc(svc)
|
|
|
|
|
|
@given("pec a mock DecisionService returning a decision with reasoning")
|
|
def step_pec_mock_decision_reasoning(context: Context) -> None:
|
|
context.pec_decision_id = str(ULID())
|
|
context.pec_decision = _make_decision(
|
|
decision_id=context.pec_decision_id,
|
|
rationale="Chose REST API for simplicity",
|
|
actor_reasoning="The LLM considered approaches...",
|
|
)
|
|
svc = MagicMock()
|
|
svc.get_decision.return_value = context.pec_decision
|
|
context.pec_container = _mock_container_with_decision_svc(svc)
|
|
|
|
|
|
@given("pec a mock DecisionService returning a decision with alternatives")
|
|
def step_pec_mock_decision_alts(context: Context) -> None:
|
|
context.pec_decision_id = str(ULID())
|
|
context.pec_decision = _make_decision(
|
|
decision_id=context.pec_decision_id,
|
|
alternatives=["GraphQL API", "gRPC service"],
|
|
)
|
|
svc = MagicMock()
|
|
svc.get_decision.return_value = context.pec_decision
|
|
context.pec_container = _mock_container_with_decision_svc(svc)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# GIVEN steps - tree
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("pec a mock DecisionService returning a list of decisions")
|
|
def step_pec_mock_tree_decisions(context: Context) -> None:
|
|
context.pec_plan_id = str(ULID())
|
|
decisions = _make_tree_decisions()
|
|
svc = MagicMock()
|
|
svc.list_decisions.return_value = decisions
|
|
context.pec_container = _mock_container_with_decision_svc(svc)
|
|
|
|
|
|
@given("pec a mock DecisionService returning a deep decision list")
|
|
def step_pec_mock_deep_decisions(context: Context) -> None:
|
|
context.pec_plan_id = str(ULID())
|
|
decisions = _make_deep_decisions()
|
|
svc = MagicMock()
|
|
svc.list_decisions.return_value = decisions
|
|
context.pec_container = _mock_container_with_decision_svc(svc)
|
|
|
|
|
|
@given("pec a mock DecisionService returning decisions with superseded")
|
|
def step_pec_mock_superseded_decisions(context: Context) -> None:
|
|
context.pec_plan_id = str(ULID())
|
|
root_id = str(ULID())
|
|
old_id = str(ULID())
|
|
new_id = str(ULID())
|
|
decisions = [
|
|
_make_decision(decision_id=root_id, sequence=0),
|
|
_make_decision(
|
|
decision_id=old_id,
|
|
parent_id=root_id,
|
|
sequence=1,
|
|
dtype=DecisionType.STRATEGY_CHOICE,
|
|
question="Which framework?",
|
|
chosen="Flask",
|
|
superseded_by=new_id,
|
|
),
|
|
_make_decision(
|
|
decision_id=new_id,
|
|
parent_id=root_id,
|
|
sequence=2,
|
|
dtype=DecisionType.STRATEGY_CHOICE,
|
|
question="Which framework?",
|
|
chosen="FastAPI",
|
|
is_correction=True,
|
|
corrects_decision_id=old_id,
|
|
correction_reason="Better performance",
|
|
),
|
|
]
|
|
svc = MagicMock()
|
|
svc.list_decisions.return_value = decisions
|
|
context.pec_container = _mock_container_with_decision_svc(svc)
|
|
|
|
|
|
@given("pec a mock DecisionService returning an empty list")
|
|
def step_pec_mock_empty_decisions(context: Context) -> None:
|
|
context.pec_plan_id = str(ULID())
|
|
svc = MagicMock()
|
|
svc.list_decisions.return_value = []
|
|
context.pec_container = _mock_container_with_decision_svc(svc)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# GIVEN steps - orphan edge case
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("pec a decision list where a child references a missing parent")
|
|
def step_pec_orphan_decisions(context: Context) -> None:
|
|
"""Build three decisions where the grandchild is superseded.
|
|
|
|
With ``show_superseded=False`` the grandchild is excluded from
|
|
``by_id`` but its entry in ``children_map`` (built from ALL
|
|
decisions) still references its ``decision_id`` under its parent.
|
|
This forces the BFS orphan guard (``child_id not in by_id``) to
|
|
fire and skip the missing node.
|
|
"""
|
|
root_id = str(ULID())
|
|
child_id = str(ULID())
|
|
grandchild_id = str(ULID())
|
|
context.pec_orphan_root_id = root_id
|
|
context.pec_orphan_decisions = [
|
|
_make_decision(decision_id=root_id, sequence=0),
|
|
_make_decision(
|
|
decision_id=child_id,
|
|
parent_id=root_id,
|
|
sequence=1,
|
|
dtype=DecisionType.STRATEGY_CHOICE,
|
|
question="Child question?",
|
|
chosen="Child answer",
|
|
),
|
|
_make_decision(
|
|
decision_id=grandchild_id,
|
|
parent_id=child_id,
|
|
sequence=2,
|
|
dtype=DecisionType.IMPLEMENTATION_CHOICE,
|
|
question="Grandchild question?",
|
|
chosen="Grandchild answer",
|
|
superseded_by=str(ULID()),
|
|
),
|
|
]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# GIVEN steps - resolve active plan
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("pec a lifecycle service with one active plan")
|
|
def step_pec_lifecycle_active(context: Context) -> None:
|
|
context.pec_expected_plan_id = str(ULID())
|
|
plan = MagicMock()
|
|
plan.is_terminal = False
|
|
plan.identity.plan_id = context.pec_expected_plan_id
|
|
svc = MagicMock()
|
|
svc.list_plans.return_value = [plan]
|
|
context.pec_lifecycle_svc = svc
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# GIVEN steps - revert error handlers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("pec a lifecycle service that raises InvalidPhaseTransitionError on revert")
|
|
def step_pec_revert_invalid_phase(context: Context) -> None:
|
|
from cleveragents.domain.models.core.plan import PlanPhase
|
|
|
|
context.pec_plan_id = str(ULID())
|
|
svc = MagicMock()
|
|
svc.revert_plan.side_effect = InvalidPhaseTransitionError(
|
|
from_phase=PlanPhase.APPLY,
|
|
to_phase=PlanPhase.STRATEGIZE,
|
|
message="Cannot revert from apply phase",
|
|
)
|
|
context.pec_lifecycle_svc = svc
|
|
|
|
|
|
@given("pec a lifecycle service that raises PlanError on revert")
|
|
def step_pec_revert_plan_error(context: Context) -> None:
|
|
context.pec_plan_id = str(ULID())
|
|
svc = MagicMock()
|
|
svc.revert_plan.side_effect = PlanError("Plan is terminal")
|
|
context.pec_lifecycle_svc = svc
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# GIVEN steps - correct
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("pec a mock CorrectionService with dry-run impact")
|
|
def step_pec_correct_dry_run(context: Context) -> None:
|
|
context.pec_decision_id = str(ULID())
|
|
context.pec_plan_id = str(ULID())
|
|
|
|
request = MagicMock()
|
|
request.correction_id = str(ULID())
|
|
request.mode.value = "revert"
|
|
request.target_decision_id = context.pec_decision_id
|
|
request.guidance = "Use FastAPI instead"
|
|
|
|
impact = MagicMock()
|
|
impact.affected_decisions = ["DEC-A", "DEC-B"]
|
|
impact.affected_files = ["src/api.py"]
|
|
impact.estimated_cost = "low"
|
|
impact.risk_level = "medium"
|
|
|
|
svc = MagicMock()
|
|
svc.request_correction.return_value = request
|
|
svc.analyze_impact.return_value = impact
|
|
context.pec_correction_svc = svc
|
|
|
|
|
|
@given("pec a mock CorrectionService with execute result")
|
|
def step_pec_correct_execute(context: Context) -> None:
|
|
context.pec_decision_id = str(ULID())
|
|
context.pec_plan_id = str(ULID())
|
|
|
|
request = MagicMock()
|
|
request.correction_id = str(ULID())
|
|
request.mode.value = "revert"
|
|
request.target_decision_id = context.pec_decision_id
|
|
request.guidance = "Use FastAPI instead"
|
|
|
|
result = MagicMock()
|
|
result.correction_id = request.correction_id
|
|
result.status.value = "applied"
|
|
result.new_decisions = []
|
|
result.reverted_decisions = []
|
|
|
|
svc = MagicMock()
|
|
svc.request_correction.return_value = request
|
|
svc.execute_correction.return_value = result
|
|
context.pec_correction_svc = svc
|
|
|
|
|
|
@given("pec a mock CorrectionService with reverted and new decisions")
|
|
def step_pec_correct_with_changes(context: Context) -> None:
|
|
context.pec_decision_id = str(ULID())
|
|
context.pec_plan_id = str(ULID())
|
|
|
|
request = MagicMock()
|
|
request.correction_id = str(ULID())
|
|
request.mode.value = "revert"
|
|
request.target_decision_id = context.pec_decision_id
|
|
request.guidance = "Use FastAPI instead"
|
|
|
|
result = MagicMock()
|
|
result.correction_id = request.correction_id
|
|
result.status.value = "applied"
|
|
result.new_decisions = ["DEC-NEW-1"]
|
|
result.reverted_decisions = ["DEC-OLD-1"]
|
|
|
|
svc = MagicMock()
|
|
svc.request_correction.return_value = request
|
|
svc.execute_correction.return_value = result
|
|
context.pec_correction_svc = svc
|
|
|
|
|
|
@given("pec a mock CorrectionService that raises ResourceNotFoundError")
|
|
def step_pec_correct_rnf(context: Context) -> None:
|
|
context.pec_decision_id = str(ULID())
|
|
context.pec_plan_id = str(ULID())
|
|
from cleveragents.core.exceptions import ResourceNotFoundError
|
|
|
|
svc = MagicMock()
|
|
svc.request_correction.side_effect = ResourceNotFoundError("Decision not found")
|
|
context.pec_correction_svc = svc
|
|
|
|
|
|
@given("pec a mock CorrectionService that raises ValidationError")
|
|
def step_pec_correct_validation(context: Context) -> None:
|
|
context.pec_decision_id = str(ULID())
|
|
context.pec_plan_id = str(ULID())
|
|
|
|
svc = MagicMock()
|
|
svc.request_correction.side_effect = ValidationError("Invalid correction mode")
|
|
context.pec_correction_svc = svc
|
|
|
|
|
|
@given("pec a mock CorrectionService that raises CleverAgentsError")
|
|
def step_pec_correct_ca_error(context: Context) -> None:
|
|
context.pec_decision_id = str(ULID())
|
|
context.pec_plan_id = str(ULID())
|
|
|
|
svc = MagicMock()
|
|
svc.request_correction.side_effect = CleverAgentsError("Service unavailable")
|
|
context.pec_correction_svc = svc
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# GIVEN steps - resume
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("pec a mock PlanResumeService returning a summary")
|
|
def step_pec_resume_svc(context: Context) -> None:
|
|
context.pec_plan_id = str(ULID())
|
|
summary = MagicMock()
|
|
summary.plan_id = context.pec_plan_id
|
|
summary.phase = "execute"
|
|
summary.processing_state = "in_progress"
|
|
summary.last_completed_step = 3
|
|
summary.next_step_index = 4
|
|
summary.total_steps = 10
|
|
summary.decision_id = str(ULID())
|
|
summary.last_checkpoint_id = str(ULID())
|
|
summary.sandbox_ref = "/tmp/sandbox/plan-001"
|
|
summary.as_cli_dict.return_value = {
|
|
"plan_id": summary.plan_id,
|
|
"phase": "execute",
|
|
"next_step": 4,
|
|
}
|
|
context.pec_resume_summary = summary
|
|
|
|
svc = MagicMock()
|
|
svc.resume_plan.return_value = summary
|
|
context.pec_resume_svc = svc
|
|
|
|
|
|
@given("pec a mock PlanResumeService that raises PlanError")
|
|
def step_pec_resume_error(context: Context) -> None:
|
|
context.pec_plan_id = str(ULID())
|
|
svc = MagicMock()
|
|
svc.resume_plan.side_effect = PlanError("Plan is terminal, cannot resume")
|
|
context.pec_resume_svc = svc
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# GIVEN steps - read-only plan
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("pec a lifecycle service returning a read-only plan")
|
|
def step_pec_readonly_plan(context: Context) -> None:
|
|
context.pec_plan_id = str(ULID())
|
|
plan = MagicMock()
|
|
plan.read_only = True
|
|
svc = MagicMock()
|
|
svc.get_plan.return_value = plan
|
|
context.pec_lifecycle_svc = svc
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# WHEN steps - explain
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when('pec I invoke "explain" with the decision id')
|
|
def step_pec_invoke_explain(context: Context) -> None:
|
|
with patch(_PATCH_CONTAINER, return_value=context.pec_container):
|
|
context.pec_result = runner.invoke(
|
|
plan_app, ["explain", context.pec_decision_id]
|
|
)
|
|
|
|
|
|
@when('pec I invoke "explain" with format "{fmt}"')
|
|
def step_pec_invoke_explain_fmt(context: Context, fmt: str) -> None:
|
|
with patch(_PATCH_CONTAINER, return_value=context.pec_container):
|
|
context.pec_result = runner.invoke(
|
|
plan_app, ["explain", context.pec_decision_id, "--format", fmt]
|
|
)
|
|
|
|
|
|
@when('pec I invoke "explain" with flags "{flags}"')
|
|
def step_pec_invoke_explain_flags(context: Context, flags: str) -> None:
|
|
with patch(_PATCH_CONTAINER, return_value=context.pec_container):
|
|
context.pec_result = runner.invoke(
|
|
plan_app,
|
|
["explain", context.pec_decision_id, *flags.split()],
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# WHEN steps - tree
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when('pec I invoke "tree" with a plan id')
|
|
def step_pec_invoke_tree(context: Context) -> None:
|
|
with patch(_PATCH_CONTAINER, return_value=context.pec_container):
|
|
context.pec_result = runner.invoke(plan_app, ["tree", context.pec_plan_id])
|
|
|
|
|
|
@when('pec I invoke "tree" with format "{fmt}"')
|
|
def step_pec_invoke_tree_fmt(context: Context, fmt: str) -> None:
|
|
with patch(_PATCH_CONTAINER, return_value=context.pec_container):
|
|
context.pec_result = runner.invoke(
|
|
plan_app, ["tree", context.pec_plan_id, "--format", fmt]
|
|
)
|
|
|
|
|
|
@when('pec I invoke "tree" with format "{fmt}" and depth {depth:d}')
|
|
def step_pec_invoke_tree_fmt_depth(context: Context, fmt: str, depth: int) -> None:
|
|
with patch(_PATCH_CONTAINER, return_value=context.pec_container):
|
|
context.pec_result = runner.invoke(
|
|
plan_app,
|
|
["tree", context.pec_plan_id, "--format", fmt, "--depth", str(depth)],
|
|
)
|
|
|
|
|
|
@when('pec I invoke "tree" with depth {depth:d}')
|
|
def step_pec_invoke_tree_depth(context: Context, depth: int) -> None:
|
|
with patch(_PATCH_CONTAINER, return_value=context.pec_container):
|
|
context.pec_result = runner.invoke(
|
|
plan_app, ["tree", context.pec_plan_id, "--depth", str(depth)]
|
|
)
|
|
|
|
|
|
@when('pec I invoke "tree" with flags "{flags}"')
|
|
def step_pec_invoke_tree_flags(context: Context, flags: str) -> None:
|
|
with patch(_PATCH_CONTAINER, return_value=context.pec_container):
|
|
context.pec_result = runner.invoke(
|
|
plan_app, ["tree", context.pec_plan_id, *flags.split()]
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# WHEN steps - orphan edge case
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when("pec I build the tree from orphan decisions")
|
|
def step_pec_build_orphan_tree(context: Context) -> None:
|
|
from cleveragents.cli.commands.plan import build_decision_tree
|
|
|
|
context.pec_tree = build_decision_tree(context.pec_orphan_decisions)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# WHEN steps - resolve active plan
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when("pec I call resolve active plan id")
|
|
def step_pec_resolve_active(context: Context) -> None:
|
|
from cleveragents.cli.commands.plan import _resolve_active_plan_id
|
|
|
|
with patch(_PATCH_LIFECYCLE, return_value=context.pec_lifecycle_svc):
|
|
context.pec_resolved_id = _resolve_active_plan_id()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# WHEN steps - revert
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when('pec I invoke "revert" with plan id and target "{target}"')
|
|
def step_pec_invoke_revert(context: Context, target: str) -> None:
|
|
with patch(_PATCH_LIFECYCLE, return_value=context.pec_lifecycle_svc):
|
|
context.pec_result = runner.invoke(
|
|
plan_app, ["revert", context.pec_plan_id, "--to-phase", target]
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# WHEN steps - correct
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _invoke_correct(
|
|
context: Context, extra_args: list[str] | None = None, input_text: str | None = None
|
|
) -> None:
|
|
args = [
|
|
"correct",
|
|
context.pec_decision_id,
|
|
"--mode",
|
|
"revert",
|
|
"--guidance",
|
|
"Use FastAPI instead",
|
|
"--plan",
|
|
context.pec_plan_id,
|
|
]
|
|
if extra_args:
|
|
args.extend(extra_args)
|
|
|
|
# Mock DecisionService resolved via DI container (issue #606 fix)
|
|
mock_decision_svc = MagicMock()
|
|
mock_decision_svc.list_decisions.return_value = []
|
|
mock_decision_svc.get_influence_edges.return_value = {}
|
|
mock_container = MagicMock()
|
|
mock_container.decision_service.return_value = mock_decision_svc
|
|
mock_container.correction_service.return_value = context.pec_correction_svc
|
|
|
|
with patch(_PATCH_CONTAINER, return_value=mock_container):
|
|
context.pec_result = runner.invoke(plan_app, args, input=input_text)
|
|
|
|
|
|
@when('pec I invoke "correct" in dry-run mode with rich format')
|
|
def step_pec_correct_dryrun(context: Context) -> None:
|
|
_invoke_correct(context, extra_args=["--dry-run"])
|
|
|
|
|
|
@when('pec I invoke "correct" with yes flag')
|
|
def step_pec_correct_yes(context: Context) -> None:
|
|
_invoke_correct(context, extra_args=["--yes"])
|
|
|
|
|
|
@when('pec I invoke "correct" without yes and decline')
|
|
def step_pec_correct_decline(context: Context) -> None:
|
|
_invoke_correct(context, input_text="n\n")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# WHEN steps - resume
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when('pec I invoke "resume" with a plan id')
|
|
def step_pec_invoke_resume(context: Context) -> None:
|
|
with (
|
|
patch(_PATCH_LIFECYCLE, return_value=MagicMock()),
|
|
patch(
|
|
_PATCH_RESUME_SVC_MOD,
|
|
return_value=context.pec_resume_svc,
|
|
),
|
|
):
|
|
context.pec_result = runner.invoke(plan_app, ["resume", context.pec_plan_id])
|
|
|
|
|
|
@when('pec I invoke "resume" with dry-run flag')
|
|
def step_pec_invoke_resume_dryrun(context: Context) -> None:
|
|
with (
|
|
patch(_PATCH_LIFECYCLE, return_value=MagicMock()),
|
|
patch(
|
|
_PATCH_RESUME_SVC_MOD,
|
|
return_value=context.pec_resume_svc,
|
|
),
|
|
):
|
|
context.pec_result = runner.invoke(
|
|
plan_app, ["resume", context.pec_plan_id, "--dry-run"]
|
|
)
|
|
|
|
|
|
@when('pec I invoke "resume" with format "{fmt}"')
|
|
def step_pec_invoke_resume_fmt(context: Context, fmt: str) -> None:
|
|
with (
|
|
patch(_PATCH_LIFECYCLE, return_value=MagicMock()),
|
|
patch(
|
|
_PATCH_RESUME_SVC_MOD,
|
|
return_value=context.pec_resume_svc,
|
|
),
|
|
):
|
|
context.pec_result = runner.invoke(
|
|
plan_app, ["resume", context.pec_plan_id, "--format", fmt]
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# WHEN steps - read-only guards
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when('pec I invoke "execute" with the read-only plan id')
|
|
def step_pec_invoke_execute_readonly(context: Context) -> None:
|
|
with patch(_PATCH_LIFECYCLE, return_value=context.pec_lifecycle_svc):
|
|
context.pec_result = runner.invoke(plan_app, ["execute", context.pec_plan_id])
|
|
|
|
|
|
@when('pec I invoke "apply" with the read-only plan id')
|
|
def step_pec_invoke_apply_readonly(context: Context) -> None:
|
|
with patch(_PATCH_LIFECYCLE, return_value=context.pec_lifecycle_svc):
|
|
context.pec_result = runner.invoke(
|
|
plan_app, ["apply", "--yes", context.pec_plan_id]
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# THEN steps
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@then("pec the exit code should be 0")
|
|
def step_pec_exit_0(context: Context) -> None:
|
|
assert context.pec_result.exit_code == 0, (
|
|
f"Expected exit 0, got {context.pec_result.exit_code}.\n"
|
|
f"Output: {context.pec_result.output}"
|
|
)
|
|
|
|
|
|
@then("pec the exit code should be {code:d}")
|
|
def step_pec_exit_code(context: Context, code: int) -> None:
|
|
assert context.pec_result.exit_code == code, (
|
|
f"Expected exit {code}, got {context.pec_result.exit_code}.\n"
|
|
f"Output: {context.pec_result.output}"
|
|
)
|
|
|
|
|
|
@then("pec the exit code should be nonzero")
|
|
def step_pec_exit_nonzero(context: Context) -> None:
|
|
assert context.pec_result.exit_code != 0, (
|
|
f"Expected nonzero exit, got {context.pec_result.exit_code}.\n"
|
|
f"Output: {context.pec_result.output}"
|
|
)
|
|
|
|
|
|
@then('pec the output should contain "{text}"')
|
|
def step_pec_output_contains(context: Context, text: str) -> None:
|
|
assert text in context.pec_result.output, (
|
|
f"Expected '{text}' in output.\nOutput: {context.pec_result.output}"
|
|
)
|
|
|
|
|
|
@then('pec the output should not contain "{text}"')
|
|
def step_pec_output_not_contains(context: Context, text: str) -> None:
|
|
assert text not in context.pec_result.output, (
|
|
f"Did not expect '{text}' in output.\nOutput: {context.pec_result.output}"
|
|
)
|
|
|
|
|
|
@then("pec the output should be valid json")
|
|
def step_pec_output_valid_json(context: Context) -> None:
|
|
parsed = json.loads(context.pec_result.output.strip())
|
|
data = _unwrap_envelope(parsed)
|
|
assert isinstance(data, dict), "Expected JSON object"
|
|
|
|
|
|
@then("pec the output should be valid json list")
|
|
def step_pec_output_valid_json_list(context: Context) -> None:
|
|
parsed = json.loads(context.pec_result.output.strip())
|
|
data = _unwrap_envelope(parsed)
|
|
assert isinstance(data, list), "Expected JSON array"
|
|
|
|
|
|
@then("pec the tree should exclude the superseded grandchild")
|
|
def step_pec_tree_excludes_orphan(context: Context) -> None:
|
|
# Tree should have exactly one root with one child and zero grandchildren.
|
|
assert len(context.pec_tree) == 1, f"Expected 1 root, got {len(context.pec_tree)}"
|
|
root = context.pec_tree[0]
|
|
assert root["decision_id"] == context.pec_orphan_root_id
|
|
children = root["children"]
|
|
assert len(children) == 1, ( # type: ignore[arg-type]
|
|
f"Expected 1 child, got {len(children)}" # type: ignore[arg-type]
|
|
)
|
|
# The grandchild was superseded and filtered from by_id; the BFS
|
|
# orphan guard must have skipped it, leaving no grandchildren.
|
|
grandchildren = children[0]["children"] # type: ignore[index]
|
|
assert len(grandchildren) == 0, ( # type: ignore[arg-type]
|
|
f"Expected 0 grandchildren, got {len(grandchildren)}" # type: ignore[arg-type]
|
|
)
|
|
|
|
|
|
@then("pec the resolved plan id should match the active plan")
|
|
def step_pec_resolved_matches(context: Context) -> None:
|
|
assert context.pec_resolved_id == context.pec_expected_plan_id
|