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

"""Test multi-user encrypted database functionality."""

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

import pytest

from local_deep_research.database.encrypted_db import DatabaseManager
from local_deep_research.database.models import (
    ResearchHistory,
)
from local_deep_research.database.models.auth import User


class TestMultiUserDatabase:
    """Test suite for multi-user encrypted database functionality."""

    @pytest.fixture
    def temp_dir(self):
        """Create a temporary directory for test databases."""
        temp_dir = tempfile.mkdtemp()
        yield temp_dir
        shutil.rmtree(temp_dir)

    @pytest.fixture
    def db_manager(self, temp_dir):
        """Create a database manager with a custom data directory."""
        # Mock the data directory to use our temp directory
        with patch(
            "local_deep_research.database.encrypted_db.get_data_directory"
        ) as mock_get_dir:
            mock_get_dir.return_value = Path(temp_dir)
            manager = DatabaseManager()
            manager.data_dir = Path(temp_dir) / "encrypted_databases"
            manager.data_dir.mkdir(parents=True, exist_ok=True)
            yield manager

    @pytest.fixture
    def mock_auth_db(self, monkeypatch):
        """Mock the auth database functions."""
        mock_session = MagicMock()
        mock_user = MagicMock(spec=User)
        mock_user.username = "testuser"

        mock_session.query.return_value.filter_by.return_value.first.return_value = mock_user
        mock_session.close = MagicMock()

        def mock_get_auth_db_session():
            return mock_session

        monkeypatch.setattr(
            "local_deep_research.database.auth_db.get_auth_db_session",
            mock_get_auth_db_session,
        )

        return mock_session

    def test_database_isolation_without_sqlcipher(
        self, db_manager, mock_auth_db
    ):
        """Test that the system handles missing SQLCipher gracefully."""
        # This test verifies the system's behavior when SQLCipher is not available
        # In a real deployment, SQLCipher would be required

        # Mock SQLAlchemy to simulate SQLCipher not being available
        with patch(
            "local_deep_research.database.encrypted_db.create_engine"
        ) as mock_engine:
            mock_engine.side_effect = ImportError(
                "No module named 'pysqlcipher3'"
            )

            # Attempt to create a user database
            with pytest.raises(ImportError):
                db_manager.create_user_database("testuser", "password123")

    def test_user_exists_check(self, db_manager, mock_auth_db):
        """Test checking if a user exists."""
        # Test existing user
        assert db_manager.user_exists("testuser") is True

        # Test non-existing user
        mock_auth_db.query.return_value.filter_by.return_value.first.return_value = None
        assert db_manager.user_exists("nonexistent") is False

    def test_database_path_generation(self, db_manager):
        """Test that database paths are generated correctly."""
        # Test path generation for different usernames
        path1 = db_manager._get_user_db_path("user1")
        path2 = db_manager._get_user_db_path("user2")

        # Paths should be different
        assert path1 != path2

        # Paths should be in the correct directory
        assert path1.parent == db_manager.data_dir
        assert path2.parent == db_manager.data_dir

        # Paths should use hashed usernames
        assert "user1" not in str(path1)
        assert "user2" not in str(path2)

    def test_memory_usage_tracking(self, db_manager):
        """Test memory usage statistics."""
        stats = db_manager.get_memory_usage()

        assert stats["active_connections"] == 0
        assert stats["active_sessions"] == 0
        assert stats["estimated_memory_mb"] == 0

    def test_session_management_without_sqlcipher(self, db_manager):
        """Test session management when SQLCipher is not available."""
        # Without an open database, get_session should return None
        session = db_manager.get_session("testuser")
        assert session is None

    @pytest.mark.skipif(
        True,  # Always skip for now since SQLCipher is not installed
        reason="SQLCipher not available in test environment",
    )
    def test_full_multiuser_flow(self, db_manager, mock_auth_db):
        """Test complete multi-user flow with SQLCipher (skipped if not available)."""
        # This test would run if SQLCipher were installed

        # Create databases for two users

        # Get sessions for each user
        session1 = db_manager.get_session("user1")
        session2 = db_manager.get_session("user2")

        # Add research to user1's database
        research1 = ResearchHistory(
            query="User 1 research", mode="quick", status="completed"
        )
        session1.add(research1)
        session1.commit()

        # Add research to user2's database
        research2 = ResearchHistory(
            query="User 2 research", mode="deep", status="completed"
        )
        session2.add(research2)
        session2.commit()

        # Verify isolation - each user only sees their own data
        user1_research = session1.query(ResearchHistory).all()
        user2_research = session2.query(ResearchHistory).all()

        assert len(user1_research) == 1
        assert len(user2_research) == 1
        assert user1_research[0].query == "User 1 research"
        assert user2_research[0].query == "User 2 research"

        # Test password change
        success = db_manager.change_password(
            "user1", "password1", "newpassword1"
        )
        assert success is True

        # Close databases
        db_manager.close_user_database("user1")
        db_manager.close_user_database("user2")

        # Verify databases are closed
        assert "user1" not in db_manager.connections
        assert "user2" not in db_manager.connections