"""Step definitions for database models uncovered-branch coverage tests. Targets four specific coverage gaps in models.py: 1. LifecyclePlanModel.from_domain with empty arguments (L948 loop zero-iter) 2. ToolModel.from_domain with a plain dict (L1665-1666 dict branch of _get) 3. SessionModel.from_domain with non-empty messages (L1920-1921) 4. SkillModel column definitions short_name / description (L2131-2132) """ from __future__ import annotations from datetime import UTC, datetime from typing import Any from behave import given, then, when from behave.runner import Context from ulid import ULID # ═══════════════════════════════════════════════════════════════════════════ # Helpers # ═══════════════════════════════════════════════════════════════════════════ _UTC = UTC def _ulid() -> str: return str(ULID()) def _make_minimal_plan( *, arguments: dict[str, Any] | None = None, arguments_order: list[str] | None = None, ) -> Any: """Build a minimal Plan domain object suitable for from_domain. Uses real domain classes so the ORM helper exercises the same paths as production code. """ from cleveragents.domain.models.core.plan import ( NamespacedName, Plan, PlanIdentity, PlanPhase, PlanTimestamps, ProcessingState, ) return Plan( identity=PlanIdentity(plan_id=_ulid()), namespaced_name=NamespacedName(namespace="local", name="test-plan"), description="A test plan for branch coverage", action_name="local/test-action", phase=PlanPhase.STRATEGIZE, processing_state=ProcessingState.QUEUED, strategy_actor="local/strategy", execution_actor="local/executor", timestamps=PlanTimestamps(), arguments=arguments if arguments is not None else {}, arguments_order=arguments_order if arguments_order is not None else [], ) # ═══════════════════════════════════════════════════════════════════════════ # Scenario 1 - PlanModel.from_domain with empty arguments # ═══════════════════════════════════════════════════════════════════════════ @given("db models branch a Plan domain object with empty arguments") def step_given_plan_empty_args(context: Context) -> None: """Create a Plan domain object where arguments and arguments_order are empty.""" context.dbm_plan = _make_minimal_plan(arguments={}, arguments_order=[]) @when("db models branch the plan is converted via LifecyclePlanModel from_domain") def step_when_plan_from_domain(context: Context) -> None: from cleveragents.infrastructure.database.models import LifecyclePlanModel context.dbm_plan_model = LifecyclePlanModel.from_domain(context.dbm_plan) @then("db models branch the resulting plan model has no argument rows") def step_then_plan_no_args(context: Context) -> None: model = context.dbm_plan_model assert model.arguments_rel is not None, "arguments_rel should be a list" assert len(model.arguments_rel) == 0, ( f"Expected 0 argument rows, got {len(model.arguments_rel)}" ) # ═══════════════════════════════════════════════════════════════════════════ # Scenario 2 - ToolModel.from_domain with a plain dict (dict branch of _get) # ═══════════════════════════════════════════════════════════════════════════ @given("db models branch a plain dict describing a tool") def step_given_tool_dict(context: Context) -> None: """Create a plain dict with the fields that ToolModel.from_domain reads.""" context.dbm_tool_dict: dict[str, Any] = { "name": "local/coverage-tool", "description": "A tool created from a dict for branch coverage", "tool_type": "tool", "source": "builtin", "timeout": 60, } @when("db models branch the dict is converted via ToolModel from_domain") def step_when_tool_dict_from_domain(context: Context) -> None: from cleveragents.infrastructure.database.models import ToolModel context.dbm_tool_model = ToolModel.from_domain(context.dbm_tool_dict) @then("db models branch the resulting tool model has the expected name and description") def step_then_tool_model_fields(context: Context) -> None: model = context.dbm_tool_model assert model.name == "local/coverage-tool", f"name={model.name}" assert model.short_name == "coverage-tool", f"short_name={model.short_name}" assert model.namespace == "local", f"namespace={model.namespace}" assert "dict for branch coverage" in model.description, ( f"description={model.description}" ) assert model.tool_type == "tool", f"tool_type={model.tool_type}" assert model.source == "builtin", f"source={model.source}" assert model.timeout == 60, f"timeout={model.timeout}" # ── Scenario 2b - dict tool with resource bindings ──────────────────────── @given("db models branch a plain dict describing a tool with resource bindings") def step_given_tool_dict_with_bindings(context: Context) -> None: context.dbm_tool_dict_bindings: dict[str, Any] = { "name": "local/binding-tool", "description": "Tool with bindings from dict", "tool_type": "tool", "source": "builtin", "timeout": 120, "resource_bindings": [ { "slot_name": "repo", "resource_type": "git-checkout", "access_mode": "read_write", "binding_mode": "static", "static_resource": "my-repo", "required": True, "description": "The repository", }, ], } @when("db models branch the binding dict is converted via ToolModel from_domain") def step_when_tool_dict_bindings_from_domain(context: Context) -> None: from cleveragents.infrastructure.database.models import ToolModel context.dbm_tool_model_bindings = ToolModel.from_domain( context.dbm_tool_dict_bindings, ) @then("db models branch the resulting tool model has resource binding rows") def step_then_tool_model_bindings(context: Context) -> None: model = context.dbm_tool_model_bindings assert len(model.resource_bindings_rel) == 1, ( f"Expected 1 binding, got {len(model.resource_bindings_rel)}" ) binding = model.resource_bindings_rel[0] assert binding.slot_name == "repo", f"slot_name={binding.slot_name}" assert binding.access_mode == "read_write", f"access_mode={binding.access_mode}" assert binding.binding_mode == "static", f"binding_mode={binding.binding_mode}" assert binding.static_resource == "my-repo", ( f"static_resource={binding.static_resource}" ) # ═══════════════════════════════════════════════════════════════════════════ # Scenario 3 - SessionModel.from_domain with non-empty messages # ═══════════════════════════════════════════════════════════════════════════ @given("db models branch a Session domain object with two messages") def step_given_session_with_messages(context: Context) -> None: """Create a Session domain object that has two messages.""" from cleveragents.domain.models.core.session import ( MessageRole, Session, SessionMessage, SessionTokenUsage, ) now = datetime.now(tz=_UTC) msg1 = SessionMessage( message_id=_ulid(), role=MessageRole.USER, content="Hello from user", sequence=0, timestamp=now, ) msg2 = SessionMessage( message_id=_ulid(), role=MessageRole.ASSISTANT, content="Hello from assistant", sequence=1, timestamp=now, ) context.dbm_session = Session( session_id=_ulid(), actor_name=None, namespace="local", messages=[msg1, msg2], linked_plan_ids=[], token_usage=SessionTokenUsage( input_tokens=10, output_tokens=20, estimated_cost=0.001, ), metadata={"source": "test"}, created_at=now, updated_at=now, ) @when("db models branch the session is converted via SessionModel from_domain") def step_when_session_from_domain(context: Context) -> None: from cleveragents.infrastructure.database.models import SessionModel context.dbm_session_model = SessionModel.from_domain(context.dbm_session) @then("db models branch the resulting session model has two message children") def step_then_session_has_messages(context: Context) -> None: model = context.dbm_session_model assert model.messages_rel is not None, "messages_rel should be a list" assert len(model.messages_rel) == 2, ( f"Expected 2 message children, got {len(model.messages_rel)}" ) roles = [m.role for m in model.messages_rel] assert "user" in roles, f"Expected 'user' role in {roles}" assert "assistant" in roles, f"Expected 'assistant' role in {roles}" # Verify content was serialised contents = [m.content for m in model.messages_rel] assert "Hello from user" in contents, f"Missing user content in {contents}" assert "Hello from assistant" in contents, ( f"Missing assistant content in {contents}" ) # ═══════════════════════════════════════════════════════════════════════════ # Scenario 4 - SkillModel.from_domain populates short_name and description # ═══════════════════════════════════════════════════════════════════════════ @given("db models branch a Skill domain dict with name and description") def step_given_skill_dict(context: Context) -> None: context.dbm_skill_dict: dict[str, Any] = { "name": "local/cov-skill", "description": "Skill for coverage test", "tool_refs": ["local/some-tool"], } @when("db models branch the skill dict is converted via SkillModel from_domain") def step_when_skill_from_domain(context: Context) -> None: from cleveragents.infrastructure.database.models import SkillModel context.dbm_skill_model = SkillModel.from_domain(context.dbm_skill_dict) @then( "db models branch the resulting skill model has correct short_name and description" ) def step_then_skill_model_columns(context: Context) -> None: model = context.dbm_skill_model assert model.short_name == "cov-skill", f"short_name={model.short_name}" assert model.description == "Skill for coverage test", ( f"description={model.description}" ) assert model.namespace == "local", f"namespace={model.namespace}" assert model.name == "local/cov-skill", f"name={model.name}"