cleveragents-core/robot/helper_plan_phase_migration.py

"""Helper script for plan phase migration Robot integration tests.

Tests that the rebaselined v3_plans schema accepts Action phase
and Apply terminal processing states.
"""

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, ActionState
from cleveragents.domain.models.core.plan import (
    NamespacedName,
    Plan,
    PlanIdentity,
    PlanPhase,
    PlanTimestamps,
    ProcessingState,
)
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 _ensure_action(session: Session) -> None:
    """Ensure a parent action exists for FK references."""
    action = Action(
        namespaced_name=NamespacedName.parse("local/phase-migration-action"),
        description="Migration 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),
    )
    db_model = LifecycleActionModel.from_domain(action)
    session.add(db_model)
    session.commit()


def _action_queued() -> None:
    """Test: insert plan with phase='action', processing_state='queued'."""
    session, tmp = _create_temp_db()
    try:
        _ensure_action(session)

        plan = Plan(
            identity=PlanIdentity(plan_id="01HV000000000000000000MG01"),
            action_name="local/phase-migration-action",
            namespaced_name=NamespacedName.parse("local/phase-migration-action"),
            description="Phase migration test plan",
            phase=PlanPhase.ACTION,
            processing_state=ProcessingState.QUEUED,
            strategy_actor="local/s",
            execution_actor="local/e",
            timestamps=PlanTimestamps(
                created_at=datetime(2026, 2, 15, tzinfo=UTC),
                updated_at=datetime(2026, 2, 15, tzinfo=UTC),
            ),
        )

        plan_db = LifecyclePlanModel.from_domain(
            plan, action_name="local/phase-migration-action"
        )
        session.add(plan_db)
        session.commit()

        loaded = (
            session.query(LifecyclePlanModel)
            .filter_by(plan_id="01HV000000000000000000MG01")
            .one()
        )
        assert loaded.phase == "action"
        assert loaded.processing_state == "queued"

        print("action-queued-ok")
    finally:
        session.close()
        Path(tmp).unlink(missing_ok=True)


def _apply_applied() -> None:
    """Test: insert plan with phase='apply', processing_state='applied'."""
    session, tmp = _create_temp_db()
    try:
        _ensure_action(session)

        plan = Plan(
            identity=PlanIdentity(plan_id="01HV000000000000000000MG02"),
            action_name="local/phase-migration-action",
            namespaced_name=NamespacedName.parse("local/phase-migration-action"),
            description="Apply terminal test plan",
            phase=PlanPhase.APPLY,
            processing_state=ProcessingState.APPLIED,
            strategy_actor="local/s",
            execution_actor="local/e",
            timestamps=PlanTimestamps(
                created_at=datetime(2026, 2, 15, tzinfo=UTC),
                updated_at=datetime(2026, 2, 15, tzinfo=UTC),
            ),
        )

        plan_db = LifecyclePlanModel.from_domain(
            plan, action_name="local/phase-migration-action"
        )
        session.add(plan_db)
        session.commit()

        loaded = (
            session.query(LifecyclePlanModel)
            .filter_by(plan_id="01HV000000000000000000MG02")
            .one()
        )
        assert loaded.phase == "apply"
        assert loaded.processing_state == "applied"

        print("apply-applied-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-queued": _action_queued,
        "apply-applied": _apply_applied,
    }
    if command not in commands:
        raise SystemExit(f"Unknown command: {command}")
    commands[command]()


if __name__ == "__main__":
    main()