"""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}" )