"""Step definitions for db_repositories_coverage_boost.feature. Targets specific uncovered lines in repositories.py: - Lines 1080, 1083: ActionRepository.update - arg_type / requirement as plain strings (not enums with .value), hitting the ``else`` branches. - Line 1340: LifecyclePlanRepository.update - automation_profile truthy path. - Line 1357: LifecyclePlanRepository.update - validation_summary not None. - Line 1367: LifecyclePlanRepository.update - execution_env_priority.value. - Lines 1472-1473: LifecyclePlanRepository.list_all - DatabaseError branch. - Lines 1730, 1734, 1737: ResourceTypeRepository.create - namespaced name with "/" and string (non-enum) resource_kind / sandbox_strategy. - Lines 1857, 1862, 1867-1872: ResourceTypeRepository.update - string enums and cli_args serialization. """ from __future__ import annotations from datetime import datetime from types import SimpleNamespace from unittest.mock import MagicMock from behave import given, then, when from behave.runner import Context from sqlalchemy import create_engine from sqlalchemy.exc import OperationalError from sqlalchemy.orm import Session, sessionmaker from cleveragents.core.exceptions import DatabaseError from cleveragents.domain.models.core.action import ( Action, ActionArgument, ActionState, ) from cleveragents.domain.models.core.plan import ( AutomationProfileProvenance, AutomationProfileRef, ExecutionEnvPriority, NamespacedName, Plan, PlanIdentity, PlanPhase, PlanTimestamps, ProcessingState, ) from cleveragents.domain.models.core.resource_type import ( ResourceKind, ResourceTypeArgument, SandboxStrategy, ) from cleveragents.infrastructure.database.models import Base from cleveragents.infrastructure.database.repositories import ( ActionRepository, LifecyclePlanRepository, ResourceTypeRepository, ) # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _next_ulid() -> str: from ulid import ULID return str(ULID()) def _make_action( name: str = "local/test-action", state: str = "available", arguments: list[ActionArgument] | None = None, invariants: list[str] | None = None, ) -> Action: parts = name.split("/", 1) namespace = parts[0] if len(parts) == 2 else "local" short_name = parts[1] if len(parts) == 2 else parts[0] return Action( namespaced_name=NamespacedName(namespace=namespace, name=short_name), description=f"Test action {short_name}", long_description=None, definition_of_done=f"Verify {short_name} completes", strategy_actor="local/strategist", execution_actor="local/executor", estimation_actor=None, review_actor=None, arguments=arguments or [], invariants=invariants or [], reusable=True, read_only=False, state=ActionState(state), created_at=datetime.now(), updated_at=datetime.now(), created_by=None, tags=[], ) def _make_plan( action_name: str, plan_id: str | None = None, phase: PlanPhase = PlanPhase.STRATEGIZE, ns_name: str = "local/test-plan", automation_profile: AutomationProfileRef | None = None, validation_summary: dict | None = None, execution_environment: str | None = None, execution_env_priority: ExecutionEnvPriority | None = None, ) -> Plan: pid = plan_id or _next_ulid() parts = ns_name.split("/", 1) namespace = parts[0] if len(parts) == 2 else "local" short_name = parts[1] if len(parts) == 2 else parts[0] return Plan( identity=PlanIdentity( plan_id=pid, parent_plan_id=None, root_plan_id=None, attempt=1, ), namespaced_name=NamespacedName(namespace=namespace, name=short_name), action_name=action_name, description=f"Test plan for {action_name}", definition_of_done="Tests pass", phase=phase, processing_state=ProcessingState.QUEUED, strategy_actor="local/strategist", execution_actor="local/executor", timestamps=PlanTimestamps( created_at=datetime.now(), updated_at=datetime.now(), ), created_by=None, tags=[], reusable=True, read_only=False, automation_profile=automation_profile, validation_summary=validation_summary, execution_environment=execution_environment, execution_env_priority=execution_env_priority, ) # ======================================================================== # Background # ======================================================================== @given("a fresh in-memory database for repo coverage boost") def step_fresh_db(context: Context) -> None: engine = create_engine("sqlite:///:memory:") Base.metadata.create_all(engine) context.rcb_engine = engine session = sessionmaker(bind=engine)() context.rcb_session = session context.rcb_session_factory = lambda: session context.rcb_error = None @given("an action repository for coverage boost") def step_action_repo(context: Context) -> None: context.rcb_action_repo = ActionRepository( session_factory=context.rcb_session_factory, ) @given("a lifecycle plan repository for coverage boost") def step_plan_repo(context: Context) -> None: context.rcb_plan_repo = LifecyclePlanRepository( session_factory=context.rcb_session_factory, ) @given("a resource type repository for coverage boost") def step_resource_type_repo(context: Context) -> None: context.rcb_rt_repo = ResourceTypeRepository( session_factory=context.rcb_session_factory, ) # ======================================================================== # ActionRepository.update - plain-string arg_type/requirement (L1080, L1083) # ======================================================================== @given('a valid action named "{name}" for coverage boost') def step_make_action(context: Context, name: str) -> None: context.rcb_action = _make_action(name=name) @given("the action is persisted for coverage boost") def step_persist_action(context: Context) -> None: context.rcb_action_repo.create(context.rcb_action) context.rcb_session.commit() @when("the action arguments are replaced with plain-string arg_type and requirement") def step_replace_args_strings(context: Context) -> None: """Build ActionArguments whose arg_type and requirement are plain strings (not enum members) so that ``hasattr(arg.arg_type, 'value')`` returns False in the repository update code, exercising lines 1080 and 1083. """ # We bypass pydantic validation with model_construct so that arg_type # and requirement remain plain strings (no .value attribute), hitting # the ``else`` branch in ActionRepository.update(). patched_arg = ActionArgument.model_construct( name="target_file", arg_type="string", # plain str, no .value requirement="required", # plain str, no .value description="Target file path", default_value=None, min_value=None, max_value=None, validation_pattern=None, ) context.rcb_action = context.rcb_action.model_copy( update={"arguments": [patched_arg]} ) @when("the action is updated for coverage boost") def step_update_action(context: Context) -> None: try: context.rcb_action = context.rcb_action.model_copy( update={"updated_at": datetime.now()} ) context.rcb_result = context.rcb_action_repo.update(context.rcb_action) context.rcb_session.commit() except Exception as exc: context.rcb_error = exc @then("the update should succeed for coverage boost") def step_verify_no_error(context: Context) -> None: assert context.rcb_error is None, f"Unexpected error: {context.rcb_error}" @then("the retrieved action should have the string-typed arguments") def step_verify_string_args(context: Context) -> None: fetched = context.rcb_action_repo.get_by_name( str(context.rcb_action.namespaced_name) ) assert fetched is not None, "Action not found after update" assert len(fetched.arguments) == 1 arg = fetched.arguments[0] assert arg.name == "target_file" # ======================================================================== # LifecyclePlanRepository.update - automation_profile (L1340) # ======================================================================== @given('a lifecycle plan linked to "{action_name}" with automation profile') def step_make_plan_with_auto_profile(context: Context, action_name: str) -> None: context.rcb_plan = _make_plan( action_name=action_name, ns_name=f"local/auto-prof-plan-{_next_ulid()[:8]}", automation_profile=AutomationProfileRef( profile_name="trusted", provenance=AutomationProfileProvenance.PLAN, ), ) @given("the lifecycle plan is persisted for coverage boost") def step_persist_plan(context: Context) -> None: context.rcb_plan_repo.create(context.rcb_plan) context.rcb_session.commit() @when("the plan automation profile is updated to a new profile ref") def step_update_auto_profile(context: Context) -> None: context.rcb_plan = context.rcb_plan.model_copy( update={ "automation_profile": AutomationProfileRef( profile_name="local/careful-auto", provenance=AutomationProfileProvenance.ACTION, ), "timestamps": context.rcb_plan.timestamps.model_copy( update={"updated_at": datetime.now()} ), } ) @when("the lifecycle plan is updated for coverage boost") def step_update_plan(context: Context) -> None: try: context.rcb_plan_repo.update(context.rcb_plan) context.rcb_session.commit() except Exception as exc: context.rcb_error = exc @then("the plan update should succeed for coverage boost") def step_verify_plan_update_ok(context: Context) -> None: assert context.rcb_error is None, f"Unexpected error: {context.rcb_error}" @then("the retrieved plan should have the automation profile set") def step_verify_auto_profile(context: Context) -> None: fetched = context.rcb_plan_repo.get(str(context.rcb_plan.identity.plan_id)) assert fetched is not None, "Plan not found" assert fetched.automation_profile is not None, ( "automation_profile should not be None" ) assert fetched.automation_profile.profile_name == "local/careful-auto" # ======================================================================== # LifecyclePlanRepository.update - validation_summary (L1357) # ======================================================================== @given('a lifecycle plan linked to "{action_name}" with validation summary') def step_make_plan_with_val_summary(context: Context, action_name: str) -> None: context.rcb_plan = _make_plan( action_name=action_name, ns_name=f"local/val-sum-plan-{_next_ulid()[:8]}", validation_summary={"total": 5, "passed": 4, "failed": 1}, ) @when("the plan validation summary is set to a non-None value") def step_set_val_summary(context: Context) -> None: context.rcb_plan = context.rcb_plan.model_copy( update={ "validation_summary": {"total": 10, "passed": 10, "failed": 0}, "timestamps": context.rcb_plan.timestamps.model_copy( update={"updated_at": datetime.now()} ), } ) @then("the retrieved plan should have the validation summary") def step_verify_val_summary(context: Context) -> None: fetched = context.rcb_plan_repo.get(str(context.rcb_plan.identity.plan_id)) assert fetched is not None, "Plan not found" assert fetched.validation_summary is not None, ( "validation_summary should not be None" ) assert fetched.validation_summary["total"] == 10 assert fetched.validation_summary["passed"] == 10 # ======================================================================== # LifecyclePlanRepository.update - execution_env_priority (L1367) # ======================================================================== @given('a lifecycle plan linked to "{action_name}" with env priority') def step_make_plan_with_env_priority(context: Context, action_name: str) -> None: context.rcb_plan = _make_plan( action_name=action_name, ns_name=f"local/env-pri-plan-{_next_ulid()[:8]}", execution_environment="host", execution_env_priority=ExecutionEnvPriority.FALLBACK, ) @when("the plan execution env priority is set") def step_set_env_priority(context: Context) -> None: context.rcb_plan = context.rcb_plan.model_copy( update={ "execution_environment": "container", "execution_env_priority": ExecutionEnvPriority.OVERRIDE, "timestamps": context.rcb_plan.timestamps.model_copy( update={"updated_at": datetime.now()} ), } ) @then("the retrieved plan should have execution env priority set") def step_verify_env_priority(context: Context) -> None: fetched = context.rcb_plan_repo.get(str(context.rcb_plan.identity.plan_id)) assert fetched is not None, "Plan not found" assert fetched.execution_env_priority is not None assert ( str(fetched.execution_env_priority) == "override" or fetched.execution_env_priority == ExecutionEnvPriority.OVERRIDE ) # ======================================================================== # LifecyclePlanRepository.list_all - DatabaseError (L1472-1473) # ======================================================================== @given("a lifecycle plan repository with a broken session factory") def step_broken_plan_repo(context: Context) -> None: def broken_factory() -> Session: mock_session = MagicMock(spec=Session) mock_session.query.side_effect = OperationalError( "database is locked", None, None ) return mock_session context.rcb_broken_plan_repo = LifecyclePlanRepository( session_factory=broken_factory, ) @when("list_all is called on the broken plan repository") def step_call_list_all_broken(context: Context) -> None: try: context.rcb_broken_plan_repo.list_all() except Exception as exc: context.rcb_error = exc @then('a DatabaseError mentioning "{text}" should be raised for coverage boost') def step_verify_db_error(context: Context, text: str) -> None: assert context.rcb_error is not None, "Expected DatabaseError, got no error" assert isinstance(context.rcb_error, DatabaseError), ( f"Expected DatabaseError, got {type(context.rcb_error).__name__}: {context.rcb_error}" ) assert text in str(context.rcb_error), ( f"Expected '{text}' in error message: {context.rcb_error}" ) # ======================================================================== # ResourceTypeRepository.create - namespaced name with "/" (L1730) # and string resource_kind / sandbox_strategy (L1734, L1737) # ======================================================================== def _make_resource_type_spec( name: str, description: str = "Test resource type", resource_kind: str | ResourceKind = "physical", sandbox_strategy: str | SandboxStrategy = "none", user_addable: bool = True, handler: str | None = None, cli_args: list[ResourceTypeArgument] | None = None, parent_types: list[str] | None = None, child_types: list[str] | None = None, auto_discovery: dict | None = None, capabilities: dict | None = None, equivalence: dict | None = None, ) -> SimpleNamespace: """Create a SimpleNamespace that looks like a ResourceTypeSpec for the repository, but allows plain strings for resource_kind / sandbox_strategy to hit the ``else`` branches. """ return SimpleNamespace( name=name, description=description, resource_kind=resource_kind, sandbox_strategy=sandbox_strategy, user_addable=user_addable, handler=handler, cli_args=cli_args or [], parent_types=parent_types or [], child_types=child_types or [], auto_discovery=auto_discovery, capabilities=capabilities, equivalence=equivalence, source=None, ) @given('a resource type spec with namespaced name "{name}" and string enums') def step_make_rt_namespaced(context: Context, name: str) -> None: context.rcb_rt_spec = _make_resource_type_spec( name=name, resource_kind="physical", # plain string, no .value sandbox_strategy="none", # plain string, no .value ) @when("the resource type is created for coverage boost") def step_create_rt(context: Context) -> None: try: context.rcb_rt_repo.create(context.rcb_rt_spec) context.rcb_session.commit() except Exception as exc: context.rcb_error = exc @then("the resource type creation should succeed for coverage boost") def step_verify_rt_created(context: Context) -> None: assert context.rcb_error is None, f"Unexpected error: {context.rcb_error}" @then('the retrieved resource type should have namespace "{expected_ns}"') def step_verify_rt_namespace(context: Context, expected_ns: str) -> None: fetched = context.rcb_rt_repo.get(context.rcb_rt_spec.name) assert fetched is not None, f"Resource type '{context.rcb_rt_spec.name}' not found" # The name includes namespace if "/" in context.rcb_rt_spec.name: actual_ns = context.rcb_rt_spec.name.split("/")[0] assert actual_ns == expected_ns, ( f"Expected namespace '{expected_ns}', got '{actual_ns}'" ) # ======================================================================== # ResourceTypeRepository.update - string enums + cli_args (L1857, L1862, L1867-1872) # ======================================================================== @given('a resource type spec with name "{name}" and string enums') def step_make_rt_for_update(context: Context, name: str) -> None: context.rcb_rt_spec = _make_resource_type_spec( name=name, resource_kind="physical", sandbox_strategy="none", ) @given("the resource type is persisted for coverage boost") def step_persist_rt(context: Context) -> None: context.rcb_rt_repo.create(context.rcb_rt_spec) context.rcb_session.commit() @when("the resource type is updated with new description and cli_args") def step_update_rt(context: Context) -> None: try: # Build an update spec with cli_args and plain-string enums context.rcb_rt_spec = _make_resource_type_spec( name=context.rcb_rt_spec.name, description="Updated description with CLI args", resource_kind="physical", # plain string, hits else on L1857 sandbox_strategy="copy_on_write", # plain string, hits else on L1862 cli_args=[ ResourceTypeArgument( name="target-path", type="string", required=True, description="Path to the target", default=None, validation_pattern=None, ), ResourceTypeArgument( name="verbose", type="boolean", required=False, description="Enable verbose output", default="false", validation_pattern=None, ), ], ) context.rcb_rt_repo.update(context.rcb_rt_spec) context.rcb_session.commit() except Exception as exc: context.rcb_error = exc @then("the resource type update should succeed for coverage boost") def step_verify_rt_updated(context: Context) -> None: assert context.rcb_error is None, f"Unexpected error: {context.rcb_error}" @then("the retrieved resource type should have updated cli_args") def step_verify_rt_cli_args(context: Context) -> None: fetched = context.rcb_rt_repo.get(context.rcb_rt_spec.name) assert fetched is not None, f"Resource type '{context.rcb_rt_spec.name}' not found" assert fetched.description == "Updated description with CLI args" assert len(fetched.cli_args) == 2 arg_names = [a.name for a in fetched.cli_args] assert "target-path" in arg_names assert "verbose" in arg_names # ======================================================================== # ResourceTypeRepository.create - builtin name without "/" (L1730 else branch) # ======================================================================== @given('a resource type spec with builtin name "{name}" and string enums') def step_make_rt_builtin(context: Context, name: str) -> None: context.rcb_rt_spec = _make_resource_type_spec( name=name, resource_kind="physical", sandbox_strategy="none", ) @then('the retrieved builtin resource type should have namespace "builtin"') def step_verify_rt_builtin_ns(context: Context) -> None: fetched = context.rcb_rt_repo.get(context.rcb_rt_spec.name) assert fetched is not None, f"Resource type '{context.rcb_rt_spec.name}' not found" # For builtin names (no "/"), the repo stores namespace="builtin" # The domain object doesn't expose the namespace column directly, # but we can verify via the database row from cleveragents.infrastructure.database.models import ResourceTypeModel row = ( context.rcb_session.query(ResourceTypeModel) .filter_by(name=context.rcb_rt_spec.name) .first() ) assert row is not None assert row.namespace == "builtin", ( f"Expected namespace 'builtin', got '{row.namespace}'" )