learningcircuit--local-deep-research/tests/database/test_research_strategy_fk_regression.py

"""Regression tests for the ResearchStrategy FK retarget (migration 0008).

Pre-fix: ``ResearchStrategy.research_id`` was ``Integer FK research.id``,
pointing at a dormant table that no production path uses. The live writer
``save_research_strategy`` passes ``research_history`` UUID strings, so
every commit raised ``FOREIGN KEY constraint failed`` once v1.6.0 enabled
``PRAGMA foreign_keys``.

Post-fix: the FK targets ``research_history.id`` (String(36)) with cascade
delete. This module verifies both the schema-level repair and the live
behavior the production code depends on.
"""

from __future__ import annotations

import pytest
from sqlalchemy import create_engine, event, inspect, text
from sqlalchemy.orm import Session

from local_deep_research.database.alembic_runner import run_migrations
from local_deep_research.database.models.research import ResearchStrategy


@pytest.fixture
def fk_enforced_engine(tmp_path):
    """Migrated engine with PRAGMA foreign_keys=ON for every connection.

    Mirrors the production hook ``apply_performance_pragmas`` so the test
    actually exercises FK enforcement (off by default in a bare SQLite
    connection).
    """
    db_path = tmp_path / "research_strategy_fk_regression.db"
    engine = create_engine(f"sqlite:///{db_path}")

    @event.listens_for(engine, "connect")
    def _enable_fk(dbapi_connection, _):
        dbapi_connection.execute("PRAGMA foreign_keys = ON")

    run_migrations(engine)
    yield engine
    engine.dispose()


def _seed_research_history(engine, research_id: str) -> None:
    with engine.begin() as conn:
        conn.execute(
            text(
                "INSERT INTO research_history (id, query, mode, status, created_at) "
                "VALUES (:id, 'q', 'quick_summary', 'completed', '2026-01-01')"
            ),
            {"id": research_id},
        )


class TestResearchStrategyFKTarget:
    """Schema-level: column type and FK target match the live writer."""

    def test_research_id_is_varchar_36_fk_research_history(
        self, fk_enforced_engine
    ):
        inspector = inspect(fk_enforced_engine)
        cols = {
            c["name"]: c for c in inspector.get_columns("research_strategies")
        }
        assert "VARCHAR" in str(cols["research_id"]["type"]) or "CHAR" in str(
            cols["research_id"]["type"]
        )

        fks = inspector.get_foreign_keys("research_strategies")
        research_id_fks = [
            fk for fk in fks if "research_id" in fk["constrained_columns"]
        ]
        assert len(research_id_fks) == 1
        assert research_id_fks[0]["referred_table"] == "research_history"
        assert research_id_fks[0]["referred_columns"] == ["id"]

    def test_cascade_on_delete(self, fk_enforced_engine):
        # Read the on_delete action from PRAGMA directly. SQLAlchemy's SQLite
        # reflection has historically been inconsistent about populating
        # ``options.ondelete``; PRAGMA always returns it.
        with fk_enforced_engine.connect() as conn:
            rows = conn.execute(
                text("PRAGMA foreign_key_list(research_strategies)")
            ).fetchall()
        research_id_fk = next(row for row in rows if row[3] == "research_id")
        # PRAGMA columns: id, seq, table, from, to, on_update, on_delete, match
        assert research_id_fk[6] == "CASCADE"


class TestResearchStrategyLiveWrite:
    """Behavior: the live save path no longer raises FK errors."""

    def test_save_strategy_with_research_history_uuid_succeeds(
        self, fk_enforced_engine
    ):
        """The exact pattern from save_research_strategy: insert by UUID."""
        rid = "11111111-1111-1111-1111-111111111111"
        _seed_research_history(fk_enforced_engine, rid)

        with Session(fk_enforced_engine) as session:
            session.add(
                ResearchStrategy(
                    research_id=rid, strategy_name="langgraph-agent"
                )
            )
            session.commit()

        with Session(fk_enforced_engine) as session:
            stored = (
                session.query(ResearchStrategy).filter_by(research_id=rid).one()
            )
            assert stored.strategy_name == "langgraph-agent"

    def test_orphan_strategy_insert_is_rejected(self, fk_enforced_engine):
        """FK enforcement is real: insert without parent row must fail."""
        from sqlalchemy.exc import IntegrityError

        with Session(fk_enforced_engine) as session:
            session.add(
                ResearchStrategy(
                    research_id="22222222-2222-2222-2222-222222222222",
                    strategy_name="x",
                )
            )
            with pytest.raises(IntegrityError):
                session.commit()

    def test_cascade_delete_removes_strategy(self, fk_enforced_engine):
        rid = "33333333-3333-3333-3333-333333333333"
        _seed_research_history(fk_enforced_engine, rid)

        with Session(fk_enforced_engine) as session:
            session.add(
                ResearchStrategy(research_id=rid, strategy_name="standard")
            )
            session.commit()

        with fk_enforced_engine.begin() as conn:
            conn.execute(
                text("DELETE FROM research_history WHERE id = :id"),
                {"id": rid},
            )

        with Session(fk_enforced_engine) as session:
            assert (
                session.query(ResearchStrategy)
                .filter_by(research_id=rid)
                .first()
                is None
            )