109 lines
3.0 KiB
Python
109 lines
3.0 KiB
Python
"""Helper script for plan_repository.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
|
|
from sqlalchemy.orm import 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,
|
|
)
|
|
|
|
|
|
def main() -> None:
|
|
engine = create_engine("sqlite:///:memory:")
|
|
Base.metadata.create_all(engine)
|
|
session = sessionmaker(bind=engine)()
|
|
factory = lambda: session # noqa: E731
|
|
|
|
# Create prerequisite action
|
|
action_repo = ActionRepository(session_factory=factory)
|
|
action = Action(
|
|
namespaced_name=NamespacedName(namespace="local", name="robot-action"),
|
|
description="Robot 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=datetime.now(),
|
|
updated_at=datetime.now(),
|
|
created_by=None,
|
|
tags=[],
|
|
)
|
|
action_repo.create(action)
|
|
session.commit()
|
|
|
|
# Test plan CRUD
|
|
repo = LifecyclePlanRepository(session_factory=factory)
|
|
now = datetime.now()
|
|
plan = V3Plan(
|
|
identity=PlanIdentity(plan_id="01KHG51GK1G5TJ678GT8JEF55Y", attempt=1),
|
|
namespaced_name=NamespacedName(namespace="local", name="robot-plan"),
|
|
action_name="local/robot-action",
|
|
description="Robot test",
|
|
phase=PlanPhase("action"),
|
|
processing_state=ProcessingState("queued"),
|
|
strategy_actor="local/s",
|
|
execution_actor="local/e",
|
|
timestamps=PlanTimestamps(created_at=now, updated_at=now),
|
|
created_by="robot",
|
|
tags=[],
|
|
reusable=True,
|
|
read_only=False,
|
|
)
|
|
|
|
# Create
|
|
repo.create(plan)
|
|
session.commit()
|
|
|
|
# Get
|
|
got = repo.get("01KHG51GK1G5TJ678GT8JEF55Y")
|
|
assert got is not None, "Plan not found"
|
|
assert got.description == "Robot test", f"Bad desc: {got.description}"
|
|
|
|
# List
|
|
plans = repo.list_plans(phase="action")
|
|
assert len(plans) == 1, f"Expected 1, got {len(plans)}"
|
|
|
|
# Count
|
|
count = repo.count(phase="action")
|
|
assert count == 1, f"Expected count 1, got {count}"
|
|
|
|
# Delete
|
|
deleted = repo.delete("01KHG51GK1G5TJ678GT8JEF55Y")
|
|
session.commit()
|
|
assert deleted is True, "Delete failed"
|
|
|
|
gone = repo.get("01KHG51GK1G5TJ678GT8JEF55Y")
|
|
assert gone is None, "Plan still exists"
|
|
|
|
print("PASS: plan_repository smoke test")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|