338 lines
12 KiB
Python
338 lines
12 KiB
Python
"""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()
|