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

"""Coverage tests for encrypted_db.py using regular SQLite (no SQLCipher required).

Covers:
- DatabaseManager initialisation helpers
- _is_valid_encryption_key
- is_user_connected
- get_connected_usernames
- get_memory_usage
- close_user_database / close_all_databases
- check_database_integrity success/failure paths
- get_pool_kwargs
"""

import threading
from contextlib import contextmanager
from unittest.mock import MagicMock, patch

import pytest  # noqa: F401

MODULE = "local_deep_research.database.encrypted_db"


# ---------------------------------------------------------------------------
# Helpers – create a DatabaseManager with encryption disabled
# ---------------------------------------------------------------------------


@contextmanager
def _unencrypted_manager():
    """Yield a DatabaseManager that has encryption disabled (SQLCipher not available)."""
    with (
        patch(
            f"{MODULE}.get_data_directory",
            return_value=MagicMock(
                __truediv__=lambda self, other: MagicMock(
                    mkdir=MagicMock(),
                    __truediv__=lambda self2, other2: MagicMock(),
                )
            ),
        ),
        patch(f"{MODULE}.get_env_setting", return_value=True),
        patch(
            f"{MODULE}.get_sqlcipher_module",
            side_effect=ImportError("no sqlcipher"),
        ),
    ):
        from local_deep_research.database.encrypted_db import DatabaseManager

        mgr = DatabaseManager.__new__(DatabaseManager)
        mgr.connections = {}
        mgr._connections_lock = threading.RLock()
        # __init__ is bypassed here; mirror its per-user init-lock dict so the
        # close_*/open paths that reference mgr._init_locks don't AttributeError.
        mgr._init_locks = {}
        mgr.has_encryption = False
        import os

        mgr._use_static_pool = bool(os.environ.get("TESTING"))
        from sqlalchemy.pool import StaticPool, QueuePool

        mgr._pool_class = StaticPool if mgr._use_static_pool else QueuePool
        yield mgr


# ---------------------------------------------------------------------------
# _is_valid_encryption_key
# ---------------------------------------------------------------------------


class TestIsValidEncryptionKey:
    def test_none_is_invalid(self):
        with _unencrypted_manager() as mgr:
            assert mgr._is_valid_encryption_key(None) is False

    def test_empty_string_is_invalid(self):
        with _unencrypted_manager() as mgr:
            assert mgr._is_valid_encryption_key("") is False

    def test_whitespace_only_is_invalid(self):
        with _unencrypted_manager() as mgr:
            assert mgr._is_valid_encryption_key("   ") is False

    def test_valid_password(self):
        with _unencrypted_manager() as mgr:
            assert mgr._is_valid_encryption_key("secret123") is True

    def test_single_char_password(self):
        with _unencrypted_manager() as mgr:
            assert mgr._is_valid_encryption_key("x") is True


# ---------------------------------------------------------------------------
# is_user_connected / get_connected_usernames
# ---------------------------------------------------------------------------


class TestConnectionState:
    def test_is_user_connected_false_when_absent(self):
        with _unencrypted_manager() as mgr:
            assert mgr.is_user_connected("alice") is False

    def test_is_user_connected_true_when_present(self):
        with _unencrypted_manager() as mgr:
            mgr.connections["alice"] = MagicMock()
            assert mgr.is_user_connected("alice") is True

    def test_get_connected_usernames_empty(self):
        with _unencrypted_manager() as mgr:
            assert mgr.get_connected_usernames() == set()

    def test_get_connected_usernames_snapshot(self):
        with _unencrypted_manager() as mgr:
            mgr.connections["alice"] = MagicMock()
            mgr.connections["bob"] = MagicMock()
            names = mgr.get_connected_usernames()
            assert names == {"alice", "bob"}


# ---------------------------------------------------------------------------
# get_memory_usage
# ---------------------------------------------------------------------------


class TestGetMemoryUsage:
    def test_empty_state(self):
        with _unencrypted_manager() as mgr:
            stats = mgr.get_memory_usage()
            assert stats["active_connections"] == 0
            assert "thread_engines" not in stats
            assert stats["estimated_memory_mb"] == 0.0

    def test_with_connections(self):
        with _unencrypted_manager() as mgr:
            mgr.connections["user1"] = MagicMock()
            mgr.connections["user2"] = MagicMock()
            stats = mgr.get_memory_usage()
            assert stats["active_connections"] == 2
            assert stats["estimated_memory_mb"] == pytest.approx(2 * 3.5)


# ---------------------------------------------------------------------------
# _get_pool_kwargs
# ---------------------------------------------------------------------------


class TestGetPoolKwargs:
    def test_static_pool_returns_empty(self):
        with _unencrypted_manager() as mgr:
            mgr._use_static_pool = True
            result = mgr._get_pool_kwargs()
            assert result == {}

    def test_queue_pool_returns_kwargs(self):
        with _unencrypted_manager() as mgr:
            mgr._use_static_pool = False
            result = mgr._get_pool_kwargs()
            assert "pool_size" in result
            assert "max_overflow" in result


# ---------------------------------------------------------------------------
# close_user_database
# ---------------------------------------------------------------------------


class TestCloseUserDatabase:
    def test_closes_and_removes_connection(self):
        with _unencrypted_manager() as mgr:
            mock_engine = MagicMock()
            mgr.connections["alice"] = mock_engine
            mgr.close_user_database("alice")
            mock_engine.dispose.assert_called_once()
            assert "alice" not in mgr.connections

    def test_no_error_when_user_not_connected(self):
        with _unencrypted_manager() as mgr:
            mgr.close_user_database("nonexistent")

    def test_dispose_error_handled_gracefully(self):
        with _unencrypted_manager() as mgr:
            mock_engine = MagicMock()
            mock_engine.dispose.side_effect = RuntimeError("dispose failed")
            mgr.connections["alice"] = mock_engine
            mgr.close_user_database("alice")
            assert "alice" not in mgr.connections


# ---------------------------------------------------------------------------
# close_all_databases
# ---------------------------------------------------------------------------


class TestCloseAllDatabases:
    def test_disposes_all_engines(self):
        with _unencrypted_manager() as mgr:
            e1, e2 = MagicMock(), MagicMock()
            mgr.connections["a"] = e1
            mgr.connections["b"] = e2
            mgr.close_all_databases()
            e1.dispose.assert_called_once()
            e2.dispose.assert_called_once()
            assert mgr.connections == {}