backend/tests/test_db_runtime_paths.py

"""
Regression tests for shared database path and schema helpers.
"""

import os
import sys

# Add backend to path
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

import db_utils
import runtime_paths
import settings


def test_runtime_db_paths_follow_shared_directory_overrides(tmp_path, monkeypatch):
    data_root = tmp_path / "data-root"
    db_root = tmp_path / "db-root"
    custom_settings = tmp_path / "custom" / "settings.sqlite"

    monkeypatch.setenv("SEARCH_UI_DATA_ROOT", str(data_root))
    monkeypatch.delenv("SEARCH_UI_DB_DIR", raising=False)
    monkeypatch.delenv("SEARCH_UI_SEARCH_DB_PATH", raising=False)
    monkeypatch.delenv("SEARCH_UI_IMAGE_DB_PATH", raising=False)
    monkeypatch.delenv("SEARCH_UI_FACE_DB_PATH", raising=False)
    monkeypatch.delenv("SEARCH_UI_SPEAKER_DB_PATH", raising=False)
    monkeypatch.delenv("SEARCH_UI_SETTINGS_DB_PATH", raising=False)

    assert runtime_paths.get_db_dir() == str(data_root)
    assert runtime_paths.get_search_db_path() == str(data_root / "database.db")
    assert runtime_paths.get_image_db_path() == str(data_root / "images_database.db")
    assert runtime_paths.get_face_db_path() == str(data_root / "faces_database.db")
    assert runtime_paths.get_speaker_db_path() == str(data_root / "speakers_database.db")
    assert runtime_paths.get_settings_db_path() == str(data_root / "settings.db")
    assert runtime_paths.get_publications_db_path() == str(data_root / "publications_database.db")
    assert runtime_paths.get_json_dir() == str(data_root / "json")
    assert runtime_paths.get_subtitles_dir() == str(data_root / "subtitles")
    assert runtime_paths.get_videos_dir() == str(data_root / "videos")
    assert runtime_paths.get_publications_dir() == str(data_root / "publications")
    assert runtime_paths.get_logs_dir() == str(data_root / "logs")
    assert runtime_paths.get_checkpoints_dir() == str(data_root / "checkpoints")

    monkeypatch.setenv("SEARCH_UI_DB_DIR", str(db_root))
    monkeypatch.setenv("SEARCH_UI_SETTINGS_DB_PATH", str(custom_settings))

    assert runtime_paths.get_db_dir() == str(db_root)
    assert runtime_paths.get_search_db_path() == str(db_root / "database.db")
    assert runtime_paths.get_settings_db_path() == str(custom_settings)


def test_connect_sqlite_creates_parent_dir_and_stores_schema_metadata(tmp_path):
    db_path = tmp_path / "nested" / "app.db"

    conn = db_utils.connect_sqlite(str(db_path))
    db_utils.register_schema_metadata(
        conn,
        db_name="unit_test_db",
        schema_version=7,
        embedding_type="binary",
        embedding_model="test-model-v1",
        embedding_recipe={"recipe_version": 1, "model_name": "test-model-v1"},
    )
    conn.commit()

    metadata = db_utils.fetch_schema_metadata(conn, "unit_test_db")
    conn.close()

    assert db_path.exists()
    assert metadata == {
        "db_name": "unit_test_db",
        "schema_version": 7,
        "embedding_type": "binary",
        "embedding_model": "test-model-v1",
        "embedding_recipe": {"recipe_version": 1, "model_name": "test-model-v1"},
        "created_at": metadata["created_at"],
    }
    assert metadata["created_at"]


def test_register_schema_metadata_migrates_existing_table_for_recipe(tmp_path):
    db_path = tmp_path / "legacy.db"
    conn = db_utils.connect_sqlite(str(db_path))
    conn.execute(
        """
        CREATE TABLE _schema_metadata (
            db_name TEXT PRIMARY KEY,
            schema_version INTEGER NOT NULL DEFAULT 1,
            embedding_type TEXT,
            embedding_model TEXT,
            created_at TEXT
        )
        """
    )
    conn.commit()

    db_utils.register_schema_metadata(
        conn,
        db_name="legacy_db",
        schema_version=2,
        embedding_type="binary",
        embedding_model="legacy-model",
        embedding_recipe={"recipe_version": 9},
    )
    conn.commit()

    metadata = db_utils.fetch_schema_metadata(conn, "legacy_db")
    columns = {
        row[1]
        for row in conn.execute("PRAGMA table_info(_schema_metadata)").fetchall()
    }
    conn.close()

    assert "embedding_recipe" in columns
    assert metadata["embedding_recipe"] == {"recipe_version": 9}


def test_settings_init_db_tracks_initialized_paths_per_database(tmp_path, monkeypatch):
    first_path = tmp_path / "settings-a" / "settings.db"
    second_path = tmp_path / "settings-b" / "settings.db"

    monkeypatch.setattr(settings, "_initialized_db_paths", set())
    monkeypatch.setenv("SEARCH_UI_SETTINGS_DB_PATH", str(first_path))
    settings.init_settings_db()

    monkeypatch.setenv("SEARCH_UI_SETTINGS_DB_PATH", str(second_path))
    settings.init_settings_db()

    assert first_path.exists()
    assert second_path.exists()

    first_conn = db_utils.connect_sqlite(str(first_path))
    second_conn = db_utils.connect_sqlite(str(second_path))
    try:
        first_tables = {
            row[0]
            for row in first_conn.execute(
                "SELECT name FROM sqlite_master WHERE type = 'table'"
            ).fetchall()
        }
        second_tables = {
            row[0]
            for row in second_conn.execute(
                "SELECT name FROM sqlite_master WHERE type = 'table'"
            ).fetchall()
        }
    finally:
        first_conn.close()
        second_conn.close()

    assert "global_settings" in first_tables
    assert "global_settings" in second_tables
    assert settings._initialized_db_paths == {str(first_path), str(second_path)}