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

"""Tests for migration 0019: retire the 'both' egress scope.

Pins the upgrade semantics:
- ``policy.egress_scope`` rows holding ``both`` are rewritten to ``adaptive``;
  every other (concrete) scope is preserved untouched.
- The settings ``value`` column stores JSON text (``"both"`` with quotes); the
  rewritten value must be stored as ``"adaptive"`` (single JSON encoding) — a
  raw-bytes assertion guards against double-encoding (see migration 0018).
- Idempotency: re-running the upgrade is a no-op.
- A DB with no matching row upgrades cleanly (rowcount 0, no error).
"""

import json

import pytest
from alembic import command
from sqlalchemy import create_engine, text

from local_deep_research.database.alembic_runner import get_alembic_config

EGRESS_SCOPE_KEY = "policy.egress_scope"


def _run_upgrade_to(engine, revision):
    config = get_alembic_config(engine)
    with engine.begin() as conn:
        config.attributes["connection"] = conn
        command.upgrade(config, revision)


def _seed_setting(engine, key, value):
    with engine.begin() as conn:
        conn.execute(
            text(
                "INSERT INTO settings "
                "(key, value, type, name, ui_element, visible, editable) "
                "VALUES (:key, :value, 'app', :name, 'select', 1, 1)"
            ),
            {"key": key, "value": json.dumps(value), "name": key},
        )


def _read_setting(engine, key):
    with engine.begin() as conn:
        row = conn.execute(
            text("SELECT value FROM settings WHERE key = :key"),
            {"key": key},
        ).fetchone()
    if row is None:
        return None
    raw = row[0]
    return json.loads(raw) if isinstance(raw, str) else raw


def _read_setting_raw(engine, key):
    """Return the stored bytes of ``settings.value`` without decoding."""
    with engine.begin() as conn:
        row = conn.execute(
            text("SELECT value FROM settings WHERE key = :key"),
            {"key": key},
        ).fetchone()
    return None if row is None else row[0]


@pytest.fixture
def migrated_to_0018_engine(tmp_path):
    """Database fully migrated through 0018 (the revision before 0019)."""
    db_path = tmp_path / "test_0019.db"
    engine = create_engine(f"sqlite:///{db_path}")
    _run_upgrade_to(engine, "0018")
    yield engine
    engine.dispose()


class TestMigration0019RetireBoth:
    def test_both_rewritten_to_adaptive(self, migrated_to_0018_engine):
        engine = migrated_to_0018_engine
        _seed_setting(engine, EGRESS_SCOPE_KEY, "both")

        _run_upgrade_to(engine, "0019")

        assert _read_setting(engine, EGRESS_SCOPE_KEY) == "adaptive"

    def test_rewritten_value_stored_single_encoded(
        self, migrated_to_0018_engine
    ):
        """The stored bytes must be ``"adaptive"`` (single JSON encoding), not a
        double-encoded ``"\\"adaptive\\""``. This is what catches a WHERE-clause
        desync from the on-disk JSON form."""
        engine = migrated_to_0018_engine
        _seed_setting(engine, EGRESS_SCOPE_KEY, "both")

        _run_upgrade_to(engine, "0019")

        assert _read_setting_raw(engine, EGRESS_SCOPE_KEY) == json.dumps(
            "adaptive"
        )

    @pytest.mark.parametrize(
        "concrete",
        ["adaptive", "public_only", "private_only", "strict", "unprotected"],
    )
    def test_concrete_scope_preserved(self, migrated_to_0018_engine, concrete):
        engine = migrated_to_0018_engine
        _seed_setting(engine, EGRESS_SCOPE_KEY, concrete)

        _run_upgrade_to(engine, "0019")

        assert _read_setting(engine, EGRESS_SCOPE_KEY) == concrete

    def test_upgrade_is_idempotent(self, migrated_to_0018_engine):
        engine = migrated_to_0018_engine
        _seed_setting(engine, EGRESS_SCOPE_KEY, "both")

        _run_upgrade_to(engine, "0019")
        _run_upgrade_to(engine, "0019")

        assert _read_setting(engine, EGRESS_SCOPE_KEY) == "adaptive"

    def test_missing_row_is_a_clean_noop(self, migrated_to_0018_engine):
        engine = migrated_to_0018_engine

        _run_upgrade_to(engine, "0019")

        assert _read_setting(engine, EGRESS_SCOPE_KEY) is None


class TestMigration0019HeadAlignment:
    """Head-alignment guard. By convention this lives in the newest
    migration's test file (moved here from the 0018 test file when 0019 was
    added). Catches a new migration landing without the chain being wired up."""

    def test_0019_chains_correctly_to_0018(self):
        from alembic.config import Config
        from alembic.script import ScriptDirectory

        from local_deep_research.database.alembic_runner import (
            get_migrations_dir,
        )

        config = Config()
        config.set_main_option("script_location", str(get_migrations_dir()))
        script = ScriptDirectory.from_config(config)
        rev_0019 = script.get_revision("0019")
        assert rev_0019.down_revision == "0018"

    # NOTE: the head-alignment guard (assert head == latest) moved to the
    # newest migration's test file — test_migration_0020_add_zotero_tables.py.
    # 0019 is no longer the head, so asserting it here would break every
    # future migration.