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

"""
Tests for SQLCipher 4.6+ Community Edition compatibility.

These tests verify that the cipher_default_* pragmas are used correctly
and that the PRAGMA ordering (before/after key) is correct.
"""

import shutil
import tempfile
from pathlib import Path

import pytest

from src.local_deep_research.database.encrypted_db import DatabaseManager
from src.local_deep_research.database.sqlcipher_compat import (
    get_sqlcipher_module,
)
from src.local_deep_research.database.sqlcipher_utils import (
    apply_cipher_defaults_before_key,
    apply_sqlcipher_pragmas,
    get_sqlcipher_settings,
    set_sqlcipher_key,
)


@pytest.fixture
def temp_data_dir():
    """Create a temporary data directory for testing."""
    temp_dir = tempfile.mkdtemp()
    yield Path(temp_dir)
    shutil.rmtree(temp_dir)


@pytest.fixture
def temp_db_path(temp_data_dir):
    """Create a temporary database path."""
    return temp_data_dir / "test_cipher.db"


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


@pytest.fixture
def db_manager(temp_data_dir, monkeypatch):
    """Create a DatabaseManager with test configuration."""
    monkeypatch.setenv("LDR_DATA_DIR", str(temp_data_dir))
    manager = DatabaseManager()
    manager.data_dir = temp_data_dir / "encrypted_databases"
    manager.data_dir.mkdir(parents=True, exist_ok=True)
    return manager


@pytest.fixture
def auth_user(temp_data_dir, monkeypatch):
    """Create a test user in auth database."""
    monkeypatch.setenv("LDR_DATA_DIR", str(temp_data_dir))

    from src.local_deep_research.database.auth_db import (
        get_auth_db_session,
        init_auth_database,
    )
    from src.local_deep_research.database.models.auth import User

    init_auth_database()
    auth_db = get_auth_db_session()
    user = User(username="testuser")
    auth_db.add(user)
    auth_db.commit()
    auth_db.close()
    return user


class TestSQLCipher46Compatibility:
    """Tests for SQLCipher 4.6+ compatibility with cipher_default_* pragmas."""

    def test_cipher_default_page_size_applied(
        self, sqlcipher_module, temp_db_path
    ):
        """Verify cipher_default_page_size pragma is applied correctly."""
        settings = get_sqlcipher_settings()
        expected_page_size = settings["page_size"]

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

        # Apply cipher settings BEFORE key (SQLCipher 4.6+ requirement)
        apply_cipher_defaults_before_key(cursor)
        set_sqlcipher_key(cursor, "testpassword")
        cursor.close()

        # Verify the page size is applied
        result = conn.execute("PRAGMA cipher_page_size").fetchone()
        assert result is not None
        assert str(result[0]) == str(expected_page_size), (
            f"Expected cipher_page_size={expected_page_size}, got {result[0]}"
        )

        conn.close()

    def test_cipher_default_hmac_algorithm_applied(
        self, sqlcipher_module, temp_db_path
    ):
        """Verify cipher_default_hmac_algorithm pragma is applied correctly."""
        settings = get_sqlcipher_settings()
        expected_hmac = settings["hmac_algorithm"]

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

        apply_cipher_defaults_before_key(cursor)
        set_sqlcipher_key(cursor, "testpassword")
        cursor.close()

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

        conn.close()

    def test_cipher_default_kdf_algorithm_applied(
        self, sqlcipher_module, temp_db_path
    ):
        """Verify cipher_default_kdf_algorithm pragma is applied correctly."""
        settings = get_sqlcipher_settings()
        expected_kdf = settings["kdf_algorithm"]

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

        apply_cipher_defaults_before_key(cursor)
        set_sqlcipher_key(cursor, "testpassword")
        cursor.close()

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

        conn.close()

    def test_pragma_order_cipher_settings_before_key(
        self, sqlcipher_module, temp_db_path
    ):
        """Verify that cipher_default_* pragmas work when applied BEFORE key."""
        settings = get_sqlcipher_settings()

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

        # Correct order: cipher settings BEFORE key
        apply_cipher_defaults_before_key(cursor)
        set_sqlcipher_key(cursor, "testpassword")

        # Should be able to create tables and query
        conn.execute("CREATE TABLE test (id INTEGER PRIMARY KEY, value TEXT)")
        conn.execute("INSERT INTO test VALUES (1, 'hello')")
        result = conn.execute("SELECT * FROM test").fetchone()

        assert result == (1, "hello"), (
            "Database should be accessible with correct order"
        )

        # Verify settings were applied
        page_size = conn.execute("PRAGMA cipher_page_size").fetchone()[0]
        assert str(page_size) == str(settings["page_size"])

        cursor.close()
        conn.close()

    def test_pragma_order_kdf_iter_after_key(
        self, sqlcipher_module, temp_db_path
    ):
        """Verify that kdf_iter pragma works when applied AFTER key."""
        settings = get_sqlcipher_settings()

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

        # Correct order
        apply_cipher_defaults_before_key(cursor)
        set_sqlcipher_key(cursor, "testpassword")
        apply_sqlcipher_pragmas(cursor, creation_mode=True)
        cursor.close()

        # Verify kdf_iter was applied
        result = conn.execute("PRAGMA kdf_iter").fetchone()
        assert result is not None
        assert str(result[0]) == str(settings["kdf_iterations"]), (
            f"Expected kdf_iter={settings['kdf_iterations']}, got {result[0]}"
        )

        conn.close()

    def test_cipher_default_pragmas_set_correct_defaults(
        self, sqlcipher_module, temp_db_path
    ):
        """
        Verify that cipher_default_* pragmas properly configure the database.

        This test confirms that using cipher_default_page_size (not cipher_page_size)
        correctly sets the page size for new databases in SQLCipher 4.6+.
        """
        conn = sqlcipher_module.connect(str(temp_db_path))
        cursor = conn.cursor()

        # Use cipher_default_page_size (the correct way for SQLCipher 4.6+)
        cursor.execute("PRAGMA cipher_default_page_size = 16384")
        cursor.execute("PRAGMA key = 'testpassword'")
        cursor.close()

        # The page size should be 16384 as we set it
        result = conn.execute("PRAGMA cipher_page_size").fetchone()
        assert result[0] == "16384", (
            f"cipher_default_page_size should set page size to 16384, got {result[0]}"
        )

        # Create a table to ensure database is usable
        conn.execute("CREATE TABLE test (id INTEGER PRIMARY KEY)")
        conn.execute("INSERT INTO test VALUES (1)")
        result = conn.execute("SELECT * FROM test").fetchone()
        assert result == (1,), "Database should be functional"

        conn.close()

    def test_database_created_with_correct_settings(
        self, db_manager, auth_user
    ):
        """Verify database is created with all correct SQLCipher settings."""
        from sqlalchemy import text

        settings = get_sqlcipher_settings()
        engine = db_manager.create_user_database("testuser", "testpassword123")

        with engine.connect() as conn:
            # Check all cipher settings
            page_size = conn.execute(text("PRAGMA cipher_page_size")).scalar()
            hmac_algo = conn.execute(
                text("PRAGMA cipher_hmac_algorithm")
            ).scalar()
            kdf_algo = conn.execute(
                text("PRAGMA cipher_kdf_algorithm")
            ).scalar()
            kdf_iter = conn.execute(text("PRAGMA kdf_iter")).scalar()

            assert str(page_size) == str(settings["page_size"]), (
                f"cipher_page_size mismatch: expected {settings['page_size']}, got {page_size}"
            )
            assert hmac_algo == settings["hmac_algorithm"], (
                f"cipher_hmac_algorithm mismatch: expected {settings['hmac_algorithm']}, got {hmac_algo}"
            )
            assert kdf_algo == settings["kdf_algorithm"], (
                f"cipher_kdf_algorithm mismatch: expected {settings['kdf_algorithm']}, got {kdf_algo}"
            )
            assert str(kdf_iter) == str(settings["kdf_iterations"]), (
                f"kdf_iter mismatch: expected {settings['kdf_iterations']}, got {kdf_iter}"
            )

    def test_database_reopened_with_matching_settings(
        self, db_manager, auth_user
    ):
        """Verify database can be reopened and settings remain consistent."""
        from sqlalchemy import text

        settings = get_sqlcipher_settings()

        # Create database
        engine = db_manager.create_user_database("testuser", "testpassword123")

        # Create a test table and write some data
        with engine.connect() as conn:
            conn.execute(
                text(
                    "CREATE TABLE IF NOT EXISTS test_data (id INTEGER PRIMARY KEY, value TEXT)"
                )
            )
            conn.execute(
                text(
                    "INSERT INTO test_data (id, value) VALUES (1, 'test_value')"
                )
            )
            conn.commit()

        # Close the database
        db_manager.close_user_database("testuser")
        assert not db_manager.is_user_connected("testuser")

        # Reopen the database
        engine2 = db_manager.open_user_database("testuser", "testpassword123")
        assert engine2 is not None

        with engine2.connect() as conn:
            # Verify settings are still correct
            page_size = conn.execute(text("PRAGMA cipher_page_size")).scalar()
            hmac_algo = conn.execute(
                text("PRAGMA cipher_hmac_algorithm")
            ).scalar()
            kdf_algo = conn.execute(
                text("PRAGMA cipher_kdf_algorithm")
            ).scalar()

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

            # Verify data is still there
            result = conn.execute(
                text("SELECT value FROM test_data WHERE id = 1")
            ).scalar()
            assert result == "test_value", "Data should persist after reopen"


class TestSQLCipherVersionDetection:
    """Tests for SQLCipher version detection and compatibility."""

    def test_sqlcipher_version_is_4x(self, sqlcipher_module, temp_db_path):
        """Verify we're testing against SQLCipher 4.x."""
        conn = sqlcipher_module.connect(str(temp_db_path))
        cursor = conn.cursor()

        # Set key first (required to query version in some setups)
        cursor.execute("PRAGMA key = 'test'")

        result = conn.execute("PRAGMA cipher_version").fetchone()
        assert result is not None

        version = result[0]
        # Should be 4.x.x (e.g., "4.6.1 community")
        assert version.startswith("4."), (
            f"Expected SQLCipher 4.x, got {version}"
        )

        cursor.close()
        conn.close()

    def test_cipher_settings_function_returns_valid_settings(self):
        """Verify get_sqlcipher_settings returns 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 required setting: {key}"

        # Verify types
        assert isinstance(settings["kdf_iterations"], int)
        assert isinstance(settings["page_size"], int)
        assert isinstance(settings["hmac_algorithm"], str)
        assert isinstance(settings["kdf_algorithm"], str)

        # Verify reasonable values
        assert settings["kdf_iterations"] >= 100000, (
            "KDF iterations should be >= 100000"
        )
        assert settings["page_size"] in [
            1024,
            4096,
            8192,
            16384,
            32768,
            65536,
        ], "Page size should be a valid SQLite page size"