"""Helper utilities for v3 database lifecycle model Robot integration tests. Covers the integration between: - LifecycleActionModel (database/models.py) <-> Action domain model (action.py) - LifecyclePlanModel (database/models.py) <-> Plan domain model (plan.py) - Round-trip: domain -> SQLAlchemy -> domain """ from __future__ import annotations import sys import tempfile from datetime import UTC, datetime from pathlib import Path from sqlalchemy.orm import Session from cleveragents.domain.models.core.action import Action, ActionArgument, ActionState from cleveragents.domain.models.core.plan import ( NamespacedName, Plan, PlanIdentity, PlanPhase, PlanTimestamps, ProcessingState, ProjectLink, ) from cleveragents.infrastructure.database.models import ( LifecycleActionModel, LifecyclePlanModel, init_database, ) def _create_temp_db() -> tuple[Session, str]: """Create a temporary SQLite database for testing.""" tmp = tempfile.mktemp(suffix=".db") db_url = f"sqlite:///{tmp}" engine = init_database(db_url) session = Session(bind=engine) return session, tmp def _action_round_trip() -> None: """Integration test: Action domain -> LifecycleActionModel -> Action domain.""" session, tmp = _create_temp_db() try: # Create domain Action action = Action( namespaced_name=NamespacedName.parse("local/test-action"), description="Test action", long_description="A detailed test action", definition_of_done="All tests pass", strategy_actor="local/planner", execution_actor="local/executor", review_actor="local/reviewer", arguments=[ ActionArgument.parse("target:str:required:The target directory"), ], reusable=True, read_only=False, state=ActionState.AVAILABLE, created_at=datetime(2026, 1, 1, tzinfo=UTC), updated_at=datetime(2026, 1, 2, tzinfo=UTC), created_by="test-user", tags=["test", "ci"], ) # Convert to SQLAlchemy model and persist db_model = LifecycleActionModel.from_domain(action) session.add(db_model) session.commit() # Read back and convert to domain loaded = ( session.query(LifecycleActionModel) .filter_by(namespaced_name="local/test-action") .one() ) restored = loaded.to_domain() # Verify round-trip fidelity assert str(restored.namespaced_name) == str(action.namespaced_name) assert str(restored.namespaced_name) == str(action.namespaced_name) assert restored.description == action.description assert restored.long_description == action.long_description assert restored.definition_of_done == action.definition_of_done assert restored.strategy_actor == action.strategy_actor assert restored.execution_actor == action.execution_actor assert restored.review_actor == action.review_actor assert restored.state == action.state assert restored.reusable == action.reusable assert restored.read_only == action.read_only assert restored.created_by == action.created_by assert len(restored.tags) == 2 assert len(restored.arguments) == 1 assert restored.arguments[0].name == "target" print("action-round-trip-ok") finally: session.close() Path(tmp).unlink(missing_ok=True) def _plan_round_trip() -> None: """Integration test: Plan domain -> LifecyclePlanModel -> Plan domain.""" session, tmp = _create_temp_db() try: # First create an action (required for FK) action = Action( namespaced_name=NamespacedName.parse("local/plan-test-action"), description="Plan test action", definition_of_done="Done", strategy_actor="local/s", execution_actor="local/e", state=ActionState.AVAILABLE, created_at=datetime(2026, 1, 1, tzinfo=UTC), updated_at=datetime(2026, 1, 1, tzinfo=UTC), ) action_db = LifecycleActionModel.from_domain(action) session.add(action_db) session.commit() # Create domain Plan plan = Plan( identity=PlanIdentity( plan_id="01HV000000000000000000CCCC", ), action_name="local/plan-test-action", namespaced_name=NamespacedName.parse("local/plan-test-action"), description="Integration test plan", definition_of_done="All assertions pass", phase=PlanPhase.STRATEGIZE, processing_state=ProcessingState.QUEUED, strategy_actor="local/s", execution_actor="local/e", project_links=[ ProjectLink(project_name="proj-001"), ProjectLink(project_name="proj-002"), ], timestamps=PlanTimestamps( created_at=datetime(2026, 2, 1, tzinfo=UTC), updated_at=datetime(2026, 2, 1, tzinfo=UTC), ), created_by="integration-test", tags=["ci", "test"], ) # Convert to SQLAlchemy model and persist plan_db = LifecyclePlanModel.from_domain( plan, action_name="local/plan-test-action" ) session.add(plan_db) session.commit() # Read back and convert to domain loaded = ( session.query(LifecyclePlanModel) .filter_by(plan_id="01HV000000000000000000CCCC") .one() ) restored = loaded.to_domain() # Verify round-trip fidelity assert restored.identity.plan_id == plan.identity.plan_id assert str(restored.namespaced_name) == str(plan.namespaced_name) assert restored.description == plan.description assert restored.definition_of_done == plan.definition_of_done assert restored.phase == plan.phase assert restored.state == plan.state assert restored.strategy_actor == plan.strategy_actor assert restored.execution_actor == plan.execution_actor assert restored.created_by == plan.created_by assert len(restored.project_links) == 2 assert len(restored.tags) == 2 print("plan-round-trip-ok") finally: session.close() Path(tmp).unlink(missing_ok=True) def _plan_hierarchy_persistence() -> None: """Integration test: parent/child plan relationships persist correctly.""" session, tmp = _create_temp_db() try: # Create action action = Action( namespaced_name=NamespacedName.parse("local/hierarchy-action"), description="Hierarchy test action", definition_of_done="Done", strategy_actor="local/s", execution_actor="local/e", state=ActionState.AVAILABLE, created_at=datetime(2026, 1, 1, tzinfo=UTC), updated_at=datetime(2026, 1, 1, tzinfo=UTC), ) action_db = LifecycleActionModel.from_domain(action) session.add(action_db) session.commit() # Create parent plan parent = Plan( identity=PlanIdentity(plan_id="01HV0000000000000000PAR3N0"), action_name="local/hierarchy-action", namespaced_name=NamespacedName.parse("local/hierarchy-action"), description="Parent plan", definition_of_done="Done", phase=PlanPhase.EXECUTE, processing_state=ProcessingState.PROCESSING, strategy_actor="local/s", execution_actor="local/e", timestamps=PlanTimestamps( created_at=datetime(2026, 2, 1, tzinfo=UTC), updated_at=datetime(2026, 2, 1, tzinfo=UTC), ), ) parent_db = LifecyclePlanModel.from_domain( parent, action_name="local/hierarchy-action" ) session.add(parent_db) session.commit() # Create child plan child = Plan( identity=PlanIdentity( plan_id="01HV0000000000000000CHD100", parent_plan_id="01HV0000000000000000PAR3N0", root_plan_id="01HV0000000000000000PAR3N0", ), action_name="local/hierarchy-action", namespaced_name=NamespacedName.parse("local/hierarchy-action"), description="Child plan 1", definition_of_done="Done", phase=PlanPhase.STRATEGIZE, processing_state=ProcessingState.QUEUED, strategy_actor="local/s", execution_actor="local/e", timestamps=PlanTimestamps( created_at=datetime(2026, 2, 1, tzinfo=UTC), updated_at=datetime(2026, 2, 1, tzinfo=UTC), ), ) child_db = LifecyclePlanModel.from_domain( child, action_name="local/hierarchy-action" ) session.add(child_db) session.commit() # Verify hierarchy loaded_child = ( session.query(LifecyclePlanModel) .filter_by(plan_id="01HV0000000000000000CHD100") .one() ) restored_child = loaded_child.to_domain() assert restored_child.is_subplan assert restored_child.identity.parent_plan_id == "01HV0000000000000000PAR3N0" assert restored_child.identity.root_plan_id == "01HV0000000000000000PAR3N0" # Verify parent is root loaded_parent = ( session.query(LifecyclePlanModel) .filter_by(plan_id="01HV0000000000000000PAR3N0") .one() ) restored_parent = loaded_parent.to_domain() assert not restored_parent.is_subplan print("plan-hierarchy-ok") finally: session.close() Path(tmp).unlink(missing_ok=True) def _action_state_query() -> None: """Integration test: query actions by state from database.""" session, tmp = _create_temp_db() try: # Create multiple actions with different states state_names = { ActionState.AVAILABLE: "action-available", ActionState.ARCHIVED: "action-archived", } for state, name in state_names.items(): action = Action( namespaced_name=NamespacedName.parse(f"local/{name}"), description=f"Action in {state.value} state", definition_of_done="Done", strategy_actor="local/s", execution_actor="local/e", state=state, created_at=datetime(2026, 1, 1, tzinfo=UTC), updated_at=datetime(2026, 1, 1, tzinfo=UTC), ) db_model = LifecycleActionModel.from_domain(action) session.add(db_model) session.commit() # Query by state available = ( session.query(LifecycleActionModel).filter_by(state="available").all() ) assert len(available) == 1 archived = session.query(LifecycleActionModel).filter_by(state="archived").all() assert len(archived) == 1 # Query all by namespace local = session.query(LifecycleActionModel).filter_by(namespace="local").all() assert len(local) == 2 print("action-state-query-ok") finally: session.close() Path(tmp).unlink(missing_ok=True) def main() -> None: if len(sys.argv) < 2: raise SystemExit("Expected command argument") command = sys.argv[1] commands = { "action-round-trip": _action_round_trip, "plan-round-trip": _plan_round_trip, "plan-hierarchy": _plan_hierarchy_persistence, "action-state-query": _action_state_query, } if command not in commands: raise SystemExit(f"Unknown command: {command}") commands[command]() if __name__ == "__main__": main()