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

"""
Tests for SQLCipher backwards compatibility.

These tests verify that databases can be created, closed, and reopened
correctly across multiple sessions, ensuring data persistence and
settings consistency.
"""

import shutil
import tempfile
from pathlib import Path

import pytest

from src.local_deep_research.database.sqlcipher_utils import (
    apply_cipher_defaults_before_key,
    apply_performance_pragmas,
    apply_sqlcipher_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_compat.db"
    yield db_path
    shutil.rmtree(temp_dir)


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


def create_configured_connection(sqlcipher_module, db_path, password):
    """Helper to create a properly configured connection.

    Uses the correct PRAGMA ordering:
    - New DB: cipher_default_* -> key -> kdf_iter -> performance
    - Existing DB: key -> cipher_* + kdf_iter -> performance
    """
    creation_mode = not db_path.exists()
    conn = sqlcipher_module.connect(str(db_path))
    cursor = conn.cursor()
    if creation_mode:
        apply_cipher_defaults_before_key(cursor)
    set_sqlcipher_key(cursor, password)
    apply_sqlcipher_pragmas(cursor, creation_mode=creation_mode)
    apply_performance_pragmas(cursor)
    cursor.close()
    return conn


class TestCreateCloseReopen:
    """Tests for create -> close -> reopen cycles."""

    def test_create_and_reopen_same_session(
        self, sqlcipher_module, temp_db_path
    ):
        """Verify database can be created and reopened within same test."""
        password = "test_password_123"

        # Create database
        conn1 = create_configured_connection(
            sqlcipher_module, temp_db_path, password
        )
        conn1.execute("CREATE TABLE test (id INTEGER PRIMARY KEY, data TEXT)")
        conn1.execute("INSERT INTO test VALUES (1, 'test_data')")
        conn1.commit()
        conn1.close()

        # Reopen database
        conn2 = create_configured_connection(
            sqlcipher_module, temp_db_path, password
        )
        result = conn2.execute("SELECT data FROM test WHERE id = 1").fetchone()
        assert result[0] == "test_data", (
            "Data should persist after close/reopen"
        )
        conn2.close()

    def test_create_reopen_multiple_times(self, sqlcipher_module, temp_db_path):
        """Verify database can be opened and closed multiple times."""
        password = "test_password_456"

        # Create database
        conn = create_configured_connection(
            sqlcipher_module, temp_db_path, password
        )
        conn.execute(
            "CREATE TABLE counter (id INTEGER PRIMARY KEY, count INTEGER)"
        )
        conn.execute("INSERT INTO counter VALUES (1, 0)")
        conn.commit()
        conn.close()

        # Open and close 5 times, incrementing counter each time
        for i in range(5):
            conn = create_configured_connection(
                sqlcipher_module, temp_db_path, password
            )
            conn.execute("UPDATE counter SET count = count + 1 WHERE id = 1")
            conn.commit()
            conn.close()

        # Final open to verify counter
        conn = create_configured_connection(
            sqlcipher_module, temp_db_path, password
        )
        result = conn.execute(
            "SELECT count FROM counter WHERE id = 1"
        ).fetchone()
        assert result[0] == 5, (
            f"Counter should be 5 after 5 increments, got {result[0]}"
        )
        conn.close()

    def test_data_persists_across_sessions(
        self, sqlcipher_module, temp_db_path
    ):
        """Verify complex data persists correctly across sessions."""
        password = "persistence_test"

        # Session 1: Create database with multiple tables and data
        conn1 = create_configured_connection(
            sqlcipher_module, temp_db_path, password
        )
        conn1.execute("""
            CREATE TABLE users (
                id INTEGER PRIMARY KEY,
                username TEXT UNIQUE,
                email TEXT
            )
        """)
        conn1.execute("""
            CREATE TABLE settings (
                user_id INTEGER,
                key TEXT,
                value TEXT,
                FOREIGN KEY(user_id) REFERENCES users(id)
            )
        """)

        # Insert test data
        conn1.execute(
            "INSERT INTO users VALUES (1, 'alice', 'alice@example.com')"
        )
        conn1.execute("INSERT INTO users VALUES (2, 'bob', 'bob@example.com')")
        conn1.execute("INSERT INTO settings VALUES (1, 'theme', 'dark')")
        conn1.execute("INSERT INTO settings VALUES (1, 'language', 'en')")
        conn1.execute("INSERT INTO settings VALUES (2, 'theme', 'light')")
        conn1.commit()
        conn1.close()

        # Session 2: Verify all data persists
        conn2 = create_configured_connection(
            sqlcipher_module, temp_db_path, password
        )

        # Check users
        users = conn2.execute(
            "SELECT username, email FROM users ORDER BY id"
        ).fetchall()
        assert users == [
            ("alice", "alice@example.com"),
            ("bob", "bob@example.com"),
        ]

        # Check settings
        alice_settings = conn2.execute(
            "SELECT key, value FROM settings WHERE user_id = 1 ORDER BY key"
        ).fetchall()
        assert alice_settings == [("language", "en"), ("theme", "dark")]

        bob_settings = conn2.execute(
            "SELECT key, value FROM settings WHERE user_id = 2"
        ).fetchall()
        assert bob_settings == [("theme", "light")]

        conn2.close()

    def test_settings_match_after_reopen(self, sqlcipher_module, temp_db_path):
        """Verify SQLCipher settings remain consistent after reopen."""
        password = "settings_test"

        # Create database
        conn1 = create_configured_connection(
            sqlcipher_module, temp_db_path, password
        )
        conn1.execute("CREATE TABLE test (id INTEGER)")
        conn1.commit()

        # Capture settings from first connection
        original_page_size = conn1.execute(
            "PRAGMA cipher_page_size"
        ).fetchone()[0]
        original_hmac = conn1.execute(
            "PRAGMA cipher_hmac_algorithm"
        ).fetchone()[0]
        original_kdf = conn1.execute("PRAGMA cipher_kdf_algorithm").fetchone()[
            0
        ]
        conn1.close()

        # Reopen and verify settings
        conn2 = create_configured_connection(
            sqlcipher_module, temp_db_path, password
        )

        new_page_size = conn2.execute("PRAGMA cipher_page_size").fetchone()[0]
        new_hmac = conn2.execute("PRAGMA cipher_hmac_algorithm").fetchone()[0]
        new_kdf = conn2.execute("PRAGMA cipher_kdf_algorithm").fetchone()[0]

        assert str(new_page_size) == str(original_page_size), (
            f"Page size changed: {original_page_size} -> {new_page_size}"
        )
        assert new_hmac == original_hmac, (
            f"HMAC algorithm changed: {original_hmac} -> {new_hmac}"
        )
        assert new_kdf == original_kdf, (
            f"KDF algorithm changed: {original_kdf} -> {new_kdf}"
        )

        conn2.close()


class TestDatabaseMigration:
    """Tests for database schema changes across sessions."""

    def test_schema_migration(self, sqlcipher_module, temp_db_path):
        """Verify schema can be modified across sessions."""
        password = "migration_test"

        # Session 1: Create initial schema
        conn1 = create_configured_connection(
            sqlcipher_module, temp_db_path, password
        )
        conn1.execute("CREATE TABLE data (id INTEGER PRIMARY KEY, value TEXT)")
        conn1.execute("INSERT INTO data VALUES (1, 'initial')")
        conn1.commit()
        conn1.close()

        # Session 2: Modify schema (add column)
        conn2 = create_configured_connection(
            sqlcipher_module, temp_db_path, password
        )
        conn2.execute("ALTER TABLE data ADD COLUMN created_at TEXT")
        conn2.execute("UPDATE data SET created_at = '2024-01-01' WHERE id = 1")
        conn2.commit()
        conn2.close()

        # Session 3: Verify modified schema
        conn3 = create_configured_connection(
            sqlcipher_module, temp_db_path, password
        )
        result = conn3.execute(
            "SELECT value, created_at FROM data WHERE id = 1"
        ).fetchone()
        assert result == ("initial", "2024-01-01")
        conn3.close()

    def test_create_index_persists(self, sqlcipher_module, temp_db_path):
        """Verify indexes persist across sessions."""
        password = "index_test"

        # Session 1: Create table and index
        conn1 = create_configured_connection(
            sqlcipher_module, temp_db_path, password
        )
        conn1.execute("CREATE TABLE users (id INTEGER PRIMARY KEY, email TEXT)")
        conn1.execute("CREATE INDEX idx_email ON users(email)")
        conn1.commit()
        conn1.close()

        # Session 2: Verify index exists
        conn2 = create_configured_connection(
            sqlcipher_module, temp_db_path, password
        )
        indexes = conn2.execute(
            "SELECT name FROM sqlite_master WHERE type='index' AND tbl_name='users'"
        ).fetchall()

        index_names = [idx[0] for idx in indexes]
        assert "idx_email" in index_names, (
            f"Index not found. Indexes: {index_names}"
        )
        conn2.close()


class TestLargeDataPersistence:
    """Tests for large data persistence."""

    def test_large_blob_persists(self, sqlcipher_module, temp_db_path):
        """Verify large binary data persists correctly."""
        password = "blob_test"
        import os

        # Create 1MB of random data
        large_data = os.urandom(1024 * 1024)

        # Session 1: Store large blob
        conn1 = create_configured_connection(
            sqlcipher_module, temp_db_path, password
        )
        conn1.execute("CREATE TABLE blobs (id INTEGER PRIMARY KEY, data BLOB)")
        conn1.execute("INSERT INTO blobs VALUES (1, ?)", (large_data,))
        conn1.commit()
        conn1.close()

        # Session 2: Retrieve and verify
        conn2 = create_configured_connection(
            sqlcipher_module, temp_db_path, password
        )
        result = conn2.execute("SELECT data FROM blobs WHERE id = 1").fetchone()
        assert result[0] == large_data, "Large blob data corrupted or changed"
        conn2.close()

    def test_many_rows_persist(self, sqlcipher_module, temp_db_path):
        """Verify many rows persist correctly."""
        password = "many_rows_test"

        # Session 1: Insert 1000 rows
        conn1 = create_configured_connection(
            sqlcipher_module, temp_db_path, password
        )
        conn1.execute(
            "CREATE TABLE numbers (id INTEGER PRIMARY KEY, value INTEGER)"
        )

        for i in range(1000):
            conn1.execute("INSERT INTO numbers VALUES (?, ?)", (i, i * 2))
        conn1.commit()
        conn1.close()

        # Session 2: Verify all rows
        conn2 = create_configured_connection(
            sqlcipher_module, temp_db_path, password
        )
        count = conn2.execute("SELECT COUNT(*) FROM numbers").fetchone()[0]
        assert count == 1000, f"Expected 1000 rows, got {count}"

        # Verify some specific values
        for test_id in [0, 499, 999]:
            result = conn2.execute(
                "SELECT value FROM numbers WHERE id = ?", (test_id,)
            ).fetchone()
            assert result[0] == test_id * 2, (
                f"Row {test_id} has wrong value: {result[0]} != {test_id * 2}"
            )

        conn2.close()


class TestOldToNewPragmaOrderMigration:
    """
    Tests for backwards compatibility when migrating from OLD pragma order
    (key first, then cipher settings) to NEW pragma order (cipher_default_*
    before key for creation, cipher_* after key for existing).
    """

    def _create_connection_old_order(self, sqlcipher_module, db_path, password):
        """
        Create a connection using the OLD pragma order (pre-PR behavior).

        OLD order: key -> cipher_page_size -> cipher_hmac_algorithm -> kdf_iter
        This simulates how databases were created before this PR.
        """
        from src.local_deep_research.database.sqlcipher_utils import (
            get_sqlcipher_settings,
            set_sqlcipher_key,
        )

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

        # OLD ORDER: Set key FIRST
        set_sqlcipher_key(cursor, password)

        # OLD ORDER: Then set cipher settings AFTER the key
        settings = get_sqlcipher_settings()
        cursor.execute(f"PRAGMA cipher_page_size = {settings['page_size']}")
        cursor.execute(
            f"PRAGMA cipher_hmac_algorithm = {settings['hmac_algorithm']}"
        )
        cursor.execute(f"PRAGMA kdf_iter = {settings['kdf_iterations']}")

        cursor.close()
        return conn

    def _create_connection_new_order(self, sqlcipher_module, db_path, password):
        """
        Create a connection using the NEW pragma order (post-PR behavior).
        """
        return create_configured_connection(sqlcipher_module, db_path, password)

    def test_old_db_opens_with_new_code(self, sqlcipher_module, temp_db_path):
        """
        CRITICAL: Verify database created with OLD pragma order can be
        opened with NEW pragma order.
        """
        password = "old_to_new_migration"

        # Create database using OLD pragma order (simulating pre-PR code)
        conn1 = self._create_connection_old_order(
            sqlcipher_module, temp_db_path, password
        )
        conn1.execute(
            "CREATE TABLE migration_test (id INTEGER PRIMARY KEY, data TEXT)"
        )
        conn1.execute(
            "INSERT INTO migration_test VALUES (1, 'created_with_old_code')"
        )
        conn1.execute("INSERT INTO migration_test VALUES (2, 'should_persist')")
        conn1.commit()
        conn1.close()

        # Reopen using NEW pragma order (simulating post-PR code)
        conn2 = self._create_connection_new_order(
            sqlcipher_module, temp_db_path, password
        )

        # Verify data is accessible
        result = conn2.execute(
            "SELECT data FROM migration_test WHERE id = 1"
        ).fetchone()
        assert result[0] == "created_with_old_code", (
            "Data created with old pragma order should be readable with new order"
        )

        # Verify all rows
        all_rows = conn2.execute(
            "SELECT id, data FROM migration_test ORDER BY id"
        ).fetchall()
        assert len(all_rows) == 2, f"Expected 2 rows, got {len(all_rows)}"
        assert all_rows[0] == (1, "created_with_old_code")
        assert all_rows[1] == (2, "should_persist")

        conn2.close()

    def test_old_db_can_be_modified_with_new_code(
        self, sqlcipher_module, temp_db_path
    ):
        """
        Verify database created with OLD pragma order can be modified
        after opening with NEW pragma order.
        """
        password = "old_to_new_modify"

        # Create database using OLD pragma order
        conn1 = self._create_connection_old_order(
            sqlcipher_module, temp_db_path, password
        )
        conn1.execute(
            "CREATE TABLE data (id INTEGER PRIMARY KEY, value INTEGER)"
        )
        conn1.execute("INSERT INTO data VALUES (1, 100)")
        conn1.commit()
        conn1.close()

        # Open with NEW pragma order and modify
        conn2 = self._create_connection_new_order(
            sqlcipher_module, temp_db_path, password
        )
        conn2.execute("UPDATE data SET value = 200 WHERE id = 1")
        conn2.execute("INSERT INTO data VALUES (2, 300)")
        conn2.commit()
        conn2.close()

        # Reopen with NEW pragma order and verify changes persisted
        conn3 = self._create_connection_new_order(
            sqlcipher_module, temp_db_path, password
        )
        results = conn3.execute(
            "SELECT id, value FROM data ORDER BY id"
        ).fetchall()

        assert results == [(1, 200), (2, 300)], (
            f"Modifications should persist. Got: {results}"
        )
        conn3.close()

    def test_old_db_schema_migration_with_new_code(
        self, sqlcipher_module, temp_db_path
    ):
        """
        Verify schema changes work on databases created with OLD pragma order
        when opened with NEW pragma order.
        """
        password = "old_to_new_schema"

        # Create database using OLD pragma order
        conn1 = self._create_connection_old_order(
            sqlcipher_module, temp_db_path, password
        )
        conn1.execute("CREATE TABLE users (id INTEGER PRIMARY KEY, name TEXT)")
        conn1.execute("INSERT INTO users VALUES (1, 'Alice')")
        conn1.commit()
        conn1.close()

        # Open with NEW pragma order and modify schema
        conn2 = self._create_connection_new_order(
            sqlcipher_module, temp_db_path, password
        )
        conn2.execute("ALTER TABLE users ADD COLUMN email TEXT")
        conn2.execute(
            "UPDATE users SET email = 'alice@example.com' WHERE id = 1"
        )
        conn2.execute("CREATE INDEX idx_users_email ON users(email)")
        conn2.commit()
        conn2.close()

        # Verify schema changes persisted
        conn3 = self._create_connection_new_order(
            sqlcipher_module, temp_db_path, password
        )

        # Check data
        result = conn3.execute(
            "SELECT name, email FROM users WHERE id = 1"
        ).fetchone()
        assert result == ("Alice", "alice@example.com")

        # Check index exists
        indexes = conn3.execute(
            "SELECT name FROM sqlite_master WHERE type='index' AND tbl_name='users'"
        ).fetchall()
        index_names = [idx[0] for idx in indexes]
        assert "idx_users_email" in index_names, (
            f"Index should persist. Found indexes: {index_names}"
        )

        conn3.close()

    def test_settings_consistent_across_migration(
        self, sqlcipher_module, temp_db_path
    ):
        """
        Verify SQLCipher settings remain consistent when migrating from
        OLD to NEW pragma order.
        """
        password = "settings_migration"

        # Create database using OLD pragma order
        conn1 = self._create_connection_old_order(
            sqlcipher_module, temp_db_path, password
        )
        conn1.execute("CREATE TABLE test (id INTEGER)")
        conn1.commit()

        # Capture settings from old-order connection
        old_page_size = conn1.execute("PRAGMA cipher_page_size").fetchone()[0]
        old_hmac = conn1.execute("PRAGMA cipher_hmac_algorithm").fetchone()[0]
        conn1.close()

        # Open with NEW pragma order
        conn2 = self._create_connection_new_order(
            sqlcipher_module, temp_db_path, password
        )

        # Verify settings match
        new_page_size = conn2.execute("PRAGMA cipher_page_size").fetchone()[0]
        new_hmac = conn2.execute("PRAGMA cipher_hmac_algorithm").fetchone()[0]

        assert str(new_page_size) == str(old_page_size), (
            f"Page size mismatch after migration: {old_page_size} -> {new_page_size}"
        )
        assert new_hmac == old_hmac, (
            f"HMAC algorithm mismatch after migration: {old_hmac} -> {new_hmac}"
        )

        conn2.close()

    def test_rekey_on_old_db_with_new_code(
        self, sqlcipher_module, temp_db_path
    ):
        """
        Verify password change (rekey) works on databases created with
        OLD pragma order when using NEW pragma order code.
        """
        from src.local_deep_research.database.sqlcipher_utils import (
            set_sqlcipher_rekey,
        )

        old_password = "original_password"
        new_password = "changed_password"

        # Create database using OLD pragma order
        conn1 = self._create_connection_old_order(
            sqlcipher_module, temp_db_path, old_password
        )
        conn1.execute(
            "CREATE TABLE secrets (id INTEGER PRIMARY KEY, data TEXT)"
        )
        conn1.execute("INSERT INTO secrets VALUES (1, 'sensitive_data')")
        conn1.commit()
        conn1.close()

        # Open with NEW pragma order and change password
        conn2 = self._create_connection_new_order(
            sqlcipher_module, temp_db_path, old_password
        )

        # Verify data is accessible before rekey
        result = conn2.execute(
            "SELECT data FROM secrets WHERE id = 1"
        ).fetchone()
        assert result[0] == "sensitive_data"

        # Change the password
        set_sqlcipher_rekey(conn2, new_password)
        conn2.commit()
        conn2.close()

        # Verify old password no longer works
        try:
            conn_fail = self._create_connection_new_order(
                sqlcipher_module, temp_db_path, old_password
            )
            # If we get here, try to access data (should fail)
            conn_fail.execute("SELECT * FROM secrets").fetchone()
            conn_fail.close()
            raise AssertionError("Old password should not work after rekey")
        except Exception as e:
            if isinstance(e, AssertionError):
                raise
            # Any other exception is expected (file is not a database, etc.)

        # Verify new password works
        conn3 = self._create_connection_new_order(
            sqlcipher_module, temp_db_path, new_password
        )
        result = conn3.execute(
            "SELECT data FROM secrets WHERE id = 1"
        ).fetchone()
        assert result[0] == "sensitive_data", (
            "Data should be accessible with new password after rekey"
        )
        conn3.close()