127 lines
3.7 KiB
Python
127 lines
3.7 KiB
Python
"""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 (
|
|
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"),
|
|
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)
|