learningcircuit--local-deep-research/tests/auth_tests/test_sqlcipher_settings.py

"""
Tests for SQLCipher settings and environment variable overrides.

These tests verify that default settings are correct and that
environment variables properly override them.
"""

import shutil
import tempfile
from pathlib import Path

import pytest

from src.local_deep_research.database.sqlcipher_utils import (
    DEFAULT_KDF_ITERATIONS,
    DEFAULT_PAGE_SIZE,
    DEFAULT_HMAC_ALGORITHM,
    DEFAULT_KDF_ALGORITHM,
    get_sqlcipher_settings,
    apply_cipher_defaults_before_key,
    apply_performance_pragmas,
    set_sqlcipher_key,
)
from src.local_deep_research.database.sqlcipher_compat import (
    get_sqlcipher_module,
)


@pytest.fixture
def temp_db_path():
    """Create a temporary database path."""
    temp_dir = tempfile.mkdtemp()
    db_path = Path(temp_dir) / "test_settings.db"
    yield db_path
    shutil.rmtree(temp_dir)


@pytest.fixture
def sqlcipher_module():
    """Get the SQLCipher module."""
    return get_sqlcipher_module()


@pytest.fixture
def clean_env(monkeypatch):
    """Remove all LDR_DB_* environment variables."""
    env_vars = [
        "LDR_DB_KDF_ITERATIONS",
        "LDR_DB_PAGE_SIZE",
        "LDR_DB_HMAC_ALGORITHM",
        "LDR_DB_KDF_ALGORITHM",
        "LDR_DB_CACHE_SIZE_MB",
        "LDR_DB_JOURNAL_MODE",
        "LDR_DB_SYNCHRONOUS",
    ]
    for var in env_vars:
        monkeypatch.delenv(var, raising=False)
    return monkeypatch


class TestDefaultSettings:
    """Tests for default SQLCipher settings."""

    def test_default_kdf_iterations(self, clean_env):
        """Verify default KDF iterations value."""
        settings = get_sqlcipher_settings()
        assert settings["kdf_iterations"] == DEFAULT_KDF_ITERATIONS
        assert settings["kdf_iterations"] == 256000, "Default should be 256000"

    def test_default_page_size(self, clean_env):
        """Verify default page size value."""
        settings = get_sqlcipher_settings()
        assert settings["page_size"] == DEFAULT_PAGE_SIZE
        assert settings["page_size"] == 16384, "Default should be 16384 (16KB)"

    def test_default_hmac_algorithm(self, clean_env):
        """Verify default HMAC algorithm value."""
        settings = get_sqlcipher_settings()
        assert settings["hmac_algorithm"] == DEFAULT_HMAC_ALGORITHM
        assert settings["hmac_algorithm"] == "HMAC_SHA512"

    def test_default_kdf_algorithm(self, clean_env):
        """Verify default KDF algorithm value."""
        settings = get_sqlcipher_settings()
        assert settings["kdf_algorithm"] == DEFAULT_KDF_ALGORITHM
        assert settings["kdf_algorithm"] == "PBKDF2_HMAC_SHA512"

    def test_settings_returns_all_required_keys(self, clean_env):
        """Verify settings dictionary has all required keys."""
        settings = get_sqlcipher_settings()
        required_keys = [
            "kdf_iterations",
            "page_size",
            "hmac_algorithm",
            "kdf_algorithm",
        ]
        for key in required_keys:
            assert key in settings, f"Missing key: {key}"

    def test_settings_values_are_correct_types(self, clean_env):
        """Verify settings values have correct types."""
        settings = get_sqlcipher_settings()
        assert isinstance(settings["kdf_iterations"], int)
        assert isinstance(settings["page_size"], int)
        assert isinstance(settings["hmac_algorithm"], str)
        assert isinstance(settings["kdf_algorithm"], str)


class TestEnvironmentOverrides:
    """Tests for environment variable overrides."""

    def test_kdf_iterations_override(self, monkeypatch):
        """Verify LDR_DB_KDF_ITERATIONS environment variable works."""
        monkeypatch.setenv("LDR_DB_KDF_ITERATIONS", "500000")
        settings = get_sqlcipher_settings()
        assert settings["kdf_iterations"] == 500000

    def test_page_size_override(self, monkeypatch):
        """Verify LDR_DB_PAGE_SIZE environment variable works."""
        monkeypatch.setenv("LDR_DB_PAGE_SIZE", "8192")
        settings = get_sqlcipher_settings()
        assert settings["page_size"] == 8192

    def test_hmac_algorithm_override(self, monkeypatch):
        """Verify LDR_DB_HMAC_ALGORITHM environment variable works."""
        monkeypatch.setenv("LDR_DB_HMAC_ALGORITHM", "HMAC_SHA256")
        settings = get_sqlcipher_settings()
        assert settings["hmac_algorithm"] == "HMAC_SHA256"

    def test_kdf_algorithm_override(self, monkeypatch):
        """Verify LDR_DB_KDF_ALGORITHM environment variable works."""
        monkeypatch.setenv("LDR_DB_KDF_ALGORITHM", "PBKDF2_HMAC_SHA256")
        settings = get_sqlcipher_settings()
        assert settings["kdf_algorithm"] == "PBKDF2_HMAC_SHA256"


class TestPerformancePragmaSettings:
    """Tests for performance pragma environment variable overrides."""

    def test_cache_size_override(
        self, sqlcipher_module, temp_db_path, monkeypatch
    ):
        """Verify LDR_DB_CACHE_SIZE_MB environment variable works."""
        monkeypatch.setenv("LDR_DB_CACHE_SIZE_MB", "128")

        conn = sqlcipher_module.connect(str(temp_db_path))
        cursor = conn.cursor()
        apply_cipher_defaults_before_key(cursor)
        set_sqlcipher_key(cursor, "testpassword")
        apply_performance_pragmas(cursor)
        cursor.close()

        # cache_size in KB (negative means KB)
        # 128 MB = 131072 KB, stored as -131072
        result = conn.execute("PRAGMA cache_size").fetchone()
        assert result is not None
        assert result[0] == -131072, (
            f"Expected -131072 (-128*1024), got {result[0]}"
        )
        conn.close()

    def test_journal_mode_override(
        self, sqlcipher_module, temp_db_path, monkeypatch
    ):
        """Verify LDR_DB_JOURNAL_MODE environment variable works."""
        monkeypatch.setenv("LDR_DB_JOURNAL_MODE", "DELETE")

        conn = sqlcipher_module.connect(str(temp_db_path))
        cursor = conn.cursor()
        apply_cipher_defaults_before_key(cursor)
        set_sqlcipher_key(cursor, "testpassword")
        apply_performance_pragmas(cursor)
        cursor.close()

        result = conn.execute("PRAGMA journal_mode").fetchone()
        assert result is not None
        assert result[0].lower() == "delete"
        conn.close()

    def test_synchronous_override(
        self, sqlcipher_module, temp_db_path, monkeypatch
    ):
        """Verify LDR_DB_SYNCHRONOUS environment variable works."""
        monkeypatch.setenv("LDR_DB_SYNCHRONOUS", "FULL")

        conn = sqlcipher_module.connect(str(temp_db_path))
        cursor = conn.cursor()
        apply_cipher_defaults_before_key(cursor)
        set_sqlcipher_key(cursor, "testpassword")
        apply_performance_pragmas(cursor)
        cursor.close()

        result = conn.execute("PRAGMA synchronous").fetchone()
        assert result is not None
        # SQLite returns synchronous as integer: 0=OFF, 1=NORMAL, 2=FULL, 3=EXTRA
        assert result[0] == 2, f"Expected 2 (FULL), got {result[0]}"
        conn.close()

    def test_default_journal_mode_is_wal(
        self, sqlcipher_module, temp_db_path, clean_env
    ):
        """Verify default journal mode is WAL."""
        conn = sqlcipher_module.connect(str(temp_db_path))
        cursor = conn.cursor()
        apply_cipher_defaults_before_key(cursor)
        set_sqlcipher_key(cursor, "testpassword")
        apply_performance_pragmas(cursor)
        cursor.close()

        result = conn.execute("PRAGMA journal_mode").fetchone()
        assert result is not None
        assert result[0].lower() == "wal", f"Expected WAL, got {result[0]}"
        conn.close()

    def test_default_temp_store_is_memory(
        self, sqlcipher_module, temp_db_path, clean_env
    ):
        """Verify temp_store is always set to MEMORY."""
        conn = sqlcipher_module.connect(str(temp_db_path))
        cursor = conn.cursor()
        apply_cipher_defaults_before_key(cursor)
        set_sqlcipher_key(cursor, "testpassword")
        apply_performance_pragmas(cursor)
        cursor.close()

        result = conn.execute("PRAGMA temp_store").fetchone()
        assert result is not None
        # temp_store: 0=DEFAULT, 1=FILE, 2=MEMORY
        assert result[0] == 2, f"Expected 2 (MEMORY), got {result[0]}"
        conn.close()

    def test_default_busy_timeout(
        self, sqlcipher_module, temp_db_path, clean_env
    ):
        """Verify busy_timeout is set to 10000ms."""
        conn = sqlcipher_module.connect(str(temp_db_path))
        cursor = conn.cursor()
        apply_cipher_defaults_before_key(cursor)
        set_sqlcipher_key(cursor, "testpassword")
        apply_performance_pragmas(cursor)
        cursor.close()

        result = conn.execute("PRAGMA busy_timeout").fetchone()
        assert result is not None
        assert result[0] == 10000, f"Expected 10000, got {result[0]}"
        conn.close()


class TestSettingsAppliedToDatabase:
    """Tests verifying settings are actually applied to databases."""

    def test_cipher_settings_applied_before_key(
        self, sqlcipher_module, temp_db_path, clean_env
    ):
        """Verify cipher_default_* settings are applied before key."""
        settings = get_sqlcipher_settings()

        conn = sqlcipher_module.connect(str(temp_db_path))
        cursor = conn.cursor()

        # Apply settings before key
        apply_cipher_defaults_before_key(cursor)
        set_sqlcipher_key(cursor, "testpassword")
        cursor.close()

        # Verify all cipher settings
        page_size = conn.execute("PRAGMA cipher_page_size").fetchone()[0]
        hmac_algo = conn.execute("PRAGMA cipher_hmac_algorithm").fetchone()[0]
        kdf_algo = conn.execute("PRAGMA cipher_kdf_algorithm").fetchone()[0]

        assert str(page_size) == str(settings["page_size"])
        assert hmac_algo == settings["hmac_algorithm"]
        assert kdf_algo == settings["kdf_algorithm"]

        conn.close()

    def test_multiple_env_overrides_work_together(
        self, sqlcipher_module, temp_db_path, monkeypatch
    ):
        """Verify multiple environment variable overrides work together."""
        monkeypatch.setenv("LDR_DB_PAGE_SIZE", "8192")
        monkeypatch.setenv("LDR_DB_KDF_ITERATIONS", "100000")
        monkeypatch.setenv("LDR_DB_CACHE_SIZE_MB", "32")

        settings = get_sqlcipher_settings()
        assert settings["page_size"] == 8192
        assert settings["kdf_iterations"] == 100000

        conn = sqlcipher_module.connect(str(temp_db_path))
        cursor = conn.cursor()
        apply_cipher_defaults_before_key(cursor)
        set_sqlcipher_key(cursor, "testpassword")
        apply_performance_pragmas(cursor)
        cursor.close()

        # Verify page size
        page_size = conn.execute("PRAGMA cipher_page_size").fetchone()[0]
        assert str(page_size) == "8192"

        # Verify cache size (32MB = 32768KB = -32768)
        cache_size = conn.execute("PRAGMA cache_size").fetchone()[0]
        assert cache_size == -32768

        conn.close()