cleveragents-core/robot/helper_uow_lifecycle.py

"""Helper script for uow_lifecycle.robot smoke test."""

from __future__ import annotations

import sys

sys.path.insert(0, "src")
sys.path.insert(0, ".")

from datetime import datetime

from sqlalchemy import create_engine, event
from sqlalchemy.orm import Session, sessionmaker

from cleveragents.domain.models.core.action import Action, ActionState
from cleveragents.domain.models.core.plan import (
    AutomationLevel,
    NamespacedName,
    PlanIdentity,
    PlanPhase,
    PlanTimestamps,
    ProcessingState,
)
from cleveragents.domain.models.core.plan import Plan as V3Plan
from cleveragents.infrastructure.database.models import Base
from cleveragents.infrastructure.database.repositories import (
    ActionRepository,
    LifecyclePlanRepository,
)
from cleveragents.infrastructure.database.unit_of_work import UnitOfWorkContext


def main() -> None:
    engine = create_engine(
        "sqlite:///:memory:",
        echo=False,
        future=True,
        connect_args={"check_same_thread": False},
    )

    @event.listens_for(engine, "connect")
    def _fk(dbapi_conn, _rec):  # type: ignore[no-untyped-def]
        cursor = dbapi_conn.cursor()
        cursor.execute("PRAGMA foreign_keys=ON")
        cursor.close()

    Base.metadata.create_all(engine)
    sf: sessionmaker[Session] = sessionmaker(
        bind=engine,
        expire_on_commit=False,
        autoflush=False,
        autocommit=False,
        class_=Session,
    )

    # --- Test 1: create action + plan via UoW ---
    session = sf()
    ctx = UnitOfWorkContext(session)

    assert isinstance(ctx.actions, ActionRepository), "actions not ActionRepository"
    assert isinstance(ctx.lifecycle_plans, LifecyclePlanRepository), (
        "lifecycle_plans not LifecyclePlanRepository"
    )

    now = datetime.now()
    action = Action(
        namespaced_name=NamespacedName(namespace="local", name="robot-uow-action"),
        description="Robot UoW test action",
        long_description=None,
        definition_of_done="Done",
        strategy_actor="local/s",
        execution_actor="local/e",
        estimation_actor=None,
        review_actor=None,
        arguments=[],
        reusable=True,
        read_only=False,
        state=ActionState("available"),
        created_at=now,
        updated_at=now,
        created_by=None,
        tags=[],
    )
    ctx.actions.create(action)

    plan = V3Plan(
        identity=PlanIdentity(plan_id="01HGZ6FE0AQDYTR4BX00000R01", attempt=1),
        namespaced_name=NamespacedName(namespace="local", name="robot-uow-plan"),
        action_name="local/robot-uow-action",
        description="Robot UoW test plan",
        definition_of_done="Done",
        strategy_actor="local/s",
        execution_actor="local/e",
        phase=PlanPhase("action"),
        processing_state=ProcessingState("queued"),
        automation_level=AutomationLevel("manual"),
        timestamps=PlanTimestamps(created_at=now, updated_at=now),
        project_links=[],
        arguments={},
        arguments_order=[],
        invariants=[],
        reusable=True,
        read_only=False,
        created_by=None,
        tags=[],
    )
    ctx.lifecycle_plans.create(plan)
    session.commit()
    session.close()

    # --- Test 2: verify retrieval in new session ---
    session2 = sf()
    ctx2 = UnitOfWorkContext(session2)
    found_action = ctx2.actions.get_by_name("local/robot-uow-action")
    assert found_action is not None, "Action not found after commit"
    found_plan = ctx2.lifecycle_plans.get("01HGZ6FE0AQDYTR4BX00000R01")
    assert found_plan is not None, "Plan not found after commit"
    session2.close()

    print("PASS: uow_lifecycle smoke test")


if __name__ == "__main__":
    try:
        main()
    except Exception as exc:
        print(f"FAIL: {exc}", file=sys.stderr)
        sys.exit(1)