aditya
97920678b9
Fix unit test and integration test failures on Forgejo CI caused by Rich/Click text wrapping in narrow terminal environments. The assertion for "--no-estimate cannot be used with --estimation-actor" was split across lines when COLUMNS was small, breaking exact substring checks. - Normalize whitespace before assertion in cli_extensions_steps.py so the check is resilient to line wrapping regardless of terminal width. - Set COLUMNS=240 in robot/helper_plan_cli_spec.py to give CLI subprocesses a consistent wide terminal, preventing Rich from reflowing output mid-assertion. - Add 16 new BDD scenarios to estimation_coverage.feature exercising previously-uncovered estimation branches: validation constraints (max_length, min_length, frozen model), EstimationActorProtocol runtime check, all three exception handler paths in estimate_plan, parse_estimation_report generic error path, serialization of non-None values, EstimationError attribute preservation, and CLI dict / status detail rendering of estimation data. - estimation_service.py coverage: 75% → 99% (only TYPE_CHECKING and Protocol stub lines remain uncovered). - Overall coverage: 96.89% → 97.0%, meeting the project threshold. All changes are scoped to estimation feature tests and the estimation CLI assertion helpers introduced by this branch. Refs: #209
522 lines
17 KiB
Python
522 lines
17 KiB
Python
"""Step definitions for estimation_coverage.feature."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from typing import TYPE_CHECKING
|
|
|
|
from behave import given, then, when
|
|
from pydantic import ValidationError
|
|
|
|
from cleveragents.application.services.estimation_service import (
|
|
EstimationActorProtocol,
|
|
EstimationError,
|
|
EstimationService,
|
|
)
|
|
from cleveragents.domain.models.core.action import Action, ActionArgument, ActionState
|
|
from cleveragents.domain.models.core.estimation import (
|
|
EstimationReport,
|
|
EstimationSkipped,
|
|
)
|
|
from cleveragents.domain.models.core.plan import ProcessingState, ProjectLink
|
|
from features.mocks.estimation_factories import make_estimation_report
|
|
|
|
if TYPE_CHECKING:
|
|
from behave.runner import Context
|
|
|
|
|
|
@when('I create an estimation report with invalid actor "{actor}"')
|
|
def step_invalid_actor(context: Context, actor: str) -> None:
|
|
try:
|
|
EstimationReport(
|
|
cost_range_usd_min=0.5,
|
|
cost_range_usd_max=1.5,
|
|
expected_steps=2,
|
|
expected_child_plans=0,
|
|
rollback_risk=0.2,
|
|
confidence=0.8,
|
|
estimated_duration_minutes=10.0,
|
|
actor_used=actor,
|
|
)
|
|
context.error = None
|
|
except ValidationError as exc:
|
|
context.error = exc
|
|
|
|
|
|
@then("a validation error should occur")
|
|
def step_validation_error_occurs(context: Context) -> None:
|
|
assert isinstance(context.error, ValidationError)
|
|
|
|
|
|
@then('the coverage validation error should mention "{text}"')
|
|
def step_validation_error_mentions(context: Context, text: str) -> None:
|
|
assert text in str(context.error)
|
|
|
|
|
|
@given("an estimation service")
|
|
def step_create_service(context: Context) -> None:
|
|
context.estimation_service = EstimationService()
|
|
|
|
|
|
@when("I serialize a None estimation report")
|
|
def step_serialize_none_report(context: Context) -> None:
|
|
context.serialization_result = (
|
|
context.estimation_service.serialize_estimation_report(None)
|
|
)
|
|
|
|
|
|
@when("I serialize a None estimation skipped")
|
|
def step_serialize_none_skipped(context: Context) -> None:
|
|
context.serialization_result = (
|
|
context.estimation_service.serialize_estimation_skipped(None)
|
|
)
|
|
|
|
|
|
@then("the serialization result should be None")
|
|
def step_assert_none(context: Context) -> None:
|
|
assert context.serialization_result is None
|
|
|
|
|
|
@when('I parse estimation report with invalid JSON "{json_str}"')
|
|
def step_parse_invalid_json(context: Context, json_str: str) -> None:
|
|
try:
|
|
context.estimation_service.parse_estimation_report(json_str, "local/test")
|
|
context.estimation_error = None
|
|
except EstimationError as exc:
|
|
context.estimation_error = exc
|
|
|
|
|
|
@when("I parse estimation report that causes a parse error")
|
|
def step_parse_invalid_payload(context: Context) -> None:
|
|
bad_payload = json.dumps({"expected_steps": "bad-type"})
|
|
try:
|
|
context.estimation_service.parse_estimation_report(bad_payload, "local/test")
|
|
context.estimation_error = None
|
|
except EstimationError as exc:
|
|
context.estimation_error = exc
|
|
|
|
|
|
@then("an estimation error should be raised")
|
|
def step_estimation_error(context: Context) -> None:
|
|
assert isinstance(context.estimation_error, EstimationError)
|
|
|
|
|
|
@then('the coverage estimation error should mention "{text}"')
|
|
def step_error_mentions(context: Context, text: str) -> None:
|
|
assert text in str(context.estimation_error)
|
|
|
|
|
|
@when('I parse estimation report without actor_used field for actor "{actor}"')
|
|
def step_parse_without_actor(context: Context, actor: str) -> None:
|
|
payload = json.dumps(
|
|
{
|
|
"cost_range_usd_min": 0.5,
|
|
"cost_range_usd_max": 1.5,
|
|
"expected_steps": 3,
|
|
"expected_child_plans": 1,
|
|
"rollback_risk": 0.2,
|
|
"confidence": 0.8,
|
|
"estimated_duration_minutes": 20.0,
|
|
}
|
|
)
|
|
context.parsed_estimation = context.estimation_service.parse_estimation_report(
|
|
payload, actor
|
|
)
|
|
|
|
|
|
@when(
|
|
'I parse estimation report from dict input without actor_used for actor "{actor}"'
|
|
)
|
|
def step_parse_without_actor_from_dict(context: Context, actor: str) -> None:
|
|
context.original_payload_dict = {
|
|
"cost_range_usd_min": 0.5,
|
|
"cost_range_usd_max": 1.5,
|
|
"expected_steps": 3,
|
|
"expected_child_plans": 1,
|
|
"rollback_risk": 0.2,
|
|
"confidence": 0.8,
|
|
"estimated_duration_minutes": 20.0,
|
|
}
|
|
context.parsed_estimation = context.estimation_service.parse_estimation_report(
|
|
context.original_payload_dict,
|
|
actor,
|
|
)
|
|
|
|
|
|
@then('the parsed estimation should have actor "{actor}"')
|
|
def step_parsed_actor(context: Context, actor: str) -> None:
|
|
assert context.parsed_estimation.actor_used == actor
|
|
|
|
|
|
@then('the original estimation dict should not contain key "{key}"')
|
|
def step_original_dict_unchanged(context: Context, key: str) -> None:
|
|
assert key not in context.original_payload_dict
|
|
|
|
|
|
@given("the action has invariants")
|
|
def step_action_has_invariants(context: Context) -> None:
|
|
context.action = Action(
|
|
namespaced_name=context.action.namespaced_name,
|
|
description=context.action.description,
|
|
definition_of_done=context.action.definition_of_done,
|
|
strategy_actor=context.action.strategy_actor,
|
|
execution_actor=context.action.execution_actor,
|
|
estimation_actor=context.action.estimation_actor,
|
|
invariants=["test_invariant_for_coverage"],
|
|
)
|
|
|
|
|
|
@given("the action has {count:d} project links")
|
|
def step_action_projects(context: Context, count: int) -> None:
|
|
context.project_count = count
|
|
|
|
|
|
@given("the action has custom arguments")
|
|
def step_action_custom_arguments(context: Context) -> None:
|
|
context.action.arguments = [
|
|
ActionArgument(
|
|
name="test_arg",
|
|
description="coverage arg",
|
|
required=False,
|
|
default_value="value",
|
|
)
|
|
]
|
|
context.has_custom_arguments = True
|
|
|
|
|
|
@when("I use the action to create a plan")
|
|
def step_use_action_create_plan(context: Context) -> None:
|
|
from cleveragents.application.services.plan_lifecycle_service import (
|
|
PlanLifecycleService,
|
|
)
|
|
from cleveragents.config.settings import Settings
|
|
|
|
lifecycle_service = PlanLifecycleService(
|
|
settings=Settings(),
|
|
estimation_service=EstimationService(),
|
|
)
|
|
context.lifecycle_service = lifecycle_service
|
|
|
|
if context.action.state != ActionState.AVAILABLE:
|
|
context.action.state = ActionState.AVAILABLE
|
|
lifecycle_service._actions[str(context.action.namespaced_name)] = context.action
|
|
|
|
project_count = getattr(context, "project_count", 1)
|
|
project_links = [
|
|
ProjectLink(project_name=f"project-{i}") for i in range(project_count)
|
|
]
|
|
arguments = (
|
|
{"test_arg": "value"}
|
|
if getattr(context, "has_custom_arguments", False)
|
|
else None
|
|
)
|
|
|
|
context.created_plan = lifecycle_service.use_action(
|
|
action_name=str(context.action.namespaced_name),
|
|
project_links=project_links,
|
|
arguments=arguments,
|
|
skip_estimation=getattr(context, "skip_estimation", False),
|
|
)
|
|
|
|
|
|
@when("I transition the created plan to execute")
|
|
def step_transition_created_plan(context: Context) -> None:
|
|
context.created_plan.processing_state = ProcessingState.COMPLETE
|
|
context.created_plan = context.lifecycle_service.execute_plan(
|
|
context.created_plan.identity.plan_id
|
|
)
|
|
|
|
|
|
@then("the plan should have estimation report")
|
|
def step_plan_has_report(context: Context) -> None:
|
|
assert context.created_plan.estimation_report is not None
|
|
|
|
|
|
@then("the cost should reflect invariant overhead")
|
|
def step_cost_reflects_invariants(context: Context) -> None:
|
|
report = context.created_plan.estimation_report
|
|
assert report is not None
|
|
assert report.cost_range_usd_min > 0.35
|
|
|
|
|
|
@then('the risk factors should mention "{text}"')
|
|
def step_rationale_mentions(context: Context, text: str) -> None:
|
|
report = context.created_plan.estimation_report
|
|
assert report is not None
|
|
assert text in report.rationale
|
|
|
|
|
|
@then("the duration estimate should be increased for invariants")
|
|
def step_duration_increase(context: Context) -> None:
|
|
report = context.created_plan.estimation_report
|
|
assert report is not None
|
|
assert report.estimated_duration_minutes >= 36.0
|
|
|
|
|
|
@when(
|
|
'I update plan overrides with strategy_actor "{strategy}" '
|
|
'and execution_actor "{execution}"'
|
|
)
|
|
def step_update_plan_overrides(context: Context, strategy: str, execution: str) -> None:
|
|
context.override_created_at = context.created_plan.timestamps.created_at
|
|
context.overridden_plan = context.lifecycle_service.update_plan_overrides(
|
|
context.created_plan.identity.plan_id,
|
|
strategy_actor=strategy,
|
|
execution_actor=execution,
|
|
)
|
|
|
|
|
|
@then('the overridden plan should have strategy_actor "{expected}"')
|
|
def step_overridden_strategy(context: Context, expected: str) -> None:
|
|
assert context.overridden_plan.strategy_actor == expected
|
|
|
|
|
|
@then('the overridden plan should have execution_actor "{expected}"')
|
|
def step_overridden_execution(context: Context, expected: str) -> None:
|
|
assert context.overridden_plan.execution_actor == expected
|
|
|
|
|
|
@then("the overridden plan updated_at should be later than creation time")
|
|
def step_overridden_updated_at(context: Context) -> None:
|
|
assert context.overridden_plan.timestamps.updated_at >= context.override_created_at
|
|
|
|
|
|
# --- Rationale max_length validation ---
|
|
|
|
|
|
@when("I create an estimation report with rationale exceeding 10000 characters")
|
|
def step_create_report_long_rationale(context: Context) -> None:
|
|
try:
|
|
make_estimation_report(rationale="x" * 10_001)
|
|
context.error = None
|
|
except ValidationError as exc:
|
|
context.error = exc
|
|
|
|
|
|
# --- EstimationSkipped empty reason validation ---
|
|
|
|
|
|
@when("I create an estimation skipped with empty reason")
|
|
def step_create_skipped_empty_reason(context: Context) -> None:
|
|
try:
|
|
EstimationSkipped(reason="")
|
|
context.error = None
|
|
except ValidationError as exc:
|
|
context.error = exc
|
|
|
|
|
|
# --- Frozen model mutation ---
|
|
|
|
|
|
@given("a valid estimation report")
|
|
def step_valid_estimation_report(context: Context) -> None:
|
|
context.estimation_report = make_estimation_report()
|
|
|
|
|
|
@when("I attempt to mutate the estimation report confidence")
|
|
def step_mutate_report_confidence(context: Context) -> None:
|
|
try:
|
|
context.estimation_report.confidence = 0.99 # type: ignore[misc]
|
|
context.error = None
|
|
except ValidationError as exc:
|
|
context.error = exc
|
|
|
|
|
|
# --- Historical basis tuple ---
|
|
|
|
|
|
@when("I create an estimation report with historical basis entries")
|
|
def step_create_report_with_history(context: Context) -> None:
|
|
context.estimation_report = make_estimation_report(
|
|
historical_basis=("plan-001", "plan-002"),
|
|
)
|
|
|
|
|
|
@then("the historical basis should be a tuple")
|
|
def step_history_is_tuple(context: Context) -> None:
|
|
assert isinstance(context.estimation_report.historical_basis, tuple)
|
|
|
|
|
|
@then("the historical basis should contain the expected entries")
|
|
def step_history_entries(context: Context) -> None:
|
|
assert context.estimation_report.historical_basis == ("plan-001", "plan-002")
|
|
|
|
|
|
# --- EstimationActorProtocol runtime check ---
|
|
|
|
|
|
@when("I check a conforming class against EstimationActorProtocol")
|
|
def step_check_protocol(context: Context) -> None:
|
|
from cleveragents.domain.models.core.plan import Plan
|
|
|
|
class _ConformingActor:
|
|
def estimate(self, plan: Plan) -> EstimationReport:
|
|
return make_estimation_report()
|
|
|
|
context.conforming_instance = _ConformingActor()
|
|
|
|
|
|
@then("the conforming class should satisfy the protocol")
|
|
def step_conforming_satisfies(context: Context) -> None:
|
|
assert isinstance(context.conforming_instance, EstimationActorProtocol)
|
|
|
|
|
|
@then("the estimation service should not satisfy the protocol")
|
|
def step_service_not_protocol(context: Context) -> None:
|
|
assert not isinstance(EstimationService(), EstimationActorProtocol)
|
|
|
|
|
|
# --- EstimationError attributes ---
|
|
|
|
|
|
@when('I create an estimation error with actor "{actor}" and reason "{reason}"')
|
|
def step_create_estimation_error(context: Context, actor: str, reason: str) -> None:
|
|
context.estimation_error = EstimationError(actor_name=actor, reason=reason)
|
|
|
|
|
|
@then('the estimation error actor_name should be "{expected}"')
|
|
def step_error_actor(context: Context, expected: str) -> None:
|
|
assert context.estimation_error.actor_name == expected
|
|
|
|
|
|
@then('the estimation error reason should be "{expected}"')
|
|
def step_error_reason(context: Context, expected: str) -> None:
|
|
assert context.estimation_error.reason == expected
|
|
|
|
|
|
@then('the estimation error message should contain "{text}"')
|
|
def step_error_message_contains(context: Context, text: str) -> None:
|
|
assert text in str(context.estimation_error)
|
|
|
|
|
|
# --- estimate_plan exception handler coverage ---
|
|
|
|
|
|
@given("the estimation service has a patched actor raising EstimationError")
|
|
def step_patch_actor_estimation_error(context: Context) -> None:
|
|
original_service = context.estimation_service
|
|
|
|
def _raise_estimation_error(plan, actor_name):
|
|
raise EstimationError(actor_name=actor_name, reason="test failure")
|
|
|
|
original_service._invoke_estimation_actor = _raise_estimation_error # type: ignore[assignment]
|
|
|
|
|
|
@given("the estimation service has a patched actor raising TypeError")
|
|
def step_patch_actor_type_error(context: Context) -> None:
|
|
original_service = context.estimation_service
|
|
|
|
def _raise_type_error(plan, actor_name):
|
|
raise TypeError("unexpected type mismatch")
|
|
|
|
original_service._invoke_estimation_actor = _raise_type_error # type: ignore[assignment]
|
|
|
|
|
|
@given("the estimation service has a patched actor raising RuntimeError")
|
|
def step_patch_actor_runtime_error(context: Context) -> None:
|
|
original_service = context.estimation_service
|
|
|
|
def _raise_runtime_error(plan, actor_name):
|
|
raise RuntimeError("unexpected failure")
|
|
|
|
original_service._invoke_estimation_actor = _raise_runtime_error # type: ignore[assignment]
|
|
|
|
|
|
# --- Serialize non-None values ---
|
|
|
|
|
|
@when("I serialize a valid estimation report")
|
|
def step_serialize_valid_report(context: Context) -> None:
|
|
report = make_estimation_report()
|
|
context.serialization_result = (
|
|
context.estimation_service.serialize_estimation_report(report)
|
|
)
|
|
|
|
|
|
@when("I serialize a valid estimation skipped")
|
|
def step_serialize_valid_skipped(context: Context) -> None:
|
|
skipped = EstimationSkipped(reason="manual skip")
|
|
context.serialization_result = (
|
|
context.estimation_service.serialize_estimation_skipped(skipped)
|
|
)
|
|
|
|
|
|
@then("the serialization result should be a JSON string")
|
|
def step_result_is_json_string(context: Context) -> None:
|
|
assert isinstance(context.serialization_result, str)
|
|
json.loads(context.serialization_result)
|
|
|
|
|
|
@then('the serialized JSON should contain key "{key}"')
|
|
def step_serialized_json_key(context: Context, key: str) -> None:
|
|
parsed = json.loads(context.serialization_result)
|
|
assert key in parsed
|
|
|
|
|
|
@then('the estimation result reason should include "{text}"')
|
|
def step_estimation_result_reason_includes(context: Context, text: str) -> None:
|
|
assert isinstance(context.estimation_result, EstimationSkipped)
|
|
assert text in context.estimation_result.reason
|
|
|
|
|
|
# --- Parse estimation report with generic error ---
|
|
|
|
|
|
@when("I parse estimation report with input causing a generic error")
|
|
def step_parse_generic_error(context: Context) -> None:
|
|
class _BrokenDict(dict):
|
|
def copy(self):
|
|
raise RuntimeError("broken copy")
|
|
|
|
try:
|
|
context.estimation_service.parse_estimation_report(
|
|
_BrokenDict({"x": 1}), "local/test"
|
|
)
|
|
context.estimation_error = None
|
|
except EstimationError as exc:
|
|
context.estimation_error = exc
|
|
|
|
|
|
# --- CLI dict rendering for estimation ---
|
|
|
|
|
|
@when("I render the created plan as a CLI dict")
|
|
def step_render_created_plan_cli_dict(context: Context) -> None:
|
|
context.cli_dict = context.created_plan.as_cli_dict()
|
|
|
|
|
|
@then('the CLI dict estimation_report should contain "{key}"')
|
|
def step_cli_dict_report_key(context: Context, key: str) -> None:
|
|
report = context.cli_dict["estimation_report"]
|
|
assert key in report
|
|
|
|
|
|
@then('the CLI dict estimation_skipped should contain "{key}"')
|
|
def step_cli_dict_skipped_key(context: Context, key: str) -> None:
|
|
skipped = context.cli_dict["estimation_skipped"]
|
|
assert key in skipped
|
|
|
|
|
|
# --- Plan status detail rendering ---
|
|
|
|
|
|
@when("I render the plan status detail")
|
|
def step_render_status_detail(context: Context) -> None:
|
|
import io
|
|
from unittest.mock import patch
|
|
|
|
from rich.console import Console
|
|
|
|
from cleveragents.cli.commands.plan import _print_lifecycle_plan
|
|
|
|
buf = io.StringIO()
|
|
console = Console(file=buf, width=200, no_color=True)
|
|
with patch("cleveragents.cli.commands.plan.console", console):
|
|
_print_lifecycle_plan(context.created_plan)
|
|
context.status_detail = buf.getvalue()
|
|
|
|
|
|
@then('the status detail should contain "{text}"')
|
|
def step_status_detail_contains(context: Context, text: str) -> None:
|
|
assert text in context.status_detail
|