"""Step definitions for subplan service coverage boost scenarios. Targets uncovered lines in subplan_service.py: - Lines 129-131: SpawnValidationError.__init__ - Line 167: decision_service property - Line 207: spawn() config=None guard - Line 209: spawn() spawn_entries=None guard - Line 220: spawn() raising SpawnValidationError - Line 301: validate_spawn() config=None guard - Line 303: validate_spawn() spawn_entries=None guard - Lines 334-336: empty action_name validation - Line 375: get_spawn_decisions() empty plan_id guard - Line 414: build_spawn_entries() decisions=None guard """ from __future__ import annotations from unittest.mock import MagicMock from behave import given, then, when from behave.runner import Context from cleveragents.application.services.subplan_service import ( SpawnEntry, SpawnValidationError, SpawnValidationResult, SubplanService, ) from cleveragents.domain.models.core.decision import ( ContextSnapshot, Decision, DecisionType, ) from cleveragents.domain.models.core.plan import ( ExecutionMode, NamespacedName, Plan, PlanIdentity, SubplanConfig, ) # --------------------------------------------------------------------------- # Shared ULID constants # --------------------------------------------------------------------------- _COV_PLAN_ID = "01HGZ6FE0AQDYTR4BXVQZ6CV00" _COV_DEC_ID = "01HGZ6FE0AQDYTR4BXVQZ6CD00" # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _mock_decision_service_cov() -> MagicMock: """Create a mock DecisionService for coverage tests.""" svc = MagicMock() svc.list_by_type = MagicMock(return_value=[]) return svc def _make_decision_cov( decision_id: str = _COV_DEC_ID, decision_type: DecisionType = DecisionType.SUBPLAN_SPAWN, chosen_option: str = "local/cov-action", ) -> Decision: """Create a Decision for coverage tests.""" return Decision( decision_id=decision_id, plan_id=_COV_PLAN_ID, decision_type=decision_type, sequence_number=0, question="Coverage test spawn?", chosen_option=chosen_option, context_snapshot=ContextSnapshot(), ) def _make_plan_cov() -> Plan: """Create a minimal Plan for coverage tests.""" return Plan( identity=PlanIdentity(plan_id=_COV_PLAN_ID), namespaced_name=NamespacedName(namespace="local", name="cov-plan"), description="Coverage boost test plan", action_name="local/cov-action", ) # --------------------------------------------------------------------------- # Given steps # --------------------------------------------------------------------------- @given("a subplan service constructed with a mock decision service") def step_given_service_with_mock(context: Context) -> None: context.cov_mock_ds = _mock_decision_service_cov() context.cov_service = SubplanService(decision_service=context.cov_mock_ds) @given("a parent plan for coverage tests") def step_given_parent_plan_cov(context: Context) -> None: context.cov_parent_plan = _make_plan_cov() @given("a default subplan config for coverage tests") def step_given_default_config_cov(context: Context) -> None: context.cov_config = SubplanConfig() @given("a subplan config with merge strategy forcibly set to None") def step_given_config_merge_none(context: Context) -> None: config = SubplanConfig(execution_mode=ExecutionMode.SEQUENTIAL) # Force merge_strategy to None to trigger validation error object.__setattr__(config, "merge_strategy", None) context.cov_config = config @given("a spawn entry with a valid spawn decision for coverage tests") def step_given_valid_spawn_entry_cov(context: Context) -> None: dec = _make_decision_cov() context.cov_spawn_entries = [ SpawnEntry( decision=dec, action_name="local/cov-sub", description="Coverage spawn entry", ) ] @given("a spawn entry with an empty action_name") def step_given_entry_empty_action(context: Context) -> None: dec = _make_decision_cov() context.cov_spawn_entries = [ SpawnEntry( decision=dec, action_name="", description="Entry with empty action_name", ) ] @given("a spawn entry with a whitespace-only action_name") def step_given_entry_whitespace_action(context: Context) -> None: dec = _make_decision_cov() context.cov_spawn_entries = [ SpawnEntry( decision=dec, action_name=" ", description="Entry with whitespace action_name", ) ] # --------------------------------------------------------------------------- # When steps # --------------------------------------------------------------------------- @when('a SpawnValidationError is created with errors "{err1}" and "{err2}"') def step_when_create_spawn_validation_error( context: Context, err1: str, err2: str ) -> None: context.cov_sve = SpawnValidationError([err1, err2]) @when("the decision_service property is accessed") def step_when_access_property(context: Context) -> None: context.cov_returned_ds = context.cov_service.decision_service @when("spawn is called with config set to None") def step_when_spawn_config_none(context: Context) -> None: context.cov_error = None dec = _make_decision_cov() entries = [SpawnEntry(decision=dec, action_name="local/x")] try: context.cov_service.spawn( parent_plan=context.cov_parent_plan, config=None, # type: ignore[arg-type] spawn_entries=entries, ) except ValueError as exc: context.cov_error = exc @when("spawn is called with spawn_entries set to None") def step_when_spawn_entries_none(context: Context) -> None: context.cov_error = None try: context.cov_service.spawn( parent_plan=context.cov_parent_plan, config=context.cov_config, spawn_entries=None, # type: ignore[arg-type] ) except ValueError as exc: context.cov_error = exc @when("spawn is called and validation is expected to fail") def step_when_spawn_validation_fail(context: Context) -> None: context.cov_error = None try: context.cov_service.spawn( parent_plan=context.cov_parent_plan, config=context.cov_config, spawn_entries=context.cov_spawn_entries, ) except SpawnValidationError as exc: context.cov_error = exc @when("validate_spawn is called with config set to None") def step_when_validate_config_none(context: Context) -> None: context.cov_error = None try: context.cov_service.validate_spawn( config=None, # type: ignore[arg-type] spawn_entries=context.cov_spawn_entries, ) except ValueError as exc: context.cov_error = exc @when("validate_spawn is called with spawn_entries set to None") def step_when_validate_entries_none(context: Context) -> None: context.cov_error = None try: context.cov_service.validate_spawn( config=context.cov_config, spawn_entries=None, # type: ignore[arg-type] ) except ValueError as exc: context.cov_error = exc @when("validate_spawn is called for coverage tests") def step_when_validate_cov(context: Context) -> None: context.cov_validation_result = context.cov_service.validate_spawn( config=context.cov_config, spawn_entries=context.cov_spawn_entries, ) @when("get_spawn_decisions is called with an empty plan_id") def step_when_get_decisions_empty(context: Context) -> None: context.cov_error = None try: context.cov_service.get_spawn_decisions("") except ValueError as exc: context.cov_error = exc @when("get_spawn_decisions is called with a whitespace-only plan_id") def step_when_get_decisions_whitespace(context: Context) -> None: context.cov_error = None try: context.cov_service.get_spawn_decisions(" ") except ValueError as exc: context.cov_error = exc @when("build_spawn_entries is called with decisions set to None") def step_when_build_entries_none(context: Context) -> None: context.cov_error = None try: context.cov_service.build_spawn_entries(None) # type: ignore[arg-type] except ValueError as exc: context.cov_error = exc # --------------------------------------------------------------------------- # Then steps # --------------------------------------------------------------------------- @then("the SpawnValidationError validation_errors list should have {n:d} entries") def step_then_sve_count(context: Context, n: int) -> None: assert len(context.cov_sve.validation_errors) == n, ( f"Expected {n} errors, got {len(context.cov_sve.validation_errors)}" ) @then('the SpawnValidationError message should contain "{text}"') def step_then_sve_message(context: Context, text: str) -> None: msg = str(context.cov_sve) assert text in msg, f"Expected '{text}' in message: {msg}" @then("the returned decision service should be the injected mock") def step_then_ds_is_same(context: Context) -> None: assert context.cov_returned_ds is context.cov_mock_ds, ( "Property did not return the injected decision service" ) @then('a ValueError should be raised with text "{text}"') def step_then_value_error_text(context: Context, text: str) -> None: assert context.cov_error is not None, ( f"Expected ValueError with '{text}' but no error was raised" ) assert isinstance(context.cov_error, ValueError), ( f"Expected ValueError, got {type(context.cov_error).__name__}" ) assert text in str(context.cov_error), f"Expected '{text}' in '{context.cov_error}'" @then("a SpawnValidationError should be raised") def step_then_sve_raised(context: Context) -> None: assert context.cov_error is not None, ( "Expected SpawnValidationError but no error was raised" ) assert isinstance(context.cov_error, SpawnValidationError), ( f"Expected SpawnValidationError, got {type(context.cov_error).__name__}" ) @then('the SpawnValidationError should mention "{text}"') def step_then_sve_mentions(context: Context, text: str) -> None: msg = str(context.cov_error) assert text.lower() in msg.lower(), f"Expected '{text}' in error message: {msg}" @then("the validation result should not be valid") def step_then_invalid_cov(context: Context) -> None: result: SpawnValidationResult = context.cov_validation_result assert not result.valid, ( f"Expected invalid result, but got valid (errors={result.errors})" ) @then('the validation errors should mention "{text}"') def step_then_errors_mention(context: Context, text: str) -> None: # Support two contexts: SpawnValidationResult (subplan) and plain list (uko) cov_result = getattr(context, "cov_validation_result", None) if cov_result is not None: errors_joined = " ".join(cov_result.errors).lower() assert text.lower() in errors_joined, ( f"Expected '{text}' in errors: {cov_result.errors}" ) else: errors_list = getattr(context, "validation_errors", None) assert errors_list is not None, ( "Neither cov_validation_result nor validation_errors found on context" ) errors_joined = " ".join(errors_list).lower() assert text.lower() in errors_joined, ( f"Expected '{text}' in errors: {errors_list}" )