"""Step definitions for container_resolve_crash.feature. Regression tests for bug #647 covering ``plan tree``, ``plan explain``, and ``plan correct`` command paths through a real DI container. All step text uses the ``cr647-`` prefix to avoid collisions with other step files. """ from __future__ import annotations import os from contextlib import suppress from behave import given, then, when from behave.runner import Context from typer.testing import CliRunner from ulid import ULID from cleveragents.infrastructure.database.engine_cache import MEMORY_ENGINES cli_runner = CliRunner() # --------------------------------------------------------------------------- # GIVEN — Real container with seeded decisions # --------------------------------------------------------------------------- @given("cr647- a real DI container with seeded decisions") def step_cr647_setup_container(context: Context) -> None: """Set up a real DI container with seeded decision tree. Uses the real Container class (not MagicMock) with seeded decisions so command output can validate DI-resolved service behavior. """ from cleveragents.application.container import get_container, reset_container from cleveragents.application.services.decision_service import DecisionService from cleveragents.application.services.plan_lifecycle_service import ( PlanLifecycleService, ) from cleveragents.config.settings import Settings from cleveragents.domain.models.core.decision import ( ContextSnapshot, Decision, DecisionType, ResourceRef, ) from cleveragents.infrastructure.database.unit_of_work import UnitOfWork # Reuse the per-scenario DB URL set by features/environment.py to avoid # orphaning temp-file bookkeeping and to keep cleanup semantics aligned. database_url = os.environ.get("CLEVERAGENTS_DATABASE_URL", "sqlite:///:memory:") # Reset container to ensure clean state reset_container() # Create UnitOfWork and initialize database uow = UnitOfWork(database_url) uow.init_database() # Use real settings for integration-style behavior (no mocks). settings = Settings(database_url=database_url, async_enabled=False) # Create DecisionService and seed decisions decision_svc = DecisionService(settings=settings, unit_of_work=uow) lifecycle_svc = PlanLifecycleService(settings=settings, unit_of_work=uow) # Seed a small tree (root + child) so tree/explain/correct assertions # can validate command output beyond crash absence. action_name = f"local/cr647-{ULID()!s}" lifecycle_svc.create_action( name=action_name, description="Container.resolve regression action", definition_of_done="Regression guard setup complete", strategy_actor="openai/gpt-4", execution_actor="openai/gpt-4", ) plan = lifecycle_svc.use_action(action_name) plan_id = plan.identity.plan_id root: Decision = decision_svc.record_decision( plan_id=plan_id, decision_type=DecisionType.PROMPT_DEFINITION, question="What should we build?", chosen_option="A REST API", alternatives_considered=["GraphQL API", "gRPC service"], confidence_score=0.95, rationale="REST fits requirements", context_snapshot=ContextSnapshot( hot_context_hash="sha256:root", hot_context_ref="store://root", relevant_resources=[ ResourceRef(resource_id=str(ULID()), path="src/main.py"), ], actor_state_ref="checkpoint://root", ), ) child: Decision = decision_svc.record_decision( plan_id=plan_id, decision_type=DecisionType.STRATEGY_CHOICE, question="Which framework should we use?", chosen_option="FastAPI", parent_decision_id=root.decision_id, alternatives_considered=["Flask", "Django"], confidence_score=0.90, rationale="FastAPI fits async service requirements", context_snapshot=ContextSnapshot( hot_context_hash="sha256:child", hot_context_ref="store://child", relevant_resources=[ ResourceRef(resource_id=str(ULID()), path="src/api.py"), ], actor_state_ref="checkpoint://child", ), ) # Store IDs for use in test steps. # explain/correct only need one valid decision_id to trigger the resolve() path. context.cr647_plan_id = plan_id context.cr647_child_id = child.decision_id # Pre-initialize the container singleton so CLI commands get DI wiring # from the seeded database. _ = get_container() # Store cleanup handler — capture ``uow`` so its engine is explicitly # disposed even when the URL is not in ``MEMORY_ENGINES`` (file-based). def cleanup() -> None: """Reset in-memory DB/cache and process-global singletons.""" with suppress(Exception): uow.engine.dispose() with suppress(Exception): engine = MEMORY_ENGINES.pop(database_url, None) if engine is not None: engine.dispose() with suppress(Exception): reset_container() with suppress(Exception): os.environ.pop("CLEVERAGENTS_DATABASE_URL", None) # Keep reset last so any env-dependent lazy settings re-init cannot leak. with suppress(Exception): Settings.reset() context.add_cleanup(cleanup) # --------------------------------------------------------------------------- # WHEN — Invoke CLI commands with real container # --------------------------------------------------------------------------- @when("cr647- I invoke the plan tree CLI command with a real container") def step_cr647_invoke_plan_tree(context: Context) -> None: """Invoke plan tree command and capture regression output.""" from cleveragents.cli.main import app as cli_app context.cr647_result = cli_runner.invoke( cli_app, ["plan", "tree", context.cr647_plan_id, "--format", "json"], # Capture exceptions so the step can assert exact failure class and output. catch_exceptions=True, ) @when("cr647- I invoke the plan explain CLI command with a real container") def step_cr647_invoke_plan_explain(context: Context) -> None: """Invoke plan explain command and capture regression output.""" from cleveragents.cli.main import app as cli_app context.cr647_result = cli_runner.invoke( cli_app, [ "plan", "explain", context.cr647_child_id, "--show-context", "--show-reasoning", "--format", "json", ], # Capture exceptions so the step can assert exact failure class and output. catch_exceptions=True, ) @when("cr647- I invoke the plan correct CLI command with a real container") def step_cr647_invoke_plan_correct(context: Context) -> None: """Invoke plan correct command and capture regression output.""" from cleveragents.cli.main import app as cli_app context.cr647_result = cli_runner.invoke( cli_app, [ "plan", "correct", context.cr647_child_id, "--mode", "revert", "--guidance", "Use Django instead", # NOTE: ``--plan`` is currently required by implementation when there # is no active plan in CLI context, though spec examples show omitted. "--plan", context.cr647_plan_id, "--dry-run", "--format", "json", ], # Capture exceptions so the step can assert exact failure class and output. catch_exceptions=True, ) # --------------------------------------------------------------------------- # THEN — Verify successful command execution # --------------------------------------------------------------------------- def _assert_command_succeeded(context: Context) -> None: """Assert common post-fix regression behavior for command execution.""" result = context.cr647_result # Defense-in-depth: if failure happens, ensure it matches historical bug class. if result.exit_code != 0 and result.exception is not None: assert isinstance(result.exception, AttributeError), ( "Unexpected non-AttributeError exception. Got " f"{type(result.exception).__name__}: {result.exception}" ) assert result.exit_code == 0, ( f"Expected exit code 0, got {result.exit_code}. " f"Output: {result.output}\nException: {result.exception}" ) assert result.exception is None, ( f"Expected no exception, got {type(result.exception).__name__}: " f"{result.exception}\nOutput: {result.output}" ) # Guard against silent error traces while still returning 0. output_lower = (result.output or "").lower() assert "attributeerror" not in output_lower, ( f"Did not expect AttributeError text in output:\n{result.output}" ) assert "has no attribute 'resolve'" not in output_lower, ( f"Did not expect resolve() attribute crash text in output:\n{result.output}" ) @then("cr647- plan tree output should include seeded decision id") def step_cr647_tree_output_contains_seeded_id(context: Context) -> None: """Verify tree command output contains seeded decision id.""" _assert_command_succeeded(context) result = context.cr647_result assert context.cr647_child_id in (result.output or ""), ( "Expected plan tree output to include seeded decision id. " f"Output: {result.output}" ) @then("cr647- plan explain output should include seeded decision details") def step_cr647_explain_output_contains_seeded_details(context: Context) -> None: """Verify explain command output includes seeded id and content.""" _assert_command_succeeded(context) result = context.cr647_result assert context.cr647_child_id in (result.output or ""), ( f"Expected plan explain output to include decision id. Output: {result.output}" ) assert "FastAPI" in (result.output or ""), ( "Expected plan explain output to include child decision content " "(FastAPI). " f"Output: {result.output}" ) @then("cr647- plan correct output should reference seeded decision") def step_cr647_correct_output_references_seeded_decision(context: Context) -> None: """Verify correct command output references seeded decision id and mode.""" _assert_command_succeeded(context) result = context.cr647_result output = result.output or "" assert context.cr647_child_id in output, ( "Expected plan correct output to reference seeded decision id. " f"Output: {output}" ) # Validate command-specific content: dry-run mode and guidance text. output_lower = output.lower() assert "revert" in output_lower or "dry" in output_lower, ( "Expected plan correct output to reference revert mode or dry-run. " f"Output: {output}" )