learningcircuit--local-deep-research/tests/database/test_sqlcipher_connection_factory.py

"""Tests for _make_sqlcipher_connection() factory method.

Verifies:
- Factory method creates connections with correct parameters
- Correct call order: key → pragmas → verify → performance
- Parameters are passed through correctly
- Failed verification raises ValueError
- Metrics path preserves error logging
"""

from pathlib import Path
from unittest.mock import MagicMock, patch

import pytest

from local_deep_research.database.encrypted_db import DatabaseManager

# Module path prefix for patching
_EDB = "local_deep_research.database.encrypted_db"


@pytest.fixture
def manager():
    """Create a DatabaseManager with encryption disabled for testing."""
    with patch(f"{_EDB}.get_data_directory") as mock_dir:
        mock_dir.return_value = Path("/tmp/test")
        with patch.object(
            DatabaseManager, "_check_encryption_available", return_value=False
        ):
            with patch.dict("os.environ", {"LDR_ALLOW_UNENCRYPTED": "true"}):
                return DatabaseManager()


@pytest.fixture
def mock_sqlcipher():
    """Mock all SQLCipher dependencies for _make_sqlcipher_connection."""
    mock_conn = MagicMock()
    mock_cursor = MagicMock()
    mock_conn.cursor.return_value = mock_cursor

    mock_module = MagicMock()
    mock_module.connect.return_value = mock_conn

    with (
        patch(
            f"{_EDB}.get_sqlcipher_module", return_value=mock_module
        ) as mock_get,
        patch(f"{_EDB}.set_sqlcipher_key") as mock_set_key,
        patch(
            f"{_EDB}.verify_sqlcipher_connection", return_value=True
        ) as mock_verify,
        patch(f"{_EDB}.apply_sqlcipher_pragmas") as mock_pragmas,
        patch(f"{_EDB}.apply_performance_pragmas") as mock_perf,
    ):
        yield {
            "module": mock_module,
            "conn": mock_conn,
            "cursor": mock_cursor,
            "get_module": mock_get,
            "set_key": mock_set_key,
            "verify": mock_verify,
            "pragmas": mock_pragmas,
            "performance": mock_perf,
        }


class TestMakeSqlcipherConnection:
    """Verify _make_sqlcipher_connection creates connections correctly."""

    def test_returns_connection(self, manager, mock_sqlcipher):
        """Factory should return the connection object from sqlcipher3.connect."""
        result = manager._make_sqlcipher_connection(
            Path("/tmp/test.db"), "secret"
        )
        assert result is mock_sqlcipher["conn"]

    def test_connect_with_default_params(self, manager, mock_sqlcipher):
        """Default params: isolation_level=IMMEDIATE, check_same_thread=False."""
        manager._make_sqlcipher_connection(Path("/tmp/test.db"), "secret")

        mock_sqlcipher["module"].connect.assert_called_once_with(
            "/tmp/test.db",
            isolation_level="IMMEDIATE",
            check_same_thread=False,
        )

    def test_connect_with_custom_isolation_level(self, manager, mock_sqlcipher):
        """Metrics path passes isolation_level='' for deferred transactions."""
        manager._make_sqlcipher_connection(
            Path("/tmp/test.db"), "secret", isolation_level=""
        )

        mock_sqlcipher["module"].connect.assert_called_once_with(
            "/tmp/test.db",
            isolation_level="",
            check_same_thread=False,
        )

    def test_connect_with_check_same_thread(self, manager, mock_sqlcipher):
        """check_same_thread should be forwarded to sqlcipher3.connect."""
        manager._make_sqlcipher_connection(
            Path("/tmp/test.db"), "secret", check_same_thread=True
        )

        mock_sqlcipher["module"].connect.assert_called_once_with(
            "/tmp/test.db",
            isolation_level="IMMEDIATE",
            check_same_thread=True,
        )

    def test_sets_sqlcipher_key(self, manager, mock_sqlcipher):
        """Must call set_sqlcipher_key with cursor, password, and db_path."""
        manager._make_sqlcipher_connection(Path("/tmp/test.db"), "my_pass")

        mock_sqlcipher["set_key"].assert_called_once_with(
            mock_sqlcipher["cursor"], "my_pass", db_path=Path("/tmp/test.db")
        )

    def test_verifies_connection(self, manager, mock_sqlcipher):
        """Must verify connection works after setting key."""
        manager._make_sqlcipher_connection(Path("/tmp/test.db"), "secret")

        mock_sqlcipher["verify"].assert_called_once_with(
            mock_sqlcipher["cursor"]
        )

    def test_applies_sqlcipher_pragmas(self, manager, mock_sqlcipher):
        """Must apply SQLCipher pragmas with creation_mode=False."""
        manager._make_sqlcipher_connection(Path("/tmp/test.db"), "secret")

        mock_sqlcipher["pragmas"].assert_called_once_with(
            mock_sqlcipher["cursor"], creation_mode=False
        )

    def test_applies_performance_pragmas(self, manager, mock_sqlcipher):
        """Must call apply_performance_pragmas with cursor."""
        manager._make_sqlcipher_connection(Path("/tmp/test.db"), "secret")

        mock_sqlcipher["performance"].assert_called_once_with(
            mock_sqlcipher["cursor"]
        )

    def test_closes_cursor(self, manager, mock_sqlcipher):
        """Cursor must be closed before returning."""
        manager._make_sqlcipher_connection(Path("/tmp/test.db"), "secret")

        mock_sqlcipher["cursor"].close.assert_called_once()

    def test_call_order(self, manager, mock_sqlcipher):
        """Operations must happen in order: key → pragmas → verify → performance."""
        call_order = []
        mock_sqlcipher["set_key"].side_effect = lambda *a, **kw: (
            call_order.append("key")
        )
        mock_sqlcipher["verify"].side_effect = lambda *a: (
            call_order.append("verify") or True
        )
        mock_sqlcipher["pragmas"].side_effect = lambda *a, **kw: (
            call_order.append("pragmas")
        )
        mock_sqlcipher["performance"].side_effect = lambda *a: (
            call_order.append("performance")
        )

        manager._make_sqlcipher_connection(Path("/tmp/test.db"), "secret")

        assert call_order == ["key", "pragmas", "verify", "performance"]

    def test_raises_on_failed_verification(self, manager, mock_sqlcipher):
        """Must raise ValueError when verification fails."""
        mock_sqlcipher["verify"].return_value = False

        with pytest.raises(ValueError, match="Failed to verify database key"):
            manager._make_sqlcipher_connection(Path("/tmp/test.db"), "secret")

    def test_no_performance_pragmas_after_failed_verification(
        self, manager, mock_sqlcipher
    ):
        """Cipher pragmas run before verify, but performance pragmas must not."""
        mock_sqlcipher["verify"].return_value = False

        with pytest.raises(ValueError):
            manager._make_sqlcipher_connection(Path("/tmp/test.db"), "secret")

        mock_sqlcipher["pragmas"].assert_called_once_with(
            mock_sqlcipher["cursor"], creation_mode=False
        )
        mock_sqlcipher["performance"].assert_not_called()

    def test_closes_cursor_and_conn_on_verification_failure(
        self, manager, mock_sqlcipher
    ):
        """Cursor and connection must be closed when verification fails."""
        mock_sqlcipher["verify"].return_value = False

        with pytest.raises(ValueError):
            manager._make_sqlcipher_connection(Path("/tmp/test.db"), "secret")

        mock_sqlcipher["cursor"].close.assert_called_once()
        mock_sqlcipher["conn"].close.assert_called_once()

    def test_closes_cursor_and_conn_on_key_error(self, manager, mock_sqlcipher):
        """Cursor and connection must be closed when set_sqlcipher_key raises."""
        mock_sqlcipher["set_key"].side_effect = RuntimeError("key failed")

        with pytest.raises(RuntimeError):
            manager._make_sqlcipher_connection(Path("/tmp/test.db"), "secret")

        mock_sqlcipher["cursor"].close.assert_called_once()
        mock_sqlcipher["conn"].close.assert_called_once()

    def test_closes_cursor_and_conn_on_pragma_error(
        self, manager, mock_sqlcipher
    ):
        """Cursor and connection must be closed when pragma application raises."""
        mock_sqlcipher["pragmas"].side_effect = RuntimeError("pragma failed")

        with pytest.raises(RuntimeError):
            manager._make_sqlcipher_connection(Path("/tmp/test.db"), "secret")

        mock_sqlcipher["cursor"].close.assert_called_once()
        mock_sqlcipher["conn"].close.assert_called_once()
        mock_sqlcipher["verify"].assert_not_called()
        mock_sqlcipher["performance"].assert_not_called()

    def test_connect_converts_path_to_string(self, manager, mock_sqlcipher):
        """Path objects must be converted to str for sqlcipher3.connect."""
        manager._make_sqlcipher_connection(Path("/some/dir/test.db"), "secret")

        call_args = mock_sqlcipher["module"].connect.call_args
        assert call_args[0][0] == "/some/dir/test.db"
        assert isinstance(call_args[0][0], str)

    def test_connect_with_none_isolation_level(self, manager, mock_sqlcipher):
        """isolation_level=None (autocommit) must be forwarded to connect."""
        manager._make_sqlcipher_connection(
            Path("/tmp/test.db"), "secret", isolation_level=None
        )

        mock_sqlcipher["module"].connect.assert_called_once_with(
            "/tmp/test.db",
            isolation_level=None,
            check_same_thread=False,
        )

    def test_conn_not_closed_on_success(self, manager, mock_sqlcipher):
        """On success, cursor is closed but connection must remain open."""
        result = manager._make_sqlcipher_connection(
            Path("/tmp/test.db"), "secret"
        )

        mock_sqlcipher["cursor"].close.assert_called_once()
        mock_sqlcipher["conn"].close.assert_not_called()
        assert result is mock_sqlcipher["conn"]

    def test_key_error_prevents_further_calls(self, manager, mock_sqlcipher):
        """If set_sqlcipher_key fails, no pragmas or verify should run."""
        mock_sqlcipher["set_key"].side_effect = RuntimeError("bad key")

        with pytest.raises(RuntimeError):
            manager._make_sqlcipher_connection(Path("/tmp/test.db"), "secret")

        mock_sqlcipher["pragmas"].assert_not_called()
        mock_sqlcipher["verify"].assert_not_called()
        mock_sqlcipher["performance"].assert_not_called()

    def test_original_exception_propagated(self, manager, mock_sqlcipher):
        """The original exception must propagate without wrapping."""
        original = RuntimeError("specific error")
        mock_sqlcipher["pragmas"].side_effect = original

        with pytest.raises(RuntimeError) as exc_info:
            manager._make_sqlcipher_connection(Path("/tmp/test.db"), "secret")

        assert exc_info.value is original

    def test_conn_closed_even_if_cursor_close_raises(
        self, manager, mock_sqlcipher
    ):
        """Connection must close even when cursor.close() raises during cleanup."""
        mock_sqlcipher["verify"].return_value = False
        mock_sqlcipher["cursor"].close.side_effect = RuntimeError(
            "cursor close failed"
        )

        with pytest.raises(ValueError, match="Failed to verify database key"):
            manager._make_sqlcipher_connection(Path("/tmp/test.db"), "secret")

        mock_sqlcipher["conn"].close.assert_called_once()

    def test_import_error_propagates(self, manager):
        """ImportError from get_sqlcipher_module() must propagate cleanly."""
        with patch(
            f"{_EDB}.get_sqlcipher_module",
            side_effect=ImportError("no sqlcipher3"),
        ):
            with pytest.raises(ImportError, match="no sqlcipher3"):
                manager._make_sqlcipher_connection(
                    Path("/tmp/test.db"), "secret"
                )

    def test_connect_failure_propagates(self, manager, mock_sqlcipher):
        """Exceptions from sqlcipher3.connect() must propagate without cleanup attempts."""
        mock_sqlcipher["module"].connect.side_effect = OSError("disk full")

        with pytest.raises(OSError, match="disk full"):
            manager._make_sqlcipher_connection(Path("/tmp/test.db"), "secret")

        mock_sqlcipher["set_key"].assert_not_called()

    def test_closes_cursor_and_conn_on_performance_pragma_error(
        self, manager, mock_sqlcipher
    ):
        """Cursor and connection must be closed when apply_performance_pragmas raises."""
        mock_sqlcipher["performance"].side_effect = RuntimeError(
            "perf pragma failed"
        )

        with pytest.raises(RuntimeError, match="perf pragma failed"):
            manager._make_sqlcipher_connection(Path("/tmp/test.db"), "secret")

        mock_sqlcipher["cursor"].close.assert_called_once()
        mock_sqlcipher["conn"].close.assert_called_once()
        mock_sqlcipher["verify"].assert_called_once()

    def test_closes_cursor_and_conn_on_verify_exception(
        self, manager, mock_sqlcipher
    ):
        """Cleanup must happen when verify_sqlcipher_connection raises (not just returns False)."""
        mock_sqlcipher["verify"].side_effect = RuntimeError("verify crashed")

        with pytest.raises(RuntimeError, match="verify crashed"):
            manager._make_sqlcipher_connection(Path("/tmp/test.db"), "secret")

        mock_sqlcipher["cursor"].close.assert_called_once()
        mock_sqlcipher["conn"].close.assert_called_once()
        mock_sqlcipher["performance"].assert_not_called()


# NOTE: The former TestMetricsThreadConnectionLogging tests exercised
# the inline SQLCipher creator inside create_thread_safe_session_for_metrics.
# That method now delegates to open_user_database(), so connection-failure
# logging is covered by the open_user_database tests instead.