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