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

"""Regression guard for the encrypted-DB CreateIndex defect.

Until v1.6.x, ``encrypted_db.create_user_database()`` only emitted
``CreateTable`` statements and never ``CreateIndex``. As a result every
model-declared index (``index=True``, ``unique=True``, explicit
``Index(...)`` in ``__table_args__``) was missing in user databases.

This test asserts that for every table in ``Base.metadata``, every named
index declared by the model is present in a freshly created encrypted
database — both column-level and table-level.
"""

from __future__ import annotations

import sys
import tempfile
from pathlib import Path

import pytest
from sqlalchemy import inspect

pytest.importorskip(
    "sqlcipher3",
    reason="SQLCipher is required to test the encrypted-DB schema path.",
)

sys.path.insert(
    0,
    str(Path(__file__).parent.parent.parent.resolve()),
)

from local_deep_research.database.encrypted_db import DatabaseManager
from local_deep_research.database.models import Base


@pytest.fixture
def temp_data_dir(monkeypatch):
    with tempfile.TemporaryDirectory() as tmpdir:
        path = Path(tmpdir)
        monkeypatch.setattr(
            "local_deep_research.database.encrypted_db.get_data_directory",
            lambda: path,
        )
        yield path


@pytest.fixture
def db_manager(temp_data_dir):
    m = DatabaseManager()
    yield m
    for username in list(m.connections.keys()):
        m.close_user_database(username)


def test_every_model_declared_index_exists_in_fresh_db(db_manager):
    """Iterate Base.metadata; assert every named index is present."""
    assert db_manager.has_encryption, (
        "sqlcipher3 imports but DatabaseManager reported has_encryption=False"
    )
    username = "indexcheck"
    password = "StrongPassword1!"
    db_manager.create_user_database(username, password)

    engine = db_manager.connections[username]
    inspector = inspect(engine)

    missing: list[tuple[str, str]] = []
    for table in Base.metadata.sorted_tables:
        if table.name == "users":
            continue
        if not inspector.has_table(table.name):
            continue
        existing = {idx["name"] for idx in inspector.get_indexes(table.name)}
        for index in table.indexes:
            if not index.name:
                continue
            if index.name not in existing:
                missing.append((table.name, index.name))

    assert not missing, (
        f"Missing model-declared indexes in fresh user DB: {missing}. "
        f"This usually means encrypted_db.create_user_database() is not "
        f"emitting CreateIndex for declared indexes."
    )


def test_download_tracker_url_hash_has_unique_backing(db_manager):
    """The exact regression that caused #3697 — url_hash must be UNIQUE.

    Without UNIQUE backing on the FK target, SQLCipher raises
    "foreign key mismatch — download_attempts referencing download_tracker"
    on cascade delete.
    """
    assert db_manager.has_encryption
    username = "trackercheck"
    password = "StrongPassword1!"
    db_manager.create_user_database(username, password)

    engine = db_manager.connections[username]
    inspector = inspect(engine)

    indexes = inspector.get_indexes("download_tracker")
    unique_on_url_hash = [
        idx
        for idx in indexes
        if idx.get("unique") and idx.get("column_names") == ["url_hash"]
    ]
    assert unique_on_url_hash, (
        "download_tracker.url_hash must have a UNIQUE backing index for "
        "FK references from download_attempts/download_duplicates to "
        f"resolve. Indexes present: {indexes}"
    )