352 lines
11 KiB
Python
352 lines
11 KiB
Python
"""Step definitions for UnitOfWork lifecycle repository wiring.
|
|
|
|
Targets ``UnitOfWork`` and ``UnitOfWorkContext`` in
|
|
``src/cleveragents/infrastructure/database/unit_of_work.py``.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import datetime
|
|
from typing import Any
|
|
|
|
from behave import given, then, when
|
|
from behave.runner import Context
|
|
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,
|
|
PlanRepository,
|
|
)
|
|
from cleveragents.infrastructure.database.unit_of_work import (
|
|
UnitOfWorkContext,
|
|
)
|
|
|
|
|
|
def _make_uow_action(name: str = "local/uow-action") -> Action:
|
|
"""Create a minimal Action domain object for testing."""
|
|
parts = name.split("/", 1)
|
|
namespace = parts[0] if len(parts) == 2 else "local"
|
|
short_name = parts[1] if len(parts) == 2 else parts[0]
|
|
now = datetime.now()
|
|
return Action(
|
|
namespaced_name=NamespacedName(namespace=namespace, name=short_name),
|
|
description=f"UoW test action {short_name}",
|
|
long_description=None,
|
|
definition_of_done=f"Verify {short_name} completes",
|
|
strategy_actor="local/strategist",
|
|
execution_actor="local/executor",
|
|
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=[],
|
|
)
|
|
|
|
|
|
def _make_uow_plan(
|
|
plan_id: str,
|
|
action_name: str = "local/uow-action",
|
|
) -> V3Plan:
|
|
"""Create a minimal Plan domain object for testing."""
|
|
now = datetime.now()
|
|
return V3Plan(
|
|
identity=PlanIdentity(plan_id=plan_id, attempt=1),
|
|
namespaced_name=NamespacedName(namespace="local", name="uow-plan"),
|
|
action_name=action_name,
|
|
description="UoW test plan",
|
|
definition_of_done="Verify UoW plan completes",
|
|
strategy_actor="local/strategist",
|
|
execution_actor="local/executor",
|
|
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=[],
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Background
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("a UnitOfWork backed by an in-memory database for lifecycle tests")
|
|
def step_uow_lifecycle_background(context: Context) -> None:
|
|
"""Set up an in-memory database with schema for UoW tests.
|
|
|
|
We bypass the migration runner and use ``Base.metadata.create_all``
|
|
directly since this is a unit test with an in-memory DB.
|
|
"""
|
|
engine = create_engine(
|
|
"sqlite:///:memory:",
|
|
echo=False,
|
|
future=True,
|
|
connect_args={"check_same_thread": False},
|
|
)
|
|
|
|
# Enable FK enforcement for SQLite
|
|
@event.listens_for(engine, "connect")
|
|
def _set_sqlite_fk(dbapi_conn: Any, _rec: Any) -> None:
|
|
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,
|
|
)
|
|
|
|
# Store on context for reuse
|
|
context.uow_engine = engine
|
|
context.uow_session_factory = sf
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Repository exposure
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when("I access the actions repository from the UoW context")
|
|
def step_access_actions_repo(context: Context) -> None:
|
|
session = context.uow_session_factory()
|
|
ctx = UnitOfWorkContext(session)
|
|
context.uow_ctx_result = ctx.actions
|
|
session.close()
|
|
|
|
|
|
@then("the actions repository should be an ActionRepository instance")
|
|
def step_check_actions_repo_type(context: Context) -> None:
|
|
assert isinstance(context.uow_ctx_result, ActionRepository), (
|
|
f"Expected ActionRepository, got {type(context.uow_ctx_result)}"
|
|
)
|
|
|
|
|
|
@when("I access the lifecycle_plans repository from the UoW context")
|
|
def step_access_lifecycle_plans_repo(context: Context) -> None:
|
|
session = context.uow_session_factory()
|
|
ctx = UnitOfWorkContext(session)
|
|
context.uow_ctx_result = ctx.lifecycle_plans
|
|
session.close()
|
|
|
|
|
|
@then("the lifecycle_plans repository should be a LifecyclePlanRepository instance")
|
|
def step_check_lifecycle_plans_repo_type(context: Context) -> None:
|
|
assert isinstance(context.uow_ctx_result, LifecyclePlanRepository), (
|
|
f"Expected LifecyclePlanRepository, got {type(context.uow_ctx_result)}"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# CRUD via UoW
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when('I create an action "{name}" via the UoW transaction')
|
|
def step_create_action_via_uow(context: Context, name: str) -> None:
|
|
session = context.uow_session_factory()
|
|
try:
|
|
ctx = UnitOfWorkContext(session)
|
|
action = _make_uow_action(name)
|
|
ctx.actions.create(action)
|
|
session.commit()
|
|
except Exception:
|
|
session.rollback()
|
|
raise
|
|
finally:
|
|
session.close()
|
|
|
|
|
|
@then('the action "{name}" should be retrievable from the actions repository')
|
|
def step_check_action_retrievable(context: Context, name: str) -> None:
|
|
session = context.uow_session_factory()
|
|
try:
|
|
ctx = UnitOfWorkContext(session)
|
|
action = ctx.actions.get_by_name(name)
|
|
assert action is not None, f"Action {name} not found"
|
|
finally:
|
|
session.close()
|
|
|
|
|
|
@given('an action "{name}" exists via the UoW')
|
|
def step_ensure_action_exists(context: Context, name: str) -> None:
|
|
session = context.uow_session_factory()
|
|
try:
|
|
ctx = UnitOfWorkContext(session)
|
|
existing = ctx.actions.get_by_name(name)
|
|
if existing is None:
|
|
action = _make_uow_action(name)
|
|
ctx.actions.create(action)
|
|
session.commit()
|
|
except Exception:
|
|
session.rollback()
|
|
raise
|
|
finally:
|
|
session.close()
|
|
|
|
|
|
@when('I create a lifecycle plan "{plan_id}" via the UoW transaction')
|
|
def step_create_plan_via_uow(context: Context, plan_id: str) -> None:
|
|
session = context.uow_session_factory()
|
|
try:
|
|
ctx = UnitOfWorkContext(session)
|
|
plan = _make_uow_plan(plan_id, action_name="local/uow-plan-action")
|
|
ctx.lifecycle_plans.create(plan)
|
|
session.commit()
|
|
except Exception:
|
|
session.rollback()
|
|
raise
|
|
finally:
|
|
session.close()
|
|
|
|
|
|
@then('the plan "{plan_id}" should be retrievable from the lifecycle_plans repository')
|
|
def step_check_plan_retrievable(context: Context, plan_id: str) -> None:
|
|
session = context.uow_session_factory()
|
|
try:
|
|
ctx = UnitOfWorkContext(session)
|
|
plan = ctx.lifecycle_plans.get(plan_id)
|
|
assert plan is not None, f"Plan {plan_id} not found"
|
|
finally:
|
|
session.close()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Cross-repository commit
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when(
|
|
'I create an action "{action_name}" and a plan "{plan_id}" in a single UoW transaction'
|
|
)
|
|
def step_create_action_and_plan(
|
|
context: Context, action_name: str, plan_id: str
|
|
) -> None:
|
|
session = context.uow_session_factory()
|
|
try:
|
|
ctx = UnitOfWorkContext(session)
|
|
action = _make_uow_action(action_name)
|
|
ctx.actions.create(action)
|
|
plan = _make_uow_plan(plan_id, action_name=action_name)
|
|
ctx.lifecycle_plans.create(plan)
|
|
session.commit()
|
|
except Exception:
|
|
session.rollback()
|
|
raise
|
|
finally:
|
|
session.close()
|
|
|
|
|
|
@then('the action "{name}" should be retrievable in a new transaction')
|
|
def step_check_action_new_txn(context: Context, name: str) -> None:
|
|
session = context.uow_session_factory()
|
|
try:
|
|
ctx = UnitOfWorkContext(session)
|
|
action = ctx.actions.get_by_name(name)
|
|
assert action is not None, f"Action {name} not found in new transaction"
|
|
finally:
|
|
session.close()
|
|
|
|
|
|
@then('the plan "{plan_id}" should be retrievable in a new transaction')
|
|
def step_check_plan_new_txn(context: Context, plan_id: str) -> None:
|
|
session = context.uow_session_factory()
|
|
try:
|
|
ctx = UnitOfWorkContext(session)
|
|
plan = ctx.lifecycle_plans.get(plan_id)
|
|
assert plan is not None, f"Plan {plan_id} not found in new transaction"
|
|
finally:
|
|
session.close()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Rollback
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when('I create an action "{action_name}" and a plan "{plan_id}" but force a rollback')
|
|
def step_create_and_rollback(context: Context, action_name: str, plan_id: str) -> None:
|
|
session = context.uow_session_factory()
|
|
try:
|
|
ctx = UnitOfWorkContext(session)
|
|
action = _make_uow_action(action_name)
|
|
ctx.actions.create(action)
|
|
plan = _make_uow_plan(plan_id, action_name=action_name)
|
|
ctx.lifecycle_plans.create(plan)
|
|
# Force rollback instead of commit
|
|
session.rollback()
|
|
finally:
|
|
session.close()
|
|
|
|
|
|
@then('the action "{name}" should not exist in a new transaction')
|
|
def step_check_action_not_exists(context: Context, name: str) -> None:
|
|
session = context.uow_session_factory()
|
|
try:
|
|
ctx = UnitOfWorkContext(session)
|
|
action = ctx.actions.get_by_name(name)
|
|
assert action is None, f"Action {name} should not exist after rollback"
|
|
finally:
|
|
session.close()
|
|
|
|
|
|
@then('the plan "{plan_id}" should not exist in a new transaction')
|
|
def step_check_plan_not_exists(context: Context, plan_id: str) -> None:
|
|
session = context.uow_session_factory()
|
|
try:
|
|
ctx = UnitOfWorkContext(session)
|
|
plan = ctx.lifecycle_plans.get(plan_id)
|
|
assert plan is None, f"Plan {plan_id} should not exist after rollback"
|
|
finally:
|
|
session.close()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Legacy accessor
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when("I access the legacy plans repository from the UoW context")
|
|
def step_access_legacy_plans_repo(context: Context) -> None:
|
|
session = context.uow_session_factory()
|
|
ctx = UnitOfWorkContext(session)
|
|
context.uow_ctx_result = ctx.plans
|
|
session.close()
|
|
|
|
|
|
@then("the legacy plans repository should be a PlanRepository instance")
|
|
def step_check_legacy_plans_repo_type(context: Context) -> None:
|
|
assert isinstance(context.uow_ctx_result, PlanRepository), (
|
|
f"Expected PlanRepository, got {type(context.uow_ctx_result)}"
|
|
)
|