HAL9000
a998c5a0bf
The step_try_insert_plan_with_phase_and_state function was using raw SQL to insert a plan with an invalid phase value to test the CHECK constraint. However, the raw SQL was missing the root_plan_id column, which is NOT NULL in the schema. This caused the insert to fail with a NOT NULL violation instead of the intended CHECK constraint violation on the phase column. This fix adds root_plan_id to the raw SQL INSERT statement, setting it to the same value as plan_id (for a root plan). This allows the test to properly exercise the CHECK constraint on the phase column, ensuring the test fails for the correct reason. ISSUES CLOSED: #9411
227 lines
7.6 KiB
Python
227 lines
7.6 KiB
Python
"""Step definitions for plan phase/state constraint rebaseline tests."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import UTC, datetime
|
|
from typing import Any
|
|
|
|
from behave import given, then, when
|
|
from sqlalchemy import create_engine, text
|
|
from sqlalchemy.exc import IntegrityError
|
|
from sqlalchemy.orm import Session, sessionmaker
|
|
|
|
from cleveragents.infrastructure.database.models import (
|
|
Base,
|
|
LifecycleActionModel,
|
|
LifecyclePlanModel,
|
|
)
|
|
|
|
_VALID_ULID_PLAN = "01HV000000000000000000PH01"
|
|
_VALID_ULID_COUNTER = 0
|
|
|
|
|
|
def _next_ulid() -> str:
|
|
global _VALID_ULID_COUNTER
|
|
_VALID_ULID_COUNTER += 1
|
|
suffix = str(_VALID_ULID_COUNTER).zfill(4)
|
|
return f"01HV0000000000000000PH{suffix}"
|
|
|
|
|
|
def _now_iso() -> str:
|
|
return datetime.now(tz=UTC).isoformat()
|
|
|
|
|
|
def _ensure_action(session: Session) -> None:
|
|
"""Ensure a parent action exists for FK references."""
|
|
existing = (
|
|
session.query(LifecycleActionModel)
|
|
.filter_by(namespaced_name="local/phase-test-action")
|
|
.first()
|
|
)
|
|
if existing is not None:
|
|
return
|
|
|
|
action = LifecycleActionModel()
|
|
action.namespaced_name = "local/phase-test-action"
|
|
action.namespace = "local"
|
|
action.name = "phase-test-action"
|
|
action.description = "Test action for phase migration"
|
|
action.definition_of_done = "Done"
|
|
action.strategy_actor = "local/s"
|
|
action.execution_actor = "local/e"
|
|
action.state = "available"
|
|
action.tags_json = "[]"
|
|
action.created_at = _now_iso()
|
|
action.updated_at = _now_iso()
|
|
session.add(action)
|
|
session.commit()
|
|
|
|
|
|
@given("the plan phase rebaseline database is initialized")
|
|
def step_plan_phase_rebaseline_db_init(context: Any) -> None:
|
|
"""Set up an in-memory database with the schema including new constraints."""
|
|
engine = create_engine("sqlite:///:memory:")
|
|
|
|
# Enable FK enforcement
|
|
from sqlalchemy import event
|
|
|
|
@event.listens_for(engine, "connect")
|
|
def set_sqlite_pragma(dbapi_connection: Any, _: Any) -> None:
|
|
cursor = dbapi_connection.cursor()
|
|
cursor.execute("PRAGMA foreign_keys=ON")
|
|
cursor.close()
|
|
|
|
Base.metadata.create_all(engine)
|
|
factory = sessionmaker(bind=engine)
|
|
session = factory()
|
|
_ensure_action(session)
|
|
context.phase_rebaseline_engine = engine
|
|
context.phase_rebaseline_session = session
|
|
|
|
|
|
@when('I insert a plan with phase "{phase}" and state "{state}"')
|
|
def step_insert_plan_with_phase_and_state(context: Any, phase: str, state: str) -> None:
|
|
"""Insert a plan with the specified phase and state."""
|
|
session: Session = context.phase_rebaseline_session
|
|
ulid = _next_ulid()
|
|
now = _now_iso()
|
|
plan = LifecyclePlanModel()
|
|
plan.plan_id = ulid
|
|
plan.root_plan_id = ulid
|
|
plan.action_name = "local/phase-test-action"
|
|
plan.namespaced_name = "local/phase-test-plan"
|
|
plan.namespace = "local"
|
|
plan.phase = phase
|
|
plan.processing_state = state
|
|
plan.description = "Test plan"
|
|
plan.tags_json = "[]"
|
|
plan.effective_profile_snapshot = "{}"
|
|
plan.created_at = now
|
|
plan.updated_at = now
|
|
try:
|
|
session.add(plan)
|
|
session.commit()
|
|
context.phase_rebaseline_result = "ok"
|
|
context.phase_rebaseline_plan_id = ulid
|
|
except (IntegrityError, Exception) as exc:
|
|
session.rollback()
|
|
context.phase_rebaseline_result = "error"
|
|
context.phase_rebaseline_error = str(exc)
|
|
|
|
|
|
@when('I try to insert a plan with phase "{phase}" and state "{state}"')
|
|
def step_try_insert_plan_with_phase_and_state(
|
|
context: Any, phase: str, state: str
|
|
) -> None:
|
|
"""Try inserting a plan without ORM defaults to exercise the phase constraint.
|
|
|
|
The direct SQL insert must provide both plan_id and root_plan_id so the
|
|
phase constraint is validated instead of failing on missing root_plan_id.
|
|
"""
|
|
session: Session = context.phase_rebaseline_session
|
|
ulid = _next_ulid()
|
|
now = _now_iso()
|
|
try:
|
|
session.execute(
|
|
text(
|
|
"INSERT INTO v3_plans "
|
|
"(plan_id, root_plan_id, action_name, namespaced_name, namespace, "
|
|
"phase, processing_state, description, tags_json, "
|
|
"created_at, updated_at) "
|
|
"VALUES (:pid, :rpid, :aname, :nname, :ns, :phase, :state, "
|
|
":desc, :tags, :cat, :uat)"
|
|
),
|
|
{
|
|
"pid": ulid,
|
|
"rpid": ulid,
|
|
"aname": "local/phase-test-action",
|
|
"nname": "local/try-plan",
|
|
"ns": "local",
|
|
"phase": phase,
|
|
"state": state,
|
|
"desc": "Test plan",
|
|
"tags": "[]",
|
|
"cat": now,
|
|
"uat": now,
|
|
},
|
|
)
|
|
session.commit()
|
|
context.phase_rebaseline_result = "ok"
|
|
except (IntegrityError, Exception) as exc:
|
|
session.rollback()
|
|
context.phase_rebaseline_result = "error"
|
|
context.phase_rebaseline_error = str(exc)
|
|
|
|
|
|
@when("I insert a plan using default phase")
|
|
def step_insert_plan_default_phase(context: Any) -> None:
|
|
"""Insert a plan without specifying a phase to test the ORM default."""
|
|
session: Session = context.phase_rebaseline_session
|
|
ulid = _next_ulid()
|
|
now = _now_iso()
|
|
# Use ORM to create plan without specifying phase — should default to 'action'
|
|
plan = LifecyclePlanModel()
|
|
plan.plan_id = ulid
|
|
plan.root_plan_id = ulid
|
|
plan.action_name = "local/phase-test-action"
|
|
plan.namespaced_name = "local/default-phase-plan"
|
|
plan.namespace = "local"
|
|
plan.description = "Default phase plan"
|
|
plan.tags_json = "[]"
|
|
plan.effective_profile_snapshot = "{}"
|
|
plan.created_at = now
|
|
plan.updated_at = now
|
|
session.add(plan)
|
|
session.commit()
|
|
context.phase_rebaseline_plan_id = ulid
|
|
|
|
|
|
@then('the plan should be persisted successfully with phase "{phase}"')
|
|
def step_verify_plan_persisted_phase(context: Any, phase: str) -> None:
|
|
"""Verify the plan was persisted with the expected phase."""
|
|
assert context.phase_rebaseline_result == "ok", (
|
|
f"Expected ok, got {context.phase_rebaseline_result}"
|
|
)
|
|
session: Session = context.phase_rebaseline_session
|
|
plan = (
|
|
session.query(LifecyclePlanModel)
|
|
.filter_by(plan_id=context.phase_rebaseline_plan_id)
|
|
.one()
|
|
)
|
|
assert plan.phase == phase
|
|
|
|
|
|
@then('the plan should be persisted successfully with state "{state}"')
|
|
def step_verify_plan_persisted_state(context: Any, state: str) -> None:
|
|
"""Verify the plan was persisted with the expected state."""
|
|
assert context.phase_rebaseline_result == "ok", (
|
|
f"Expected ok, got {context.phase_rebaseline_result}"
|
|
)
|
|
session: Session = context.phase_rebaseline_session
|
|
plan = (
|
|
session.query(LifecyclePlanModel)
|
|
.filter_by(plan_id=context.phase_rebaseline_plan_id)
|
|
.one()
|
|
)
|
|
assert plan.processing_state == state
|
|
|
|
|
|
@then("the insert should fail with a constraint violation")
|
|
def step_verify_constraint_violation(context: Any) -> None:
|
|
"""Verify the insert failed."""
|
|
assert context.phase_rebaseline_result == "error", (
|
|
f"Expected error, got {context.phase_rebaseline_result}"
|
|
)
|
|
|
|
|
|
@then('the rebaselined plan phase should be "{phase}"')
|
|
def step_verify_rebaselined_plan_phase_default(context: Any, phase: str) -> None:
|
|
"""Verify the plan's phase matches the expected value."""
|
|
session: Session = context.phase_rebaseline_session
|
|
plan = (
|
|
session.query(LifecyclePlanModel)
|
|
.filter_by(plan_id=context.phase_rebaseline_plan_id)
|
|
.one()
|
|
)
|
|
assert plan.phase == phase, f"Expected phase '{phase}', got '{plan.phase}'"
|