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

"""
Tests for SQLCipher thread safety.

These tests verify that SQLCipher connections work correctly when
accessed from multiple threads.
"""

import shutil
import tempfile
import threading
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
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_threads.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."""
    # Detect if we're creating a new database or opening an existing one
    creation_mode = not db_path.exists()
    conn = sqlcipher_module.connect(str(db_path), check_same_thread=False)
    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 TestThreadSafety:
    """Tests for basic thread safety."""

    def test_connection_per_thread(self, sqlcipher_module, temp_db_path):
        """Verify each thread can create its own connection."""
        password = "thread_test"

        # Create initial database
        conn = create_configured_connection(
            sqlcipher_module, temp_db_path, password
        )
        conn.execute("CREATE TABLE thread_data (thread_id INTEGER, value TEXT)")
        conn.commit()
        conn.close()

        results = []
        errors = []

        def thread_work(thread_id):
            try:
                # Each thread creates its own connection
                thread_conn = create_configured_connection(
                    sqlcipher_module, temp_db_path, password
                )
                thread_conn.execute(
                    "INSERT INTO thread_data VALUES (?, ?)",
                    (thread_id, f"data_from_{thread_id}"),
                )
                thread_conn.commit()
                thread_conn.close()
                results.append(thread_id)
            except Exception as e:
                errors.append((thread_id, str(e)))

        # Run 5 threads concurrently
        threads = []
        for i in range(5):
            t = threading.Thread(target=thread_work, args=(i,))
            threads.append(t)
            t.start()

        for t in threads:
            t.join()

        # Verify no errors
        assert not errors, f"Thread errors: {errors}"

        # Verify all data was written
        conn = create_configured_connection(
            sqlcipher_module, temp_db_path, password
        )
        count = conn.execute("SELECT COUNT(*) FROM thread_data").fetchone()[0]
        assert count == 5, f"Expected 5 rows, got {count}"
        conn.close()

    def test_concurrent_reads_different_connections(
        self, sqlcipher_module, temp_db_path
    ):
        """Verify concurrent reads work from different connections."""
        password = "concurrent_read_test"

        # Create and populate database
        conn = create_configured_connection(
            sqlcipher_module, temp_db_path, password
        )
        conn.execute("CREATE TABLE test (id INTEGER PRIMARY KEY, data TEXT)")
        for i in range(100):
            conn.execute("INSERT INTO test VALUES (?, ?)", (i, f"data_{i}"))
        conn.commit()
        conn.close()

        results = []
        errors = []

        def read_work(reader_id):
            try:
                reader_conn = create_configured_connection(
                    sqlcipher_module, temp_db_path, password
                )
                # Read random rows
                data = reader_conn.execute(
                    "SELECT COUNT(*) FROM test"
                ).fetchone()[0]
                results.append((reader_id, data))
                reader_conn.close()
            except Exception as e:
                errors.append((reader_id, str(e)))

        # Run 10 concurrent readers
        with ThreadPoolExecutor(max_workers=10) as executor:
            futures = [executor.submit(read_work, i) for i in range(10)]
            for future in as_completed(futures):
                pass  # Just wait for completion

        assert not errors, f"Read errors: {errors}"
        assert len(results) == 10
        # All readers should see 100 rows
        assert all(count == 100 for _, count in results)

    def test_write_serialization(self, sqlcipher_module, temp_db_path):
        """Verify concurrent writes are properly serialized."""
        # audit: PUNCHLIST reviewed 2026-05 — issue resolved by prior PR (recommendation: keep but increase timeout or mark @pytest.mark.slow).
        password = "write_serial_test"

        # 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()

        errors = []

        def increment_work(worker_id, increments):
            try:
                worker_conn = create_configured_connection(
                    sqlcipher_module, temp_db_path, password
                )
                for _ in range(increments):
                    # SQLite handles serialization with busy_timeout
                    worker_conn.execute(
                        "UPDATE counter SET count = count + 1 WHERE id = 1"
                    )
                    worker_conn.commit()
                worker_conn.close()
            except Exception as e:
                errors.append((worker_id, str(e)))

        # 5 workers each doing 10 increments = 50 total
        threads = []
        for i in range(5):
            t = threading.Thread(target=increment_work, args=(i, 10))
            threads.append(t)
            t.start()

        for t in threads:
            t.join()

        assert not errors, f"Write errors: {errors}"

        # Verify final count
        conn = create_configured_connection(
            sqlcipher_module, temp_db_path, password
        )
        count = conn.execute(
            "SELECT count FROM counter WHERE id = 1"
        ).fetchone()[0]
        assert count == 50, f"Expected 50, got {count}"
        conn.close()


class TestConnectionPooling:
    """Tests for connection pooling behavior."""

    def test_multiple_connections_same_database(
        self, sqlcipher_module, temp_db_path
    ):
        """Verify multiple simultaneous connections to same database."""
        password = "pool_test"

        # Create database
        conn0 = create_configured_connection(
            sqlcipher_module, temp_db_path, password
        )
        conn0.execute(
            "CREATE TABLE shared (id INTEGER PRIMARY KEY, owner TEXT)"
        )
        conn0.commit()

        # Create additional connections
        conns = [conn0]
        for i in range(4):
            c = create_configured_connection(
                sqlcipher_module, temp_db_path, password
            )
            conns.append(c)

        # Each connection writes to shared table
        for i, conn in enumerate(conns):
            conn.execute("INSERT INTO shared VALUES (?, ?)", (i, f"conn_{i}"))
            conn.commit()

        # All connections should see all data
        for conn in conns:
            count = conn.execute("SELECT COUNT(*) FROM shared").fetchone()[0]
            assert count == 5, f"Connection should see 5 rows, got {count}"

        # Close all connections
        for conn in conns:
            conn.close()

    def test_connection_reuse_safe(self, sqlcipher_module, temp_db_path):
        """Verify connections can be safely reused across operations."""
        password = "reuse_test"

        conn = create_configured_connection(
            sqlcipher_module, temp_db_path, password
        )
        conn.execute(
            "CREATE TABLE reuse_test (id INTEGER PRIMARY KEY, data TEXT)"
        )
        conn.commit()

        # Perform many operations on same connection
        for i in range(100):
            conn.execute(
                "INSERT INTO reuse_test VALUES (?, ?)", (i, f"data_{i}")
            )
            if i % 10 == 0:
                conn.commit()

        conn.commit()

        # Verify all data
        count = conn.execute("SELECT COUNT(*) FROM reuse_test").fetchone()[0]
        assert count == 100

        conn.close()


class TestBackgroundThreads:
    """Tests for background thread database access."""

    def test_background_thread_writes(self, sqlcipher_module, temp_db_path):
        """Verify background thread can write to database."""
        password = "bg_write_test"

        # Create database in main thread
        main_conn = create_configured_connection(
            sqlcipher_module, temp_db_path, password
        )
        main_conn.execute(
            "CREATE TABLE bg_data (id INTEGER PRIMARY KEY, source TEXT)"
        )
        main_conn.commit()
        main_conn.close()

        background_done = threading.Event()
        background_error = []

        def background_work():
            try:
                bg_conn = create_configured_connection(
                    sqlcipher_module, temp_db_path, password
                )
                bg_conn.execute(
                    "INSERT INTO bg_data VALUES (1, 'background_thread')"
                )
                bg_conn.commit()
                bg_conn.close()
            except Exception as e:
                background_error.append(str(e))
            finally:
                background_done.set()

        # Start background thread
        bg_thread = threading.Thread(target=background_work, daemon=True)
        bg_thread.start()

        # Wait for completion
        background_done.wait(timeout=10)

        assert not background_error, (
            f"Background thread error: {background_error}"
        )

        # Verify data in main thread
        main_conn = create_configured_connection(
            sqlcipher_module, temp_db_path, password
        )
        result = main_conn.execute(
            "SELECT source FROM bg_data WHERE id = 1"
        ).fetchone()
        assert result[0] == "background_thread"
        main_conn.close()

    def test_main_thread_reads_background_writes(
        self, sqlcipher_module, temp_db_path
    ):
        """Verify main thread can read data written by background thread."""
        # audit: PUNCHLIST reviewed 2026-05 — issue resolved by prior PR (recommendation: keep but consider removing sleep (not load-bearing for read-after-write semantics)).
        password = "main_bg_test"

        # Create database
        conn = create_configured_connection(
            sqlcipher_module, temp_db_path, password
        )
        conn.execute(
            "CREATE TABLE bg_writes (id INTEGER PRIMARY KEY, timestamp REAL)"
        )
        conn.commit()
        conn.close()

        write_count = 20
        writes_done = threading.Event()

        def writer_thread():
            writer_conn = create_configured_connection(
                sqlcipher_module, temp_db_path, password
            )
            for i in range(write_count):
                writer_conn.execute(
                    "INSERT INTO bg_writes VALUES (?, ?)",
                    (i, time.time()),
                )
                writer_conn.commit()
                time.sleep(0.01)  # Small delay between writes
            writer_conn.close()
            writes_done.set()

        # Start writer thread
        writer = threading.Thread(target=writer_thread, daemon=True)
        writer.start()

        # Wait for writes to complete
        writes_done.wait(timeout=30)

        # Read from main thread
        reader_conn = create_configured_connection(
            sqlcipher_module, temp_db_path, password
        )
        count = reader_conn.execute(
            "SELECT COUNT(*) FROM bg_writes"
        ).fetchone()[0]
        assert count == write_count, f"Expected {write_count} rows, got {count}"
        reader_conn.close()