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

"""
Tests for SQLCipher performance pragmas.

These tests verify that performance-related PRAGMA settings are
correctly applied for optimal database performance.
"""

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_performance.db"
    yield db_path
    shutil.rmtree(temp_dir)


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


@pytest.fixture
def configured_connection(sqlcipher_module, temp_db_path):
    """Create a fully configured database connection."""
    conn = sqlcipher_module.connect(str(temp_db_path))
    cursor = conn.cursor()
    # New database: cipher_default_* before key
    apply_cipher_defaults_before_key(cursor)
    set_sqlcipher_key(cursor, "testpassword")
    apply_sqlcipher_pragmas(cursor, creation_mode=True)
    apply_performance_pragmas(cursor)
    cursor.close()
    return conn


class TestWALMode:
    """Tests for WAL (Write-Ahead Logging) mode."""

    def test_wal_mode_enabled(self, configured_connection):
        """Verify WAL journal mode is enabled by default."""
        result = configured_connection.execute("PRAGMA journal_mode").fetchone()
        assert result is not None
        assert result[0].lower() == "wal", f"Expected WAL mode, got {result[0]}"
        configured_connection.close()

    def test_wal_creates_additional_files(self, sqlcipher_module, temp_db_path):
        """Verify WAL mode creates -wal and -shm files."""
        conn = sqlcipher_module.connect(str(temp_db_path))
        cursor = conn.cursor()
        # New database: cipher_default_* before key
        apply_cipher_defaults_before_key(cursor)
        set_sqlcipher_key(cursor, "testpassword")
        apply_performance_pragmas(cursor)
        cursor.close()

        # Create a table and write data to trigger WAL file creation
        conn.execute("CREATE TABLE test (id INTEGER PRIMARY KEY, data TEXT)")
        conn.execute("INSERT INTO test VALUES (1, 'test_data')")
        conn.commit()

        # Note: WAL files (-wal, -shm) may not exist immediately or may be cleaned up
        # The main verification is that WAL mode is enabled
        assert (
            conn.execute("PRAGMA journal_mode").fetchone()[0].lower() == "wal"
        )
        conn.close()

    def test_concurrent_reads_with_wal(self, sqlcipher_module, temp_db_path):
        """Verify WAL mode allows concurrent read connections."""
        # Create and populate database
        conn1 = sqlcipher_module.connect(str(temp_db_path))
        cursor1 = conn1.cursor()
        # New database: cipher_default_* before key
        apply_cipher_defaults_before_key(cursor1)
        set_sqlcipher_key(cursor1, "testpassword")
        apply_performance_pragmas(cursor1)
        cursor1.close()

        conn1.execute("CREATE TABLE test (id INTEGER PRIMARY KEY, data TEXT)")
        conn1.execute("INSERT INTO test VALUES (1, 'data1')")
        conn1.commit()

        # Open second read connection while first is open
        conn2 = sqlcipher_module.connect(str(temp_db_path))
        cursor2 = conn2.cursor()
        # Existing database: key first, then cipher_* pragmas
        set_sqlcipher_key(cursor2, "testpassword")
        apply_sqlcipher_pragmas(cursor2, creation_mode=False)
        apply_performance_pragmas(cursor2)
        cursor2.close()

        # Both connections should be able to read
        result1 = conn1.execute("SELECT * FROM test").fetchall()
        result2 = conn2.execute("SELECT * FROM test").fetchall()

        assert result1 == result2 == [(1, "data1")]

        conn2.close()
        conn1.close()


class TestCacheSettings:
    """Tests for cache size settings."""

    def test_cache_size_applied(self, configured_connection):
        """Verify cache_size pragma is applied."""
        result = configured_connection.execute("PRAGMA cache_size").fetchone()
        assert result is not None
        # Default is 64MB = 65536KB, stored as negative value
        assert result[0] < 0, "Cache size should be negative (KB format)"
        configured_connection.close()

    def test_temp_store_memory(self, configured_connection):
        """Verify temp_store is set to MEMORY."""
        result = configured_connection.execute("PRAGMA temp_store").fetchone()
        assert result is not None
        # 2 = MEMORY
        assert result[0] == 2, (
            f"Expected temp_store=2 (MEMORY), got {result[0]}"
        )
        configured_connection.close()


class TestBusyTimeout:
    """Tests for busy timeout setting."""

    def test_busy_timeout_applied(self, configured_connection):
        """Verify busy_timeout is set to prevent immediate lock failures."""
        result = configured_connection.execute("PRAGMA busy_timeout").fetchone()
        assert result is not None
        assert result[0] == 10000, (
            f"Expected busy_timeout=10000, got {result[0]}"
        )
        configured_connection.close()

    def test_busy_timeout_prevents_immediate_lock_error(
        self, sqlcipher_module, temp_db_path
    ):
        """Verify busy_timeout allows waiting for locks."""

        # Create database
        conn1 = sqlcipher_module.connect(str(temp_db_path))
        cursor1 = conn1.cursor()
        # New database: cipher_default_* before key
        apply_cipher_defaults_before_key(cursor1)
        set_sqlcipher_key(cursor1, "testpassword")
        apply_performance_pragmas(cursor1)
        cursor1.close()

        conn1.execute("CREATE TABLE test (id INTEGER PRIMARY KEY, data TEXT)")
        conn1.execute("INSERT INTO test VALUES (1, 'data1')")
        conn1.commit()

        # The busy_timeout should allow retries
        # This is a basic test that the setting is applied
        result = conn1.execute("PRAGMA busy_timeout").fetchone()
        assert result[0] == 10000

        conn1.close()


class TestSynchronousMode:
    """Tests for synchronous mode setting."""

    def test_synchronous_mode_normal(self, configured_connection):
        """Verify synchronous mode is set to NORMAL (good balance)."""
        result = configured_connection.execute("PRAGMA synchronous").fetchone()
        assert result is not None
        # Default is NORMAL (1)
        assert result[0] == 1, (
            f"Expected synchronous=1 (NORMAL), got {result[0]}"
        )
        configured_connection.close()


class TestPerformanceIntegration:
    """Integration tests for performance settings."""

    def test_all_performance_pragmas_applied(self, configured_connection):
        """Verify all performance pragmas are applied correctly."""
        pragmas = {
            "journal_mode": lambda x: x.lower() == "wal",
            "temp_store": lambda x: x == 2,
            "busy_timeout": lambda x: x == 10000,
            "synchronous": lambda x: x == 1,
        }

        for pragma, validator in pragmas.items():
            result = configured_connection.execute(
                f"PRAGMA {pragma}"
            ).fetchone()
            assert result is not None, f"PRAGMA {pragma} returned None"
            assert validator(result[0]), (
                f"PRAGMA {pragma} has unexpected value: {result[0]}"
            )

        configured_connection.close()

    def test_performance_after_reopen(self, sqlcipher_module, temp_db_path):
        """Verify performance settings persist across connections."""
        # Create database with performance pragmas
        conn1 = sqlcipher_module.connect(str(temp_db_path))
        cursor1 = conn1.cursor()
        # New database: cipher_default_* before key
        apply_cipher_defaults_before_key(cursor1)
        set_sqlcipher_key(cursor1, "testpassword")
        apply_performance_pragmas(cursor1)
        cursor1.close()

        conn1.execute("CREATE TABLE test (id INTEGER)")
        conn1.commit()
        conn1.close()

        # Reopen and verify
        conn2 = sqlcipher_module.connect(str(temp_db_path))
        cursor2 = conn2.cursor()
        # Existing database: key first, then cipher_* pragmas
        set_sqlcipher_key(cursor2, "testpassword")
        apply_sqlcipher_pragmas(cursor2, creation_mode=False)
        apply_performance_pragmas(cursor2)
        cursor2.close()

        # WAL mode persists (stored in database file)
        result = conn2.execute("PRAGMA journal_mode").fetchone()
        assert result[0].lower() == "wal"

        # Other settings are per-connection but should still be applied
        result = conn2.execute("PRAGMA busy_timeout").fetchone()
        assert result[0] == 10000

        conn2.close()

    def test_write_performance_with_settings(self, configured_connection):
        """Verify database performs well with applied settings."""
        # audit: PUNCHLIST reviewed 2026-05 — issue resolved by prior PR (recommendation: remove timing assertion or mark @pytest.mark.slow; keep the count==100 assertion).
        import time

        configured_connection.execute(
            "CREATE TABLE perf_test (id INTEGER PRIMARY KEY, data TEXT)"
        )

        # Insert 100 rows and measure time
        start = time.time()
        for i in range(100):
            configured_connection.execute(
                f"INSERT INTO perf_test VALUES ({i}, 'test_data_{i}')"
            )
        configured_connection.commit()
        elapsed = time.time() - start

        # Should complete in reasonable time (< 5 seconds with WAL)
        assert elapsed < 5, f"100 inserts took {elapsed:.2f}s - too slow"

        # Verify all data was written
        count = configured_connection.execute(
            "SELECT COUNT(*) FROM perf_test"
        ).fetchone()[0]
        assert count == 100

        configured_connection.close()