"""Step definitions for database_models_new_coverage.feature. Covers missed lines and branches in database models: - ToolModel.from_domain() with object (non-dict) access via getattr (line 1666-1667) - SessionModel.from_domain() with non-empty messages list (line 1920-1921) - SkillModel instantiation exercising short_name and description columns (lines 2131-2132) - LifecyclePlanModel.from_domain() arguments_order with missing keys (line 948 false branch) """ from __future__ import annotations import json from datetime import datetime from types import SimpleNamespace from behave import given, then, when from ulid import ULID from cleveragents.infrastructure.database.models import ( LifecyclePlanModel, SessionModel, SkillItemModel, SkillModel, ToolModel, ) # --------------------------------------------------------------------------- # Constants # --------------------------------------------------------------------------- NOW = datetime.now() NOW_ISO = NOW.isoformat() LATER_ISO = datetime(2025, 12, 31, 23, 59, 59).isoformat() # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _make_tool_object(**overrides): """Create a SimpleNamespace mimicking a domain tool object (non-dict).""" defaults = { "name": "local/my-tool", "description": "A test tool", "tool_type": "tool", "source": "builtin", "input_schema_json": None, "output_schema_json": None, "capability_json": None, "lifecycle_json": None, "code": None, "mcp_server": None, "mcp_tool_name": None, "agent_skill_path": None, "timeout": 300, "wraps": None, "transform": None, "mode": None, "argument_mapping_json": None, "created_at": NOW_ISO, "updated_at": NOW_ISO, "resource_bindings": [], } defaults.update(overrides) return SimpleNamespace(**defaults) def _make_session_message( *, message_id=None, role="user", content="hello", sequence=0, metadata=None, tool_call_id=None, ): """Create a SimpleNamespace mimicking a SessionMessage domain object.""" return SimpleNamespace( message_id=message_id or str(ULID()), session_id=None, role=SimpleNamespace(value=role), content=content, sequence=sequence, timestamp=NOW, metadata=metadata or {}, tool_call_id=tool_call_id, ) def _make_token_usage(input_tokens=10, output_tokens=20, estimated_cost=0.001): """Create a SimpleNamespace mimicking SessionTokenUsage.""" return SimpleNamespace( input_tokens=input_tokens, output_tokens=output_tokens, estimated_cost=estimated_cost, ) def _make_session_object(*, messages=None, session_id=None): """Create a SimpleNamespace mimicking a Session domain object.""" return SimpleNamespace( session_id=session_id or str(ULID()), name=None, actor_name="test-actor", namespace="local", linked_plan_ids=[], token_usage=_make_token_usage(), metadata={"test": True}, messages=messages or [], created_at=NOW, updated_at=NOW, ) def _make_plan_identity(): """Create a SimpleNamespace mimicking PlanIdentity.""" return SimpleNamespace( plan_id=str(ULID()), parent_plan_id=None, root_plan_id=None, attempt=1, ) def _make_plan_timestamps(): """Create a SimpleNamespace mimicking PlanTimestamps.""" return SimpleNamespace( created_at=NOW, updated_at=NOW, strategize_started_at=None, strategize_completed_at=None, execute_started_at=None, execute_completed_at=None, apply_started_at=None, applied_at=None, ) def _make_namespaced_name(full="local/test-plan"): """Create a SimpleNamespace mimicking NamespacedName.""" parts = full.split("/", 1) return SimpleNamespace( namespace=parts[0] if len(parts) > 1 else "local", __str__=lambda self=None, _f=full: _f, ) class _FakeNamespacedName: """Fake NamespacedName that has namespace and supports str().""" def __init__(self, full: str): parts = full.split("/", 1) self.namespace = parts[0] if len(parts) > 1 else "local" self._full = full def __str__(self) -> str: return self._full def _make_plan_object( *, arguments=None, arguments_order=None, action_name="local/cov-action" ): """Create a SimpleNamespace mimicking a Plan domain object.""" return SimpleNamespace( identity=_make_plan_identity(), namespaced_name=_FakeNamespacedName("local/test-plan"), action_name=action_name, description="Test plan for coverage", definition_of_done="All tests pass", phase=SimpleNamespace(value="action"), processing_state=SimpleNamespace(value="queued"), automation_profile=None, strategy_actor="local/strategy", execution_actor="local/executor", review_actor=None, apply_actor=None, estimation_actor=None, invariant_actor=None, execution_environment=None, execution_env_priority=None, project_links=[], invariants=[], arguments=arguments if arguments is not None else {}, arguments_order=arguments_order if arguments_order is not None else [], changeset_id=None, timestamps=_make_plan_timestamps(), error_message=None, error_details=None, created_by="tester", tags=[], reusable=True, read_only=False, effective_profile_snapshot="{}", ) # =========================================================================== # ToolModel.from_domain with object (non-dict) access # =========================================================================== @given("a domain tool object with attribute access") def step_tool_object_attr_access(context): """Create a non-dict domain tool object for ToolModel.from_domain().""" context.tool_obj = _make_tool_object( name="local/attr-tool", description="Tool via attribute access", tool_type="tool", source="inline", ) @when("I create a ToolModel from the domain object") def step_create_tool_model(context): """Call ToolModel.from_domain() with the non-dict object.""" context.tool_model = ToolModel.from_domain(context.tool_obj) @then("the ToolModel should have the correct name") def step_tool_model_name(context): assert context.tool_model.name == "local/attr-tool", ( f"Expected 'local/attr-tool', got '{context.tool_model.name}'" ) @then("the ToolModel should have the correct description") def step_tool_model_description(context): assert context.tool_model.description == "Tool via attribute access" @then("the ToolModel should have the correct tool_type") def step_tool_model_tool_type(context): assert context.tool_model.tool_type == "tool" @then("the ToolModel should have the correct source") def step_tool_model_source(context): assert context.tool_model.source == "inline" # -- Namespaced tool object -- @given("a domain tool object with a namespaced name attribute") def step_tool_namespaced_object(context): """Create a non-dict domain tool object with a namespace/short_name style name.""" context.ns_tool_obj = _make_tool_object( name="acme/code-lint", description="Namespaced tool", ) @when("I create a ToolModel from the namespaced domain object") def step_create_ns_tool_model(context): context.ns_tool_model = ToolModel.from_domain(context.ns_tool_obj) @then("the ToolModel namespace should be extracted correctly") def step_ns_tool_namespace(context): assert context.ns_tool_model.namespace == "acme", ( f"Expected 'acme', got '{context.ns_tool_model.namespace}'" ) @then("the ToolModel short_name should be extracted correctly") def step_ns_tool_short_name(context): assert context.ns_tool_model.short_name == "code-lint", ( f"Expected 'code-lint', got '{context.ns_tool_model.short_name}'" ) # -- Minimal tool object (defaults) -- @given("a domain tool object with minimal attributes") def step_tool_minimal_object(context): """Create a non-dict object with only name — other fields use getattr defaults.""" context.min_tool_obj = SimpleNamespace(name="simple-tool") @when("I create a ToolModel from the minimal domain object") def step_create_min_tool_model(context): context.min_tool_model = ToolModel.from_domain(context.min_tool_obj) @then("the ToolModel should use default values for missing attributes") def step_min_tool_defaults(context): m = context.min_tool_model assert m.name == "simple-tool" assert m.description == "" assert m.tool_type == "tool" assert m.source == "builtin" assert m.timeout == 300 # No namespace for non-namespaced name assert m.namespace == "" assert m.short_name == "simple-tool" # =========================================================================== # SessionModel.from_domain with non-empty messages # =========================================================================== @given("a domain session object with two messages") def step_session_two_messages(context): """Create a session domain object with two messages in the list.""" msg1 = _make_session_message( role="user", content="Hello world", sequence=0, ) msg2 = _make_session_message( role="assistant", content="Hi there", sequence=1, ) context.session_obj = _make_session_object(messages=[msg1, msg2]) context.msg1_id = msg1.message_id context.msg2_id = msg2.message_id @when("I create a SessionModel from the domain session") def step_create_session_model(context): context.session_model = SessionModel.from_domain(context.session_obj) @then("the SessionModel should have two message child models") def step_session_two_messages_count(context): assert len(context.session_model.messages_rel) == 2, ( f"Expected 2 messages, got {len(context.session_model.messages_rel)}" ) @then("the first message model should have the correct role and content") def step_first_message_role_content(context): msg = context.session_model.messages_rel[0] assert msg.role == "user" assert msg.content == "Hello world" @then("the second message model should have the correct sequence") def step_second_message_sequence(context): msg = context.session_model.messages_rel[1] assert msg.sequence == 1 # -- Single message -- @given("a domain session object with one message") def step_session_one_message(context): """Create a session domain object with a single message.""" msg = _make_session_message( role="user", content="Single message", sequence=0, ) context.single_msg_session = _make_session_object(messages=[msg]) context.single_msg_id = msg.message_id @when("I create a SessionModel from the single-message domain session") def step_create_single_msg_session_model(context): context.single_session_model = SessionModel.from_domain(context.single_msg_session) @then("the SessionModel should have exactly one message child model") def step_single_msg_count(context): assert len(context.single_session_model.messages_rel) == 1 @then("the message model should preserve the message_id") def step_single_msg_id_preserved(context): msg = context.single_session_model.messages_rel[0] assert msg.message_id == context.single_msg_id # =========================================================================== # SkillModel instantiation (short_name and description columns) # =========================================================================== @given("a SkillModel is created with short_name and description values") def step_create_skill_model(context): """Instantiate a SkillModel to exercise the short_name and description columns.""" context.skill_model = SkillModel( name="local/code-tools", namespace="local", short_name="code-tools", description="A collection of code-related tools", version="1.0.0", metadata_json=json.dumps({"overrides": {}}), created_at=NOW_ISO, updated_at=NOW_ISO, ) @then("the SkillModel short_name should match the provided value") def step_skill_short_name(context): assert context.skill_model.short_name == "code-tools", ( f"Expected 'code-tools', got '{context.skill_model.short_name}'" ) @then("the SkillModel description should match the provided value") def step_skill_description(context): assert context.skill_model.description == "A collection of code-related tools" @then("the SkillModel namespace should be set correctly") def step_skill_namespace(context): assert context.skill_model.namespace == "local" # -- SkillModel with child SkillItemModel -- @given("a SkillModel is created with a child SkillItemModel") def step_create_skill_with_item(context): """Instantiate a SkillModel with a SkillItemModel child record.""" context.skill_with_item = SkillModel( name="local/data-tools", namespace="local", short_name="data-tools", description="Data processing tools", version="2.0.0", metadata_json=None, created_at=NOW_ISO, updated_at=NOW_ISO, ) item = SkillItemModel( skill_name="local/data-tools", item_type="tool_ref", item_name="local/csv-parser", item_config=None, item_order=0, created_at=NOW_ISO, ) context.skill_with_item.items_rel.append(item) @then("the SkillModel should have one item in items_rel") def step_skill_item_count(context): assert len(context.skill_with_item.items_rel) == 1 @then("the SkillItemModel should have the correct item_type and item_name") def step_skill_item_fields(context): item = context.skill_with_item.items_rel[0] assert item.item_type == "tool_ref" assert item.item_name == "local/csv-parser" # =========================================================================== # LifecyclePlanModel.from_domain: arguments_order with key NOT in arguments_dict # =========================================================================== @given("a plan domain object with arguments_order containing a missing key") def step_plan_with_missing_arg_key(context): """Create a plan where arguments_order has a key absent from arguments_dict. arguments_dict has "file_path" but arguments_order lists both "file_path" and "nonexistent_key". The false branch of `if arg_name in arguments_dict` at line 949 should be hit for "nonexistent_key". """ context.plan_obj = _make_plan_object( arguments={"file_path": "/tmp/test.txt"}, arguments_order=["file_path", "nonexistent_key"], ) @when("I convert the plan domain object to a database model") def step_convert_plan_to_model(context): context.plan_model = LifecyclePlanModel.from_domain( context.plan_obj, action_name="local/cov-action" ) @then( "the database plan model should only have arguments for keys present in arguments_dict" ) def step_plan_args_only_present(context): args = context.plan_model.arguments_rel arg_names = [a.name for a in args] assert "file_path" in arg_names, ( f"Expected 'file_path' in arguments, got {arg_names}" ) assert len(args) == 1, f"Expected 1 argument, got {len(args)}" @then("the missing key should not appear in the argument child records") def step_plan_no_missing_key(context): arg_names = [a.name for a in context.plan_model.arguments_rel] assert "nonexistent_key" not in arg_names, ( f"'nonexistent_key' should not be in arguments: {arg_names}" ) # -- Mixed present/missing keys -- @given("a plan domain object with three ordered keys but only two in arguments_dict") def step_plan_mixed_keys(context): """Create a plan with 3 keys in arguments_order but only 2 in arguments_dict. This exercises the false branch multiple times and confirms position tracking. """ context.mixed_plan_obj = _make_plan_object( arguments={ "input_file": "data.csv", "output_dir": "/tmp/out", }, arguments_order=["input_file", "ghost_key", "output_dir"], ) @when("I convert the mixed-arguments plan to a database model") def step_convert_mixed_plan(context): context.mixed_plan_model = LifecyclePlanModel.from_domain( context.mixed_plan_obj, action_name="local/cov-action" ) @then("the database plan model should have exactly two argument child records") def step_mixed_plan_two_args(context): args = context.mixed_plan_model.arguments_rel assert len(args) == 2, f"Expected 2 arguments, got {len(args)}" @then("the argument positions should reflect their order in arguments_order") def step_mixed_plan_positions(context): args = sorted(context.mixed_plan_model.arguments_rel, key=lambda a: a.position) # "input_file" is at index 0 in arguments_order assert args[0].name == "input_file" assert args[0].position == 0 # "output_dir" is at index 2 in arguments_order (ghost_key at index 1 is skipped) assert args[1].name == "output_dir" assert args[1].position == 2