"""Step definitions for database_models_coverage_r2.feature. Targets uncovered lines in src/cleveragents/infrastructure/database/models.py identified from build/coverage.xml (round 2). """ from __future__ import annotations import json from datetime import UTC, datetime from types import SimpleNamespace from typing import Any from behave import given, then, when # type: ignore[import-untyped] from cleveragents.domain.models.core.plan import ( AutomationProfileProvenance, AutomationProfileRef, ) # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- NOW = datetime(2025, 6, 1, 12, 0, 0, tzinfo=UTC) NOW_ISO = NOW.isoformat() def _make_ulid(seed: int = 1) -> str: """Return a fake but valid-length 26-char ULID.""" return f"01ABCDEFGH{seed:016d}" def _minimal_plan_timestamps() -> SimpleNamespace: 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 _minimal_plan_identity(plan_id: str | None = None) -> SimpleNamespace: return SimpleNamespace( plan_id=plan_id or _make_ulid(1), parent_plan_id=None, root_plan_id=None, attempt=1, ) def _minimal_namespaced_name() -> SimpleNamespace: return SimpleNamespace( server=None, namespace="local", name="test-action", __str__=lambda self: f"{self.namespace}/{self.name}", ) # --------------------------------------------------------------------------- # LifecycleActionModel — safety_profile round-trip (lines 376, 427) # --------------------------------------------------------------------------- @given("a LifecycleActionModel with safety_profile_json populated") def step_action_model_with_safety(context: Any) -> None: from cleveragents.infrastructure.database.models import ( LifecycleActionModel, ) safety_data = { "require_sandbox": True, "require_checkpoints": False, "allow_unsafe_tools": False, } model = LifecycleActionModel( namespaced_name="local/test-action", namespace="local", name="test-action", description="desc", definition_of_done="dod", strategy_actor="strat", execution_actor="exec", reusable=True, read_only=False, state="available", tags_json="[]", created_at=NOW_ISO, updated_at=NOW_ISO, safety_profile_json=json.dumps(safety_data), ) # Ensure relationship lists exist model.arguments_rel = [] model.invariants_rel = [] context.action_model = model @when("I convert the action model to domain via to_domain") def step_action_to_domain(context: Any) -> None: context.domain_action = context.action_model.to_domain() @then("the domain action has a non-null safety_profile object") def step_check_safety_profile_not_none(context: Any) -> None: sp = context.domain_action.safety_profile assert sp is not None, "safety_profile should be deserialized" assert sp.require_sandbox is True @given("an Action domain object with a SafetyProfile attached") def step_action_domain_with_safety(context: Any) -> None: from cleveragents.domain.models.core.safety_profile import SafetyProfile context.domain_action_input = SimpleNamespace( namespaced_name=_minimal_namespaced_name(), description="desc", long_description=None, definition_of_done="dod", strategy_actor="strat", execution_actor="exec", review_actor=None, apply_actor=None, estimation_actor=None, invariant_actor=None, automation_profile=None, safety_profile=SafetyProfile( require_sandbox=True, require_checkpoints=False, allow_unsafe_tools=True, ), reusable=True, read_only=False, inputs_schema=None, state=SimpleNamespace(value="available"), created_by=None, tags=[], arguments=[], invariants=[], created_at=NOW, updated_at=NOW, ) @when("I convert the action to a model via from_domain") def step_action_from_domain(context: Any) -> None: from cleveragents.infrastructure.database.models import ( LifecycleActionModel, ) context.action_model_out = LifecycleActionModel.from_domain( context.domain_action_input, ) @then("the model safety_profile_json contains serialized SafetyProfile data") def step_check_safety_json(context: Any) -> None: raw = context.action_model_out.safety_profile_json assert raw is not None, "safety_profile_json must not be None" parsed = json.loads(raw) assert parsed["require_sandbox"] is True assert parsed["allow_unsafe_tools"] is True # --------------------------------------------------------------------------- # LifecycleActionModel — string arg_type/requirement (lines 466, 469) # --------------------------------------------------------------------------- @given( "an Action domain object whose arguments have plain-string arg_type and requirement" ) def step_action_plain_string_args(context: Any) -> None: arg = SimpleNamespace( name="my_arg", arg_type="string", # plain string, no .value requirement="required", # plain string, no .value description="a test arg", default_value=None, min_value=None, max_value=None, validation_pattern=None, ) context.domain_action_input = SimpleNamespace( namespaced_name=_minimal_namespaced_name(), description="desc", long_description=None, definition_of_done="dod", strategy_actor="strat", execution_actor="exec", review_actor=None, apply_actor=None, estimation_actor=None, invariant_actor=None, automation_profile=None, safety_profile=None, reusable=True, read_only=False, inputs_schema=None, state=SimpleNamespace(value="available"), created_by=None, tags=[], arguments=[arg], invariants=[], created_at=NOW, updated_at=NOW, ) @when("I convert that action to a model via from_domain") def step_that_action_from_domain(context: Any) -> None: from cleveragents.infrastructure.database.models import ( LifecycleActionModel, ) context.action_model_out = LifecycleActionModel.from_domain( context.domain_action_input, ) @then("the model argument rows use the plain-string values directly") def step_check_arg_strings(context: Any) -> None: args = context.action_model_out.arguments_rel assert len(args) == 1 assert args[0].arg_type == "string" assert args[0].requirement == "required" # --------------------------------------------------------------------------- # LifecyclePlanModel — string processing_state (line 882) # --------------------------------------------------------------------------- def _base_plan_ns( *, processing_state: Any = "queued", automation_profile: AutomationProfileRef | None = None, namespaced_name: Any = None, error_details: Any = None, ) -> SimpleNamespace: """Build a minimal Plan-like namespace for from_domain tests.""" ns_name = namespaced_name or _minimal_namespaced_name() return SimpleNamespace( identity=_minimal_plan_identity(), namespaced_name=ns_name, action_name="local/test-action", description="plan desc", definition_of_done="dod", phase=SimpleNamespace(value="action"), processing_state=processing_state, automation_profile=automation_profile, strategy_actor=None, execution_actor=None, review_actor=None, apply_actor=None, estimation_actor=None, invariant_actor=None, execution_environment=None, execution_env_priority=None, project_links=[], invariants=[], arguments={}, arguments_order=[], changeset_id=None, sandbox_refs=[], validation_summary=None, decision_root_id=None, timestamps=_minimal_plan_timestamps(), error_message=None, error_details=error_details, created_by=None, tags=[], reusable=True, read_only=False, effective_profile_snapshot="{}", ) @given("a Plan domain object whose processing_state is a plain string") def step_plan_string_state(context: Any) -> None: context.plan_domain = _base_plan_ns(processing_state="complete") @when("I convert the plan to a model via from_domain") def step_plan_from_domain(context: Any) -> None: from cleveragents.infrastructure.database.models import ( LifecyclePlanModel, ) context.plan_model_out = LifecyclePlanModel.from_domain(context.plan_domain) @then("the model processing_state equals the plain string value") def step_check_plan_state(context: Any) -> None: assert context.plan_model_out.processing_state == "complete" # --------------------------------------------------------------------------- # LifecyclePlanModel — automation_profile (line 890) # --------------------------------------------------------------------------- @given("a Plan domain object with a non-null automation_profile") def step_plan_with_automation_profile(context: Any) -> None: profile = AutomationProfileRef( profile_name="strict", provenance=AutomationProfileProvenance.ACTION, ) context.plan_domain = _base_plan_ns(automation_profile=profile) @then("the model automation_profile column contains JSON with profile_name") def step_check_automation_profile_json(context: Any) -> None: raw = context.plan_model_out.automation_profile assert raw is not None parsed = json.loads(raw) assert parsed["profile_name"] == "strict" # --------------------------------------------------------------------------- # LifecyclePlanModel — fallback namespace (line 906) # --------------------------------------------------------------------------- @given("a Plan domain object whose namespaced_name has no namespace attribute") def step_plan_no_namespace(context: Any) -> None: # Use a plain string that has no .namespace attribute context.plan_domain = _base_plan_ns(namespaced_name="local/test-plan") @then('the model namespace equals "local"') def step_check_namespace_local(context: Any) -> None: assert context.plan_model_out.namespace == "local" # --------------------------------------------------------------------------- # LifecyclePlanModel — error_details (line 923) # --------------------------------------------------------------------------- @given("a Plan domain object with non-null error_details") def step_plan_with_error_details(context: Any) -> None: context.plan_domain = _base_plan_ns( error_details={"code": "E001", "msg": "something failed"}, ) @then("the model error_details_json is a JSON string of the details dict") def step_check_error_details_json(context: Any) -> None: raw = context.plan_model_out.error_details_json assert raw is not None parsed = json.loads(raw) assert parsed["code"] == "E001" assert parsed["msg"] == "something failed" # --------------------------------------------------------------------------- # SkillModel — include with overrides (line 2327) # --------------------------------------------------------------------------- @given("a Skill domain object with an include that has overrides") def step_skill_with_include_overrides(context: Any) -> None: include = SimpleNamespace( name="other-skill/base", overrides={"timeout": 120}, ) context.skill_domain = SimpleNamespace( name="local/my-skill", description="a skill", tool_refs=[], includes=[include], anonymous_tools=[], mcp_servers=[], agent_skills=[], overrides={}, version=None, ) @when("I convert the skill to a model via from_domain") def step_skill_from_domain(context: Any) -> None: from cleveragents.infrastructure.database.models import SkillModel context.skill_model_out = SkillModel.from_domain(context.skill_domain) @then("the include item_config contains the serialized overrides") def step_check_include_overrides(context: Any) -> None: items = context.skill_model_out.items_rel include_items = [i for i in items if i.item_type == "include"] assert len(include_items) == 1 config = json.loads(include_items[0].item_config) assert config["overrides"]["timeout"] == 120 # --------------------------------------------------------------------------- # DecisionModel — string decision_type (line 2698) # --------------------------------------------------------------------------- @given("a Decision domain object whose decision_type is a plain string") def step_decision_string_type(context: Any) -> None: from cleveragents.domain.models.core.decision import ( ContextSnapshot, ) context.decision_domain = SimpleNamespace( decision_id=_make_ulid(10), plan_id=_make_ulid(11), parent_decision_id=None, sequence_number=1, decision_type="strategy_choice", # plain string, no .value question="Which strategy?", chosen_option="Option A", alternatives_considered=["Option B"], confidence_score=0.9, context_snapshot=ContextSnapshot( hot_context_hash="abc", hot_context_ref="ref1", relevant_resources=[], actor_state_ref="actor1", ), rationale="because", actor_reasoning=None, downstream_decision_ids=[], downstream_plan_ids=[], artifacts_produced=[], created_at=NOW, is_correction=False, corrects_decision_id=None, correction_reason=None, superseded_by=None, ) @when("I convert the decision to a model via from_domain") def step_decision_from_domain(context: Any) -> None: from cleveragents.infrastructure.database.models import DecisionModel context.decision_model_out = DecisionModel.from_domain( context.decision_domain, ) @then("the model decision_type equals the plain string") def step_check_decision_type_string(context: Any) -> None: assert context.decision_model_out.decision_type == "strategy_choice" # --------------------------------------------------------------------------- # CheckpointModel — to_domain (lines 2823-2852) # --------------------------------------------------------------------------- @given("a CheckpointModel with metadata_json containing reason and phase") def step_checkpoint_model_with_meta(context: Any) -> None: from cleveragents.infrastructure.database.models import CheckpointModel context.checkpoint_model = CheckpointModel( checkpoint_id=_make_ulid(20), plan_id=_make_ulid(21), sandbox_ref="abc123commit", decision_id=None, checkpoint_type="pre_write", resource_id=None, filesystem_path="checkpoints/cp1", size_bytes=1024, created_at=NOW_ISO, metadata_json=json.dumps( { "reason": "before write", "source_tool": "file_write", "phase": "execute", } ), ) @when("I convert the checkpoint model to domain via to_domain") def step_checkpoint_to_domain(context: Any) -> None: context.domain_checkpoint = context.checkpoint_model.to_domain() @then("the domain checkpoint has the correct metadata fields") def step_check_checkpoint_metadata(context: Any) -> None: cp = context.domain_checkpoint assert cp.checkpoint_id == _make_ulid(20) assert cp.plan_id == _make_ulid(21) assert cp.sandbox_ref == "abc123commit" assert cp.checkpoint_type == "pre_write" assert cp.filesystem_path == "checkpoints/cp1" assert cp.size_bytes == 1024 assert cp.metadata.reason == "before write" assert cp.metadata.source_tool == "file_write" assert cp.metadata.phase == "execute" @given("a CheckpointModel with invalid JSON in metadata_json") def step_checkpoint_bad_json(context: Any) -> None: from cleveragents.infrastructure.database.models import CheckpointModel context.checkpoint_model = CheckpointModel( checkpoint_id=_make_ulid(30), plan_id=_make_ulid(31), sandbox_ref="deadbeef", checkpoint_type="manual", filesystem_path="", size_bytes=None, created_at=NOW_ISO, metadata_json="NOT VALID JSON {{{", ) @then("the domain checkpoint metadata is empty defaults") def step_check_empty_metadata(context: Any) -> None: cp = context.domain_checkpoint assert cp.metadata.reason == "" assert cp.metadata.source_tool == "" assert cp.metadata.phase == "" # --------------------------------------------------------------------------- # CheckpointModel — from_domain (lines 2865-2879) # --------------------------------------------------------------------------- @given("a Checkpoint domain object with full metadata") def step_checkpoint_domain(context: Any) -> None: from cleveragents.domain.models.core.checkpoint import ( Checkpoint, CheckpointMetadata, ) context.checkpoint_domain = Checkpoint( checkpoint_id=_make_ulid(40), plan_id=_make_ulid(41), sandbox_ref="commitsha", decision_id=_make_ulid(42), checkpoint_type="post_step", resource_id=_make_ulid(43), filesystem_path="cp/path", size_bytes=2048, created_at=NOW, metadata=CheckpointMetadata( reason="post step save", source_tool="apply_tool", phase="apply", ), ) @when("I convert the checkpoint to a model via from_domain") def step_checkpoint_from_domain(context: Any) -> None: from cleveragents.infrastructure.database.models import CheckpointModel context.checkpoint_model_out = CheckpointModel.from_domain( context.checkpoint_domain, ) @then("the model has correct checkpoint_id plan_id and metadata_json") def step_check_checkpoint_model_fields(context: Any) -> None: m = context.checkpoint_model_out assert m.checkpoint_id == _make_ulid(40) assert m.plan_id == _make_ulid(41) assert m.sandbox_ref == "commitsha" assert m.decision_id == _make_ulid(42) assert m.checkpoint_type == "post_step" assert m.resource_id == _make_ulid(43) assert m.filesystem_path == "cp/path" assert m.size_bytes == 2048 assert m.created_at == NOW_ISO raw = m.metadata_json assert raw is not None parsed = json.loads(raw) assert parsed["reason"] == "post step save" assert parsed["source_tool"] == "apply_tool" assert parsed["phase"] == "apply" # --------------------------------------------------------------------------- # get_session (lines 2907-2908) # --------------------------------------------------------------------------- @given("an in-memory SQLAlchemy engine") def step_create_engine(context: Any) -> None: from cleveragents.infrastructure.database.models import init_database context.engine = init_database("sqlite:///:memory:") @when("I call get_session with that engine") def step_call_get_session(context: Any) -> None: from cleveragents.infrastructure.database.models import get_session context.session = get_session(context.engine) @then("I receive a valid SQLAlchemy session object") def step_check_session(context: Any) -> None: from sqlalchemy.orm import Session assert isinstance(context.session, Session) # Verify it can execute a simple query result = context.session.execute(__import__("sqlalchemy").text("SELECT 1")) assert result.scalar() == 1 context.session.close()