learningcircuit--local-deep-research/tests/chat/test_chat_user_isolation.py

"""Unit tests for ChatService user isolation.

These tests verify that users cannot access each other's data.
In LDR's architecture, each user has a separate encrypted database,
providing strong isolation at the storage layer.

These are isolation tests — they only care that the correct username
reaches ``get_user_db_session``. Whether the service then succeeds,
returns empty, or raises is irrelevant to the isolation property, so
each test catches the expected service-layer exception
(``ChatSessionNotFound`` for missing rows, ``SQLAlchemyError`` from
the now-strict read methods) and proceeds to assert on the captured
username.
"""

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

import pytest
from sqlalchemy.exc import SQLAlchemyError

from src.local_deep_research.chat.service import ChatSessionNotFound

# Exceptions every isolation test must tolerate: the MagicMock query
# chains don't match the real SQLAlchemy shape used by some methods, so
# the service may raise after the username has already been captured —
# isolation only cares about the capture.
_ISOLATION_EXPECTED = (ChatSessionNotFound, SQLAlchemyError, TypeError)


@pytest.fixture
def username_capturing_db():
    """Patch ``get_user_db_session`` to capture every username that
    reaches the data layer.

    Returns a tuple ``(captured_usernames, configure)`` where
    ``configure`` lets each test customise the inner ``MagicMock``
    session (e.g. wire a specific ``query`` chain) before the patch
    goes live. Replaces the 8-line ``captured_username`` +
    ``@contextmanager`` boilerplate that was duplicated across 7 test
    methods.
    """
    captured: list[str] = []
    mock_session = MagicMock()

    @contextmanager
    def _fake_get_user_db_session(username, password=None):
        captured.append(username)
        yield mock_session

    with patch(
        "src.local_deep_research.chat.service.get_user_db_session",
        _fake_get_user_db_session,
    ):
        yield captured, mock_session


class TestUserIsolation:
    """Tests verifying user data isolation in ChatService."""

    def test_get_session_uses_correct_username_for_db_access(
        self, username_capturing_db
    ):
        """get_session queries the correct user's database."""
        from src.local_deep_research.chat.service import ChatService

        captured, mock_session = username_capturing_db
        mock_session.query.return_value.filter_by.return_value.first.return_value = None

        service = ChatService(username="alice")
        with contextlib.suppress(ChatSessionNotFound):
            service.get_session("some-session-id")

        assert captured == ["alice"]

    def test_get_messages_uses_correct_username_for_db_access(
        self, username_capturing_db
    ):
        """get_session_messages queries the correct user's database."""
        from src.local_deep_research.chat.service import ChatService

        captured, _ = username_capturing_db

        service = ChatService(username="bob")
        with contextlib.suppress(*_ISOLATION_EXPECTED):
            service.get_session_messages("some-session-id")

        assert captured == ["bob"]

    def test_list_sessions_uses_correct_username_for_db_access(
        self, username_capturing_db
    ):
        """list_sessions queries the correct user's database."""
        from src.local_deep_research.chat.service import ChatService

        captured, _ = username_capturing_db

        service = ChatService(username="charlie")
        with contextlib.suppress(*_ISOLATION_EXPECTED):
            service.list_sessions()

        assert captured == ["charlie"]

    def test_update_context_uses_correct_username_for_db_access(
        self, username_capturing_db
    ):
        """update_accumulated_context queries the correct user's database."""
        from src.local_deep_research.chat.service import ChatService

        captured, mock_session = username_capturing_db
        mock_session.query.return_value.filter_by.return_value.with_for_update.return_value.first.return_value = None
        mock_session.query.return_value.filter_by.return_value.first.return_value = None

        service = ChatService(username="diana")
        service.update_accumulated_context("some-session-id")

        assert captured == ["diana"]

    def test_delete_session_uses_correct_username_for_db_access(
        self, username_capturing_db
    ):
        """delete_session queries the correct user's database."""
        from src.local_deep_research.chat.service import ChatService

        captured, mock_session = username_capturing_db
        mock_session.query.return_value.filter_by.return_value.first.return_value = None

        service = ChatService(username="eve")
        service.delete_session("some-session-id")

        assert captured == ["eve"]


class TestCrossUserAccess:
    """Tests verifying cross-user access is prevented."""

    def test_different_services_use_different_usernames(
        self, username_capturing_db
    ):
        """Different ChatService instances use their own usernames."""
        from src.local_deep_research.chat.service import ChatService

        captured, _ = username_capturing_db

        service_alice = ChatService(username="alice")
        service_bob = ChatService(username="bob")

        # Both create sessions
        service_alice.create_session()
        service_bob.create_session()

        # Verify each service used its own username — order-preserving.
        assert captured[0] == "alice"
        assert captured[1] == "bob"

    def test_each_operation_passes_username_to_database(
        self, username_capturing_db
    ):
        """All CRUD operations consistently use the service username."""
        from src.local_deep_research.chat.service import ChatService

        captured, mock_session = username_capturing_db
        mock_session.query.return_value.filter_by.return_value.first.return_value = None
        mock_session.query.return_value.filter_by.return_value.order_by.return_value.offset.return_value.limit.return_value.all.return_value = []
        mock_session.query.return_value.order_by.return_value.offset.return_value.limit.return_value.all.return_value = []

        service = ChatService(username="consistent_user")

        # Perform various operations. Each get_user_db_session entry
        # must record the username; the operations themselves may
        # legitimately raise (ChatSessionNotFound for missing rows,
        # SQLAlchemyError from the now-strict list/messages reads,
        # or TypeError from MagicMock chains that don't match the
        # real query shape). Isolate each call so a later op still
        # gets to record its username.
        service.create_session()
        for op in (
            lambda: service.get_session("session-1"),
            lambda: service.get_session_messages("session-1"),
            lambda: service.list_sessions(),
            lambda: service.delete_session("session-1"),
        ):
            with contextlib.suppress(*_ISOLATION_EXPECTED):
                op()

        # Every operation must have used the same username
        assert all(u == "consistent_user" for u in captured), captured
        assert len(captured) >= 5  # create + 4 ops