cleveragents-core/features/steps/db_schema_cascade_steps.py

"""Step definitions for db_migration_lifecycle.feature — cascade and persistence.

``ondelete="SET NULL"`` cascade verification for checkpoint_metadata foreign
keys, positive FK persistence tests, and trigger removal verification on
downgrade.
"""

from __future__ import annotations

from typing import Any

from behave import given, then, when
from sqlalchemy import text
from sqlalchemy.exc import IntegrityError

from features.steps.db_schema_parity_steps import (
    _ensure_test_action_and_plan,
    _insert_test_resource,
)

# ---------------------------------------------------------------------------
# Given/When/Then — positive FK persistence test
# ---------------------------------------------------------------------------


@given("valid decision and resource rows exist")
def step_valid_decision_and_resource_exist(context: Any) -> None:
    action_name = "local/test-action-persist"
    plan_id = "01ARZ3NDEKTSV4RRFFQ69G5FE0"
    decision_id = "01ARZ3NDEKTSV4RRFFQ69G5FE1"
    resource_id = "01ARZ3NDEKTSV4RRFFQ69G5FE2"

    with context.engine.begin() as conn:
        _ensure_test_action_and_plan(conn, action_name, plan_id)

        conn.execute(
            text(
                """
                INSERT INTO decisions (
                    decision_id, plan_id, decision_type, question,
                    chosen_option, context_snapshot_json, sequence_number,
                    created_at
                ) VALUES (
                    :decision_id, :plan_id, :decision_type, :question,
                    :chosen_option, :context_snapshot_json, :sequence_number,
                    :created_at
                )
                """
            ),
            {
                "decision_id": decision_id,
                "plan_id": plan_id,
                "decision_type": "strategy_choice",
                "question": "test question",
                "chosen_option": "test option",
                "context_snapshot_json": "{}",
                "sequence_number": 1,
                "created_at": "2026-01-01T00:00:00",
            },
        )

        _insert_test_resource(conn, resource_id)

    context.persist_plan_id = plan_id
    context.persist_decision_id = decision_id
    context.persist_resource_id = resource_id
    context.persist_checkpoint_id = "01ARZ3NDEKTSV4RRFFQ69G5FE3"


@when("I insert a checkpoint with valid FK references")
def step_insert_checkpoint_valid_fk(context: Any) -> None:
    with context.engine.begin() as conn:
        conn.execute(
            text(
                """
                INSERT INTO checkpoint_metadata (
                    checkpoint_id, plan_id, decision_id,
                    checkpoint_type, resource_id, sandbox_ref,
                    filesystem_path, created_at
                ) VALUES (
                    :checkpoint_id, :plan_id, :decision_id,
                    :checkpoint_type, :resource_id, :sandbox_ref,
                    :filesystem_path, :created_at
                )
                """
            ),
            {
                "checkpoint_id": context.persist_checkpoint_id,
                "plan_id": context.persist_plan_id,
                "decision_id": context.persist_decision_id,
                "checkpoint_type": "manual",
                "resource_id": context.persist_resource_id,
                "sandbox_ref": "test-ref",
                "filesystem_path": "",
                "created_at": "2026-01-01T00:00:00",
            },
        )


@then("the checkpoint row should be persisted in the database")
def step_checkpoint_persisted(context: Any) -> None:
    with context.engine.connect() as conn:
        row = conn.execute(
            text(
                "SELECT checkpoint_id, decision_id, resource_id "
                "FROM checkpoint_metadata WHERE checkpoint_id = :cid"
            ),
            {"cid": context.persist_checkpoint_id},
        ).fetchone()

    assert row is not None, (
        f"Checkpoint {context.persist_checkpoint_id!r} was not persisted"
    )
    assert row[0] == context.persist_checkpoint_id
    assert row[1] == context.persist_decision_id, (
        f"Expected decision_id {context.persist_decision_id!r}, got {row[1]!r}"
    )
    assert row[2] == context.persist_resource_id, (
        f"Expected resource_id {context.persist_resource_id!r}, got {row[2]!r}"
    )


# ---------------------------------------------------------------------------
# Given/When/Then — UPDATE trigger rejection
# ---------------------------------------------------------------------------


@given("a checkpoint exists with valid FK references")
def step_checkpoint_with_valid_fk_refs(context: Any) -> None:
    action_name = "local/test-action-upd-trg"
    plan_id = "01ARZ3NDEKTSV4RRFFQ69G5FH0"
    decision_id = "01ARZ3NDEKTSV4RRFFQ69G5FH1"
    resource_id = "01ARZ3NDEKTSV4RRFFQ69G5FH2"
    checkpoint_id = "01ARZ3NDEKTSV4RRFFQ69G5FH3"

    with context.engine.begin() as conn:
        _ensure_test_action_and_plan(conn, action_name, plan_id)

        conn.execute(
            text(
                """
                INSERT INTO decisions (
                    decision_id, plan_id, decision_type, question,
                    chosen_option, context_snapshot_json, sequence_number,
                    created_at
                ) VALUES (
                    :decision_id, :plan_id, :decision_type, :question,
                    :chosen_option, :context_snapshot_json, :sequence_number,
                    :created_at
                )
                """
            ),
            {
                "decision_id": decision_id,
                "plan_id": plan_id,
                "decision_type": "strategy_choice",
                "question": "test question",
                "chosen_option": "test option",
                "context_snapshot_json": "{}",
                "sequence_number": 1,
                "created_at": "2026-01-01T00:00:00",
            },
        )

        _insert_test_resource(conn, resource_id)

        conn.execute(
            text(
                """
                INSERT INTO checkpoint_metadata (
                    checkpoint_id, plan_id, decision_id,
                    checkpoint_type, resource_id, sandbox_ref,
                    filesystem_path, created_at
                ) VALUES (
                    :checkpoint_id, :plan_id, :decision_id,
                    :checkpoint_type, :resource_id, :sandbox_ref,
                    :filesystem_path, :created_at
                )
                """
            ),
            {
                "checkpoint_id": checkpoint_id,
                "plan_id": plan_id,
                "decision_id": decision_id,
                "checkpoint_type": "manual",
                "resource_id": resource_id,
                "sandbox_ref": "test-ref",
                "filesystem_path": "",
                "created_at": "2026-01-01T00:00:00",
            },
        )

    context.update_trg_checkpoint_id = checkpoint_id


@when("I update the checkpoint decision_id to a non-existent value")
def step_update_checkpoint_decision_orphan(context: Any) -> None:
    try:
        with context.engine.begin() as conn:
            conn.execute(
                text(
                    "UPDATE checkpoint_metadata "
                    "SET decision_id = :orphan "
                    "WHERE checkpoint_id = :cid"
                ),
                {
                    "orphan": "NONEXISTENT_DECISION_ID_UPD",
                    "cid": context.update_trg_checkpoint_id,
                },
            )
    except IntegrityError:
        context.update_trigger_rejected = True
        return

    context.update_trigger_rejected = False


@when("I update the checkpoint resource_id to a non-existent value")
def step_update_checkpoint_resource_orphan(context: Any) -> None:
    try:
        with context.engine.begin() as conn:
            conn.execute(
                text(
                    "UPDATE checkpoint_metadata "
                    "SET resource_id = :orphan "
                    "WHERE checkpoint_id = :cid"
                ),
                {
                    "orphan": "NONEXISTENT_RESOURCE_ID_UPD",
                    "cid": context.update_trg_checkpoint_id,
                },
            )
    except IntegrityError:
        context.update_trigger_rejected = True
        return

    context.update_trigger_rejected = False


@then("the update should be rejected with an integrity error")
def step_update_rejected(context: Any) -> None:
    assert context.update_trigger_rejected, (
        "Expected UPDATE to be rejected by FK trigger, but it was accepted"
    )


# ---------------------------------------------------------------------------
# Helper — enable FK enforcement on the pooled DBAPI connection
# ---------------------------------------------------------------------------


def _enable_fk_enforcement(engine: Any) -> None:
    """Enable PRAGMA foreign_keys on the engine's pooled DBAPI connection.

    For in-memory SQLite databases, SQLAlchemy uses ``StaticPool`` which
    shares a single underlying DBAPI connection across all pool checkouts.
    Calling ``raw_connection()`` returns a wrapper around that shared
    connection; ``close()`` returns it to the pool without closing the
    DBAPI connection.  The PRAGMA therefore persists for all subsequent
    ``engine.begin()`` / ``engine.connect()`` calls.

    This matches the codebase convention of setting PRAGMA foreign_keys
    at the connection level (e.g. ``resource_repository_steps.py``,
    ``plan_lifecycle_persistence_steps.py``).  The ``event.listens_for``
    variant used in engine-creation contexts is not applicable here
    because the engine's connections have already been established during
    migration; the ``"connect"`` event would not fire for existing
    pooled connections.
    """
    raw_conn = engine.raw_connection()
    try:
        cursor = raw_conn.cursor()
        cursor.execute("PRAGMA foreign_keys=ON")
        cursor.close()
    finally:
        raw_conn.close()


# ---------------------------------------------------------------------------
# Given/When/Then — ondelete="SET NULL" cascade (decision)
# ---------------------------------------------------------------------------


@given("a checkpoint references a valid decision")
def step_checkpoint_references_decision(context: Any) -> None:
    action_name = "local/test-action-set-null-d"
    plan_id = "01ARZ3NDEKTSV4RRFFQ69G5FG0"
    decision_id = "01ARZ3NDEKTSV4RRFFQ69G5FG1"
    checkpoint_id = "01ARZ3NDEKTSV4RRFFQ69G5FG2"

    with context.engine.begin() as conn:
        _ensure_test_action_and_plan(conn, action_name, plan_id)

        conn.execute(
            text(
                """
                INSERT INTO decisions (
                    decision_id, plan_id, decision_type, question,
                    chosen_option, context_snapshot_json, sequence_number,
                    created_at
                ) VALUES (
                    :decision_id, :plan_id, :decision_type, :question,
                    :chosen_option, :context_snapshot_json, :sequence_number,
                    :created_at
                )
                """
            ),
            {
                "decision_id": decision_id,
                "plan_id": plan_id,
                "decision_type": "strategy_choice",
                "question": "test question",
                "chosen_option": "test option",
                "context_snapshot_json": "{}",
                "sequence_number": 1,
                "created_at": "2026-01-01T00:00:00",
            },
        )

        conn.execute(
            text(
                """
                INSERT INTO checkpoint_metadata (
                    checkpoint_id, plan_id, decision_id,
                    checkpoint_type, sandbox_ref,
                    filesystem_path, created_at
                ) VALUES (
                    :checkpoint_id, :plan_id, :decision_id,
                    :checkpoint_type, :sandbox_ref,
                    :filesystem_path, :created_at
                )
                """
            ),
            {
                "checkpoint_id": checkpoint_id,
                "plan_id": plan_id,
                "decision_id": decision_id,
                "checkpoint_type": "manual",
                "sandbox_ref": "test-ref",
                "filesystem_path": "",
                "created_at": "2026-01-01T00:00:00",
            },
        )

    context.set_null_decision_id = decision_id
    context.set_null_checkpoint_id_d = checkpoint_id
    context.set_null_plan_id_d = plan_id


@when("the referenced decision is deleted")
def step_delete_referenced_decision(context: Any) -> None:
    # Enable FK enforcement via the established codebase pattern
    # (event listener on "connect") so that ondelete="SET NULL" is honoured.
    _enable_fk_enforcement(context.engine)

    with context.engine.begin() as conn:
        conn.execute(
            text("DELETE FROM decisions WHERE decision_id = :did"),
            {"did": context.set_null_decision_id},
        )


@then("the checkpoint decision_id should be NULL")
def step_checkpoint_decision_id_null(context: Any) -> None:
    with context.engine.connect() as conn:
        row = conn.execute(
            text(
                "SELECT decision_id FROM checkpoint_metadata WHERE checkpoint_id = :cid"
            ),
            {"cid": context.set_null_checkpoint_id_d},
        ).fetchone()

    assert row is not None, (
        f"Checkpoint {context.set_null_checkpoint_id_d!r} missing after decision deletion"
    )
    assert row[0] is None, (
        f"Expected checkpoint decision_id to be NULL after parent deletion, "
        f"got {row[0]!r}"
    )


# ---------------------------------------------------------------------------
# Given/When/Then — ondelete="SET NULL" cascade (resource)
# ---------------------------------------------------------------------------


@given("a checkpoint references a valid resource")
def step_checkpoint_references_resource(context: Any) -> None:
    action_name = "local/test-action-set-null-r"
    plan_id = "01ARZ3NDEKTSV4RRFFQ69G5FG3"
    resource_id = "01ARZ3NDEKTSV4RRFFQ69G5FG4"
    checkpoint_id = "01ARZ3NDEKTSV4RRFFQ69G5FG5"

    with context.engine.begin() as conn:
        _ensure_test_action_and_plan(conn, action_name, plan_id)

        _insert_test_resource(conn, resource_id)

        conn.execute(
            text(
                """
                INSERT INTO checkpoint_metadata (
                    checkpoint_id, plan_id, resource_id,
                    checkpoint_type, sandbox_ref,
                    filesystem_path, created_at
                ) VALUES (
                    :checkpoint_id, :plan_id, :resource_id,
                    :checkpoint_type, :sandbox_ref,
                    :filesystem_path, :created_at
                )
                """
            ),
            {
                "checkpoint_id": checkpoint_id,
                "plan_id": plan_id,
                "resource_id": resource_id,
                "checkpoint_type": "manual",
                "sandbox_ref": "test-ref",
                "filesystem_path": "",
                "created_at": "2026-01-01T00:00:00",
            },
        )

    context.set_null_resource_id = resource_id
    context.set_null_checkpoint_id_r = checkpoint_id


@when("the referenced resource is deleted")
def step_delete_referenced_resource(context: Any) -> None:
    # Enable FK enforcement via the established codebase pattern
    # (event listener on "connect") so that ondelete="SET NULL" is honoured.
    _enable_fk_enforcement(context.engine)

    with context.engine.begin() as conn:
        conn.execute(
            text("DELETE FROM resources WHERE resource_id = :rid"),
            {"rid": context.set_null_resource_id},
        )


@then("the checkpoint resource_id should be NULL")
def step_checkpoint_resource_id_null(context: Any) -> None:
    with context.engine.connect() as conn:
        row = conn.execute(
            text(
                "SELECT resource_id FROM checkpoint_metadata WHERE checkpoint_id = :cid"
            ),
            {"cid": context.set_null_checkpoint_id_r},
        ).fetchone()

    assert row is not None, (
        f"Checkpoint {context.set_null_checkpoint_id_r!r} missing after resource deletion"
    )
    assert row[0] is None, (
        f"Expected checkpoint resource_id to be NULL after parent deletion, "
        f"got {row[0]!r}"
    )