From ddc16cfc21f4fc1d17f21d62ad92be56a691b263 Mon Sep 17 00:00:00 2001 From: Vineeth Voruganti <13438633+VVoruganti@users.noreply.github.com> Date: Wed, 6 Aug 2025 16:20:22 -0400 Subject: [PATCH] Vineeth/dev 1027 (#177) * fix (deriver): Add Sentry decorators and some additional tests * feat (test): Deriver Fixtures for Testing --- src/deriver/deriver.py | 8 + tests/conftest.py | 126 +++++++++ tests/crud/test_session.py | 133 ++++++++++ tests/deriver/README.md | 72 ++++++ tests/deriver/__init__.py | 1 + tests/deriver/conftest.py | 309 +++++++++++++++++++++++ tests/deriver/test_deriver_processing.py | 113 +++++++++ tests/deriver/test_queue_operations.py | 85 +++++++ tests/deriver/test_queue_processing.py | 152 +++++++++++ tests/routes/test_messages.py | 128 ++++++++++ tests/routes/test_peers.py | 53 ++++ tests/test_llm_mock.py | 65 +++++ 12 files changed, 1245 insertions(+) create mode 100644 tests/crud/test_session.py create mode 100644 tests/deriver/README.md create mode 100644 tests/deriver/__init__.py create mode 100644 tests/deriver/conftest.py create mode 100644 tests/deriver/test_deriver_processing.py create mode 100644 tests/deriver/test_queue_operations.py create mode 100644 tests/deriver/test_queue_processing.py create mode 100644 tests/test_llm_mock.py diff --git a/src/deriver/deriver.py b/src/deriver/deriver.py index 8b7ad502..44251d98 100644 --- a/src/deriver/deriver.py +++ b/src/deriver/deriver.py @@ -3,6 +3,7 @@ import logging import time from typing import Any +import sentry_sdk from langfuse.decorators import langfuse_context from sqlalchemy.ext.asyncio import AsyncSession @@ -76,6 +77,7 @@ async def critical_analysis_call( class Deriver: """Deriver class for processing messages and extracting insights.""" + @sentry_sdk.trace async def process_message( self, payload: DeriverQueuePayload, @@ -99,6 +101,7 @@ class Deriver: else: await self.process_representation_task(db, payload) + @sentry_sdk.trace async def process_summary_task( self, db: AsyncSession, @@ -116,6 +119,7 @@ class Deriver: ) log_performance_metrics(f"deriver_message_{payload.message_id}") + @sentry_sdk.trace async def process_representation_task( self, db: AsyncSession, @@ -334,6 +338,7 @@ class CertaintyReasoner: ) @conditional_observe + @sentry_sdk.trace async def derive_new_insights( self, context: ReasoningResponseWithThinking, @@ -430,6 +435,7 @@ class CertaintyReasoner: return response @conditional_observe + @sentry_sdk.trace async def reason( self, context: ReasoningResponseWithThinking, @@ -487,6 +493,7 @@ class CertaintyReasoner: return reasoning_response @conditional_observe + @sentry_sdk.trace async def _save_new_observations( self, original_context: ReasoningResponse, @@ -559,6 +566,7 @@ class CertaintyReasoner: ) +@sentry_sdk.trace async def save_working_representation_to_peer( db: AsyncSession, workspace_name: str, diff --git a/tests/conftest.py b/tests/conftest.py index a12f55fa..78a5213d 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -377,6 +377,132 @@ def mock_mirascope_functions(): } +@pytest.fixture(autouse=True) +def mock_honcho_llm_call(): + """Generic mock for the honcho_llm_call decorator to avoid actual LLM calls during tests""" + from unittest.mock import AsyncMock, MagicMock + + from src.utils.shared_models import ( + DeductiveObservation, + ReasoningResponse, + ReasoningResponseWithThinking, + SemanticQueries, + ) + + def create_mock_response( + response_model: Any = None, + stream: bool = False, + return_call_response: bool = False, + ) -> Any: + """Create a mock response based on the expected return type""" + if stream: + # For streaming responses, return an async mock + mock_stream = AsyncMock() + mock_stream.__aiter__.return_value = iter([]) + return mock_stream + elif response_model: + # For structured responses, create appropriate mock objects + if getattr(response_model, "__name__", "") == "ReasoningResponse": + mock_response = MagicMock(spec=ReasoningResponse) + mock_response.explicit = ["Test explicit observation"] + mock_response.deductive = [ + DeductiveObservation( + conclusion="Test deductive conclusion", + premises=["Test premise 1", "Test premise 2"], + ) + ] + # Add the _response attribute that contains thinking (used in the actual code) + mock_response._response = MagicMock() + mock_response._response.thinking = "Test thinking content" + return mock_response + elif ( + getattr(response_model, "__name__", "") + == "ReasoningResponseWithThinking" + ): + mock_response = MagicMock(spec=ReasoningResponseWithThinking) + mock_response.thinking = "Test thinking content" + mock_response.explicit = ["Test explicit observation"] + mock_response.deductive = [ + DeductiveObservation( + conclusion="Test deductive conclusion", + premises=["Test premise 1", "Test premise 2"], + ) + ] + return mock_response + elif getattr(response_model, "__name__", "") == "SemanticQueries": + return SemanticQueries(queries=["test query 1", "test query 2"]) + else: + # Generic response model mock + mock_response = MagicMock(spec=response_model) + # Set some default attributes for common use cases + if hasattr(mock_response, "content"): + mock_response.content = "Test response content" + return mock_response + elif return_call_response: + # For CallResponse objects, create a mock with content and usage + mock_response = MagicMock() + mock_response.content = "Test response content" + mock_response.usage = MagicMock() + mock_response.usage.input_tokens = 100 + mock_response.usage.output_tokens = 50 + return mock_response + else: + # For string responses, return a simple string + return "Test response content" + + # Patch the honcho_llm_call decorator to prevent actual LLM calls at module level + original_decorator = None + try: + import src.utils.clients + + original_decorator = src.utils.clients.honcho_llm_call + src.utils.clients.honcho_llm_call = lambda *args, **kwargs: lambda func: func # pyright: ignore[reportUnknownLambdaType] + except ImportError: + pass + + def decorator_factory(*args: Any, **kwargs: Any) -> Callable[..., Any]: # pyright: ignore[reportUnusedParameter] + """Factory function that creates the mock decorator""" + + def mock_llm_decorator(func: Callable[..., Any]) -> Callable[..., Any]: + async def async_wrapper(*func_args: Any, **func_kwargs: Any) -> Any: # pyright: ignore[reportUnusedParameter] + # Create and return appropriate mock response + return create_mock_response( + response_model=kwargs.get("response_model"), + stream=kwargs.get("stream", False), + return_call_response=kwargs.get("return_call_response", False), + ) + + def sync_wrapper(*func_args: Any, **func_kwargs: Any) -> Any: # pyright: ignore[reportUnusedParameter] + # Create and return appropriate mock response + return create_mock_response( + response_model=kwargs.get("response_model"), + stream=kwargs.get("stream", False), + return_call_response=kwargs.get("return_call_response", False), + ) + + # Check if the original function is async + import inspect + + if inspect.iscoroutinefunction(func): + return async_wrapper + else: + return sync_wrapper + + return mock_llm_decorator + + with patch("src.utils.clients.honcho_llm_call", side_effect=decorator_factory): + yield decorator_factory + + # Restore the original decorator + if original_decorator: + try: + import src.utils.clients + + src.utils.clients.honcho_llm_call = original_decorator + except ImportError: + pass + + @pytest.fixture(autouse=True) def mock_tracked_db(db_session: AsyncSession): """Mock tracked_db to use the test database session""" diff --git a/tests/crud/test_session.py b/tests/crud/test_session.py new file mode 100644 index 00000000..e4833dfd --- /dev/null +++ b/tests/crud/test_session.py @@ -0,0 +1,133 @@ +import pytest +from nanoid import generate as generate_nanoid +from sqlalchemy.ext.asyncio import AsyncSession + +from src import crud, models, schemas +from src.exceptions import ResourceNotFoundException + + +class TestSessionCRUD: + """Test suite for session CRUD operations""" + + @pytest.mark.asyncio + async def test_get_session_peer_configuration( + self, + db_session: AsyncSession, + sample_data: tuple[models.Workspace, models.Peer], + ): + """Test retrieving peer configuration data from session""" + test_workspace, test_peer = sample_data + + # Create another peer + peer2 = models.Peer( + name=str(generate_nanoid()), workspace_name=test_workspace.name + ) + db_session.add(peer2) + await db_session.flush() + + # Create session + test_session = models.Session( + name=str(generate_nanoid()), workspace_name=test_workspace.name + ) + db_session.add(test_session) + await db_session.flush() + + # Add peers to session with different configurations + peer_configs = { + test_peer.name: schemas.SessionPeerConfig( + observe_others=True, observe_me=False + ), + peer2.name: schemas.SessionPeerConfig( + observe_others=False, observe_me=True + ), + } + + # Set up peers in session + await crud.set_peers_for_session( + db_session, + workspace_name=test_workspace.name, + session_name=test_session.name, + peer_names=peer_configs, + ) + + # Test the get_session_peer_configuration function + stmt = await crud.get_session_peer_configuration( + workspace_name=test_workspace.name, + session_name=test_session.name, + ) + result = await db_session.execute(stmt) + configurations = result.all() + + # Should return configurations for all active peers + assert len(configurations) == 2 + + # Verify the structure of returned data + for peer_name, peer_config, session_peer_config in configurations: + assert isinstance(peer_name, str) + assert isinstance(peer_config, dict) or peer_config is None + assert isinstance(session_peer_config, dict) + + # Check that session_peer_config matches what we set + expected_config = peer_configs[peer_name] + assert ( + session_peer_config["observe_others"] == expected_config.observe_others + ) + assert session_peer_config["observe_me"] == expected_config.observe_me + + @pytest.mark.asyncio + async def test_get_session_not_found(self, db_session: AsyncSession): + """Test get_session with non-existent session raises ResourceNotFoundException""" + with pytest.raises(ResourceNotFoundException): + await crud.get_session(db_session, "nonexistent", "nonexistent_workspace") + + @pytest.mark.asyncio + async def test_get_peer_config_not_found( + self, + db_session: AsyncSession, + sample_data: tuple[models.Workspace, models.Peer], + ): + """Test get_peer_config with non-existent peer raises ResourceNotFoundException""" + test_workspace, _test_peer = sample_data + + # Create session + test_session = models.Session( + name=str(generate_nanoid()), workspace_name=test_workspace.name + ) + db_session.add(test_session) + await db_session.flush() + + with pytest.raises(ResourceNotFoundException): + await crud.get_peer_config( + db_session, test_workspace.name, test_session.name, "nonexistent_peer" + ) + + @pytest.mark.asyncio + async def test_clone_session_not_found(self, db_session: AsyncSession): + """Test clone_session with non-existent session raises ResourceNotFoundException""" + with pytest.raises(ResourceNotFoundException): + await crud.clone_session(db_session, "workspace", "nonexistent_session") + + @pytest.mark.asyncio + async def test_clone_session_invalid_cutoff_message( + self, + db_session: AsyncSession, + sample_data: tuple[models.Workspace, models.Peer], + ): + """Test clone_session with invalid cutoff message raises ValueError""" + test_workspace, _test_peer = sample_data + + # Create session + test_session = models.Session( + name=str(generate_nanoid()), workspace_name=test_workspace.name + ) + db_session.add(test_session) + await db_session.flush() + + # Try to clone with invalid cutoff message ID + with pytest.raises( + ValueError, + match="Message not found or doesn't belong to the specified session", + ): + await crud.clone_session( + db_session, test_workspace.name, test_session.name, "invalid_message_id" + ) diff --git a/tests/deriver/README.md b/tests/deriver/README.md new file mode 100644 index 00000000..a814e335 --- /dev/null +++ b/tests/deriver/README.md @@ -0,0 +1,72 @@ +# Deriver Testing + +This directory contains tests for the deriver system, which handles background processing of messages to extract insights and update working representations. + +## Structure + +- `conftest.py` - Shared fixtures for deriver testing +- `test_queue_operations.py` - Tests for basic queue operations +- `test_deriver_processing.py` - Tests for deriver processing logic +- `test_queue_processing.py` - Tests for queue manager and work unit processing + +## Key Fixtures + +### Database Fixtures + +- `sample_session_with_peers` - Creates a session with multiple peers having different observation configurations +- `sample_messages` - Creates sample messages for testing +- `sample_queue_items` - Creates queue items with various payload types (representation, summary) + +### Queue Fixtures + +- `create_queue_payload` - Helper to create queue payloads for testing +- `add_queue_items` - Helper to add queue items to the database +- `create_active_queue_session` - Helper to create active queue sessions for work unit tracking + +### Mocking Fixtures + +- `mock_deriver_process` - Mocks the deriver process_message method +- `mock_critical_analysis_call` - Mocks the critical analysis LLM call +- `mock_queue_manager` - Mocks the queue manager for testing +- `mock_embedding_store` - Mocks the embedding store operations + +## Testing Patterns + +### Creating Queue Items + +```python +# Create representation payloads +payload = create_queue_payload( + message=message, + task_type="representation", + sender_name=message.peer_name, + target_name=observer_peer.name, +) + +# Add to queue +queue_items = await add_queue_items([payload], session.id) +``` + +### Testing Work Units + +```python +# Create a work unit +work_unit = WorkUnit( + session_id=session.id, + sender_name=sender.name, + target_name=target.name, + task_type="representation", +) + +# Test string representation +assert str(work_unit) == f"({session.id}, {sender.name}, {target.name}, representation)" +``` + +### Mocking Deriver Processing + +```python +# Use the mock_deriver_process fixture to avoid actual LLM calls +async def test_with_mocked_deriver(mock_deriver_process): + # Deriver processing will use the mock + await process_item(queue_item.payload) +``` diff --git a/tests/deriver/__init__.py b/tests/deriver/__init__.py new file mode 100644 index 00000000..6ced4f17 --- /dev/null +++ b/tests/deriver/__init__.py @@ -0,0 +1 @@ +# Tests for the deriver system. diff --git a/tests/deriver/conftest.py b/tests/deriver/conftest.py new file mode 100644 index 00000000..97264d1c --- /dev/null +++ b/tests/deriver/conftest.py @@ -0,0 +1,309 @@ +import asyncio +from collections.abc import Callable, Generator +from datetime import datetime, timezone +from typing import Any, Literal +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest +from nanoid import generate as generate_nanoid +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession + +from src import crud, models, schemas +from src.deriver.queue_payload import create_payload + + +@pytest.fixture +def mock_deriver_process(monkeypatch: pytest.MonkeyPatch) -> Callable[..., Any]: + """Mock the deriver process_message method to avoid actual LLM calls""" + from src.deriver.deriver import Deriver + + async def mock_process_message(_self: Any, _payload: dict[str, Any]) -> None: + # Simulate processing without making actual LLM calls + pass + + monkeypatch.setattr(Deriver, "process_message", mock_process_message) + + # Return the mock for further configuration if needed + return mock_process_message + + +@pytest.fixture +def mock_critical_analysis_call() -> Generator[Callable[..., Any], None, None]: + """Mock the critical analysis call to avoid actual LLM calls""" + + async def mock_critical_analysis_call(*_args: Any, **_kwargs: Any) -> MagicMock: + # Create a mock response that matches the expected structure + mock_response = MagicMock() + mock_response.explicit = ["Test explicit observation"] + mock_response.deductive = [] + mock_response.thinking = "Test thinking content" + mock_response._response = MagicMock() + mock_response._response.thinking = "Test thinking content" + return mock_response + + # Patch the actual function in the deriver module + with patch( + "src.deriver.deriver.critical_analysis_call", mock_critical_analysis_call + ): + yield mock_critical_analysis_call + + +@pytest.fixture +async def sample_session_with_peers( + db_session: AsyncSession, + sample_data: tuple[models.Workspace, models.Peer], +) -> tuple[models.Session, list[models.Peer]]: + """Create a sample session with multiple peers for testing deriver functionality""" + workspace, peer1 = sample_data + + # Create additional peers + peer2 = models.Peer(name=str(generate_nanoid()), workspace_name=workspace.name) + peer3 = models.Peer(name=str(generate_nanoid()), workspace_name=workspace.name) + db_session.add_all([peer2, peer3]) + await db_session.flush() + + # Create session with peer configurations + session = await crud.get_or_create_session( + db_session, + schemas.SessionCreate( + name=str(generate_nanoid()), + peers={ + peer1.name: schemas.SessionPeerConfig(observe_me=True), + peer2.name: schemas.SessionPeerConfig(observe_others=True), + peer3.name: schemas.SessionPeerConfig(), # No special observation settings + }, + ), + workspace.name, + ) + await db_session.commit() + + return session, [peer1, peer2, peer3] + + +@pytest.fixture +async def sample_messages( + db_session: AsyncSession, + sample_session_with_peers: tuple[models.Session, list[models.Peer]], +) -> list[models.Message]: + """Create sample messages for testing deriver functionality""" + session, peers = sample_session_with_peers + peer1, peer2, peer3 = peers + + # Create multiple messages from different peers + messages_data = [ + { + "session_name": session.name, + "content": "Hello, this is the first message from peer1", + "peer_name": peer1.name, + "workspace_name": session.workspace_name, + }, + { + "session_name": session.name, + "content": "Hi there! This is a response from peer2", + "peer_name": peer2.name, + "workspace_name": session.workspace_name, + }, + { + "session_name": session.name, + "content": "I'm just observing this conversation as peer3", + "peer_name": peer3.name, + "workspace_name": session.workspace_name, + }, + ] + + messages: list[models.Message] = [] + for msg_data in messages_data: + message = models.Message(**msg_data) + db_session.add(message) + messages.append(message) + + await db_session.commit() + + # Query the messages again to get the committed versions + result = await db_session.execute( + select(models.Message) + .where(models.Message.session_name == session.name) + .order_by(models.Message.id) + ) + messages = list(result.scalars().all()) + + return messages + + +@pytest.fixture +def create_queue_payload() -> Callable[..., Any]: + """Helper function to create queue payloads for testing""" + + def _create_payload( + message: models.Message, + task_type: Literal["representation", "summary"], + sender_name: str | None = None, + target_name: str | None = None, + message_seq_in_session: int | None = None, + ) -> dict[str, Any]: + """Create a queue payload for testing""" + message_dict = { + "workspace_name": message.workspace_name, + "session_name": message.session_name, + "message_id": message.id, + "content": message.content, + "created_at": message.created_at or datetime.now(timezone.utc), + } + + return create_payload( + message=message_dict, + task_type=task_type, + sender_name=sender_name, + target_name=target_name, + message_seq_in_session=message_seq_in_session, + ) + + return _create_payload + + +@pytest.fixture +async def add_queue_items( + db_session: AsyncSession, +) -> Callable[[list[dict[str, Any]], str], Any]: + """Helper function to add queue items to the database""" + + async def _add_items( + payloads: list[dict[str, Any]], session_id: str + ) -> list[models.QueueItem]: + """Add queue items to the database and return them""" + queue_items: list[models.QueueItem] = [] + for payload in payloads: + queue_item = models.QueueItem( + session_id=session_id, + payload=payload, + processed=False, + ) + db_session.add(queue_item) + queue_items.append(queue_item) + + await db_session.commit() + + # Refresh to get the actual IDs + for item in queue_items: + await db_session.refresh(item) + + return queue_items + + return _add_items + + +@pytest.fixture +async def sample_queue_items( + db_session: AsyncSession, # noqa: ARG001 # pyright: ignore[reportUnusedParameter] + sample_session_with_peers: tuple[models.Session, list[models.Peer]], + sample_messages: list[models.Message], + create_queue_payload: Callable[..., Any], + add_queue_items: Callable[..., Any], +) -> list[models.QueueItem]: + """Create sample queue items for testing""" + session, peers = sample_session_with_peers + _peer1, peer2, _peer3 = peers + messages = sample_messages + + # Create various types of queue payloads + payloads: list[dict[str, Any]] = [] + + # Create representation payloads for each message + for message in messages: + # Self-representation (peer observing themselves) + payload1 = create_queue_payload( + message=message, + task_type="representation", + sender_name=message.peer_name, + target_name=message.peer_name, + ) + payloads.append(payload1) + + # Representation for observer peer + payload2 = create_queue_payload( + message=message, + task_type="representation", + sender_name=message.peer_name, + target_name=peer2.name, # peer2 observes others + ) + payloads.append(payload2) + + # Create summary payloads for session + for i, message in enumerate(messages): + payload = create_queue_payload( + message=message, + task_type="summary", + message_seq_in_session=i + 1, + ) + payloads.append(payload) + + # Add all payloads as queue items + queue_items = await add_queue_items(payloads, session.id) + + return queue_items + + +@pytest.fixture +async def create_active_queue_session(db_session: AsyncSession) -> Callable[..., Any]: + """Helper function to create active queue sessions for testing work unit tracking""" + + async def _create_active_session( + session_id: str, + sender_name: str | None = None, + target_name: str | None = None, + task_type: str = "representation", + ) -> models.ActiveQueueSession: + """Create an active queue session""" + active_session = models.ActiveQueueSession( + session_id=session_id, + sender_name=sender_name, + target_name=target_name, + task_type=task_type, + ) + db_session.add(active_session) + await db_session.commit() + await db_session.refresh(active_session) + return active_session + + return _create_active_session + + +@pytest.fixture +def mock_queue_manager(monkeypatch: pytest.MonkeyPatch) -> AsyncMock: # pyright: ignore[reportUnusedParameter] + """Mock the queue manager to avoid actual queue processing""" + from src.deriver.queue_manager import QueueManager + + # Create a mock queue manager + mock_manager = AsyncMock(spec=QueueManager) + + # Mock the methods we might need to test + mock_manager.initialize = AsyncMock() + mock_manager.shutdown = AsyncMock() + mock_manager.process_work_unit = AsyncMock() + mock_manager.get_available_work_units = AsyncMock(return_value=[]) + mock_manager.add_task = MagicMock() + mock_manager.track_work_unit = MagicMock() + mock_manager.untrack_work_unit = MagicMock() + + # Mock the attributes + mock_manager.shutdown_event = asyncio.Event() + mock_manager.active_tasks = set() + mock_manager.owned_work_units = set() + mock_manager.queue_empty_flag = asyncio.Event() + mock_manager.workers = 1 + mock_manager.semaphore = asyncio.Semaphore(1) + + return mock_manager + + +@pytest.fixture +def mock_embedding_store(monkeypatch: pytest.MonkeyPatch) -> AsyncMock: # pyright: ignore[reportUnusedParameter] + """Mock the embedding store to avoid actual embedding operations""" + from src.utils.embedding_store import EmbeddingStore + + mock_store = AsyncMock(spec=EmbeddingStore) + mock_store.save_unified_observations = AsyncMock() + mock_store.get_relevant_observations = AsyncMock(return_value=MagicMock()) + + return mock_store diff --git a/tests/deriver/test_deriver_processing.py b/tests/deriver/test_deriver_processing.py new file mode 100644 index 00000000..af5f81bc --- /dev/null +++ b/tests/deriver/test_deriver_processing.py @@ -0,0 +1,113 @@ +import signal +from collections.abc import Callable, Generator +from typing import Any + +import pytest + +from src import models +from src.deriver.queue_manager import WorkUnit + + +@pytest.mark.asyncio +class TestDeriverProcessing: + """Test suite for deriver processing using the conftest fixtures""" + + async def test_mock_deriver_process( + self, + mock_deriver_process: Callable[..., Any], # noqa: ARG001 + sample_queue_items: list[models.QueueItem], # noqa: ARG001 + ): + """Test that the deriver process is properly mocked""" + # The mock should be in place, so processing should not make real LLM calls + assert mock_deriver_process is not None + + # Verify that we have queue items to process + assert len(sample_queue_items) > 0 + + # Verify the mock is working by checking the first queue item + first_item = sample_queue_items[0] + assert first_item.payload is not None + + async def test_mock_critical_analysis_call( + self, + mock_critical_analysis_call: Generator[Callable[..., Any], None, None], + sample_messages: list[models.Message], + ): + """Test that the critical analysis call is properly mocked""" + assert mock_critical_analysis_call is not None + assert len(sample_messages) > 0 # Verify we have messages for testing + + # The mock should be in place and return a predefined response + # This ensures no actual LLM calls are made during testing + + async def test_work_unit_creation( + self, + sample_session_with_peers: tuple[models.Session, list[models.Peer]], + ): + """Test that WorkUnit objects can be created correctly""" + session, peers = sample_session_with_peers + peer1, peer2, _ = peers + + # Create a WorkUnit for representation task + work_unit = WorkUnit( + session_id=session.id, + sender_name=peer1.name, + target_name=peer2.name, + task_type="representation", + ) + + assert work_unit.session_id == session.id + assert work_unit.sender_name == peer1.name + assert work_unit.target_name == peer2.name + assert work_unit.task_type == "representation" + + # Create a WorkUnit for summary task (sender_name and target_name should be None) + summary_work_unit = WorkUnit( + session_id=session.id, + sender_name=None, + target_name=None, + task_type="summary", + ) + + assert summary_work_unit.session_id == session.id + assert summary_work_unit.sender_name is None + assert summary_work_unit.target_name is None + assert summary_work_unit.task_type == "summary" + + async def test_mock_queue_manager( + self, + mock_queue_manager: Any, # AsyncMock object + sample_session_with_peers: tuple[models.Session, list[models.Peer]], + ): + """Test that the queue manager is properly mocked""" + session, peers = sample_session_with_peers + assert session is not None + assert len(peers) == 3 + + # Verify the mock has the expected attributes + assert mock_queue_manager is not None + assert hasattr(mock_queue_manager, "initialize") + assert hasattr(mock_queue_manager, "shutdown") + assert hasattr(mock_queue_manager, "process_work_unit") + + # Verify we can call the mocked methods + await mock_queue_manager.initialize() + await mock_queue_manager.shutdown(signal.SIGTERM) + + # Verify the mocked methods were called + mock_queue_manager.initialize.assert_called_once() # type: ignore[attr-defined] + mock_queue_manager.shutdown.assert_called_once() # type: ignore[attr-defined] + + async def test_mock_embedding_store( + self, + mock_embedding_store: Any, # AsyncMock object + ): + """Test that the embedding store is properly mocked""" + assert mock_embedding_store is not None + + # Verify we can call the mocked methods + await mock_embedding_store.save_unified_observations([]) + mock_embedding_store.get_relevant_observations.return_value = [] # type: ignore[attr-defined] + + # Verify the methods were called + assert mock_embedding_store.save_unified_observations.called # type: ignore[attr-defined] diff --git a/tests/deriver/test_queue_operations.py b/tests/deriver/test_queue_operations.py new file mode 100644 index 00000000..e48f78aa --- /dev/null +++ b/tests/deriver/test_queue_operations.py @@ -0,0 +1,85 @@ +from collections.abc import Callable +from typing import Any + +import pytest + +from src import models + + +@pytest.mark.asyncio +class TestQueueOperations: + """Test suite for queue operations using the deriver conftest fixtures""" + + async def test_sample_queue_items_created( + self, + sample_queue_items: list[models.QueueItem], + ): + """Test that sample queue items are created correctly""" + # Should have 9 items: 3 messages * 2 representations + 3 summaries + assert len(sample_queue_items) == 9 + + # Check that we have the right mix of task types + representation_items = [ + item + for item in sample_queue_items + if item.payload.get("task_type") == "representation" + ] + summary_items = [ + item + for item in sample_queue_items + if item.payload.get("task_type") == "summary" + ] + + assert len(representation_items) == 6 + assert len(summary_items) == 3 + + # Check that all items are unprocessed + assert all(not item.processed for item in sample_queue_items) + + async def test_sample_session_with_peers( + self, + sample_session_with_peers: tuple[models.Session, list[models.Peer]], + ): + """Test that sample session with peers is created correctly""" + session, peers = sample_session_with_peers + assert session is not None + assert len(peers) == 3 + + # Check that all peers have the same workspace + assert all(peer.workspace_name == session.workspace_name for peer in peers) + + async def test_sample_messages( + self, + sample_messages: list[models.Message], + ): + """Test that sample messages are created correctly""" + assert len(sample_messages) == 3 + + # Check that all messages have content + assert all(message.content for message in sample_messages) + + # Check that all messages have peer names + assert all(message.peer_name for message in sample_messages) + + async def test_create_active_queue_session( + self, + create_active_queue_session: Callable[..., Any], + sample_session_with_peers: tuple[models.Session, list[models.Peer]], + ): + """Test that we can create active queue sessions""" + session, peers = sample_session_with_peers + peer1, peer2, _ = peers + + # Create an active queue session + active_session = await create_active_queue_session( + session_id=session.id, + sender_name=peer1.name, + target_name=peer2.name, + task_type="representation", + ) + + assert active_session is not None + assert active_session.session_id == session.id + assert active_session.sender_name == peer1.name + assert active_session.target_name == peer2.name + assert active_session.task_type == "representation" diff --git a/tests/deriver/test_queue_processing.py b/tests/deriver/test_queue_processing.py new file mode 100644 index 00000000..85944830 --- /dev/null +++ b/tests/deriver/test_queue_processing.py @@ -0,0 +1,152 @@ +from typing import Any + +import pytest +from sqlalchemy.ext.asyncio import AsyncSession + +from src import models +from src.deriver.consumer import process_item +from src.deriver.queue_manager import QueueManager, WorkUnit + + +@pytest.mark.asyncio +class TestQueueProcessing: + """Test suite for queue processing functionality""" + + async def test_get_available_work_units( + self, + db_session: AsyncSession, + sample_queue_items: list[models.QueueItem], + sample_session_with_peers: tuple[models.Session, list[models.Peer]], + ): + """Test that get_available_work_units correctly identifies unprocessed work""" + session, _peers = sample_session_with_peers # pyright: ignore[reportUnusedVariable] + + # Verify we have queue items from our test setup + assert len(sample_queue_items) == 9 # 6 representation + 3 summary + + # Create a queue manager instance + queue_manager = QueueManager() + + # Get available work units + work_units = await queue_manager.get_available_work_units(db_session) + + # Should have some work units available (may include items from other tests) + assert len(work_units) > 0 + + # Check that all work units have the expected structure + for work_unit in work_units: + assert isinstance(work_unit, WorkUnit) + assert work_unit.task_type in ["representation", "summary"] + + # The test is mainly verifying that get_available_work_units works without errors + # and returns properly structured WorkUnit objects + + async def test_work_unit_claiming( + self, + db_session: AsyncSession, + sample_queue_items: list[models.QueueItem], # noqa: ARG001 # pyright: ignore[reportUnusedParameter] + sample_session_with_peers: tuple[models.Session, list[models.Peer]], + ): + """Test that work units can be claimed and are not available to other workers""" + _session, _peers = sample_session_with_peers + + # Create a queue manager instance + queue_manager = QueueManager() + + # Get available work units + work_units = await queue_manager.get_available_work_units(db_session) + assert len(work_units) > 0 + + # Claim a work unit by creating an ActiveQueueSession entry + work_unit = work_units[0] + active_session = models.ActiveQueueSession( + session_id=work_unit.session_id, + sender_name=work_unit.sender_name, + target_name=work_unit.target_name, + task_type=work_unit.task_type, + ) + db_session.add(active_session) + await db_session.commit() + + # Get available work units again - the claimed one should not be available + remaining_work_units = await queue_manager.get_available_work_units(db_session) + + # The claimed work unit should not be in the remaining list + assert work_unit not in remaining_work_units + # We can't assert exact count difference because work units are grouped by unique combinations + + async def test_stale_work_unit_cleanup( + self, + db_session: AsyncSession, + sample_session_with_peers: tuple[models.Session, list[models.Peer]], + ): + """Test that stale work units are cleaned up properly""" + _session, _peers = sample_session_with_peers + + # Create an active queue session with an old timestamp + # from datetime import datetime, timedelta, timezone + + # datetime.now(timezone.utc) - timedelta(minutes=10) + + # We'll test this by checking that the cleanup logic works in get_available_work_units + # which is called by the queue manager during normal operation + queue_manager = QueueManager() + + # Get available work units - this should clean up stale entries + work_units = await queue_manager.get_available_work_units(db_session) + + # This test ensures the cleanup logic doesn't break, though we don't have stale entries yet + assert isinstance(work_units, list) + + async def test_process_item_with_mocked_deriver( + self, + mock_deriver_process: Any, # noqa: ARG001 # pyright: ignore[reportUnusedParameter] + sample_queue_items: list[models.QueueItem], + ): + """Test that process_item works with mocked deriver""" + # Take a sample queue item and process it + queue_item = sample_queue_items[0] + + # This should not raise an exception since the deriver is mocked + await process_item(queue_item.payload) + + # The mock should have been called + # Note: We can't easily verify this since we're mocking the class method directly + # In a real test, we might want to mock at a different level + + async def test_work_unit_string_representation( + self, sample_session_with_peers: tuple[models.Session, list[models.Peer]] + ): + """Test that WorkUnit string representation works correctly""" + session, peers = sample_session_with_peers + peer1, peer2, _ = peers + + # Create a representation work unit + work_unit = WorkUnit( + session_id=session.id, + sender_name=peer1.name, + target_name=peer2.name, + task_type="representation", + ) + + # Convert to string + work_unit_str = str(work_unit) + + # Check that the string contains the expected information + assert session.id in work_unit_str + assert peer1.name in work_unit_str + assert peer2.name in work_unit_str + assert "representation" in work_unit_str + + # Create a summary work unit + summary_work_unit = WorkUnit( + session_id=session.id, + sender_name=None, + target_name=None, + task_type="summary", + ) + + summary_str = str(summary_work_unit) + assert session.id in summary_str + assert "None" in summary_str + assert "summary" in summary_str diff --git a/tests/routes/test_messages.py b/tests/routes/test_messages.py index c67fd2f0..06e9c39f 100644 --- a/tests/routes/test_messages.py +++ b/tests/routes/test_messages.py @@ -1,3 +1,5 @@ +from unittest.mock import patch + import pytest from fastapi.testclient import TestClient from nanoid import generate as generate_nanoid @@ -793,3 +795,129 @@ async def test_create_batch_messages_max_limit( assert len(data) == 100 assert data[0]["content"] == "Message 0" assert data[99]["content"] == "Message 99" + + +@pytest.mark.asyncio +async def test_get_messages_handles_crud_value_error( + client: TestClient, db_session: AsyncSession, sample_data: tuple[Workspace, Peer] +): + """Test that ValueError from CRUD is properly handled in get_messages""" + test_workspace, _test_peer = sample_data + + # Create a test session + test_session = models.Session( + workspace_name=test_workspace.name, name=str(generate_nanoid()) + ) + db_session.add(test_session) + await db_session.commit() + + # Mock the CRUD function to raise ValueError + with patch("src.routers.messages.crud.get_messages") as mock_get: + mock_get.side_effect = ValueError("Test CRUD error") + + response = client.post( + f"/v2/workspaces/{test_workspace.name}/sessions/{test_session.name}/messages/list", + json={}, + ) + + # Should raise ResourceNotFoundException which gets converted to 404 + assert response.status_code == 404 + + +@pytest.mark.asyncio +async def test_get_message_handles_not_found( + client: TestClient, db_session: AsyncSession, sample_data: tuple[Workspace, Peer] +): + """Test that ResourceNotFoundException is properly handled in get_message""" + test_workspace, _ = sample_data + + # Create a test session + test_session = models.Session( + workspace_name=test_workspace.name, name=str(generate_nanoid()) + ) + db_session.add(test_session) + await db_session.commit() + + # Try to get a non-existent message + with patch("src.routers.messages.crud.get_message") as mock_get: + mock_get.return_value = None + + response = client.get( + f"/v2/workspaces/{test_workspace.name}/sessions/{test_session.name}/messages/nonexistent" + ) + + # Should raise ResourceNotFoundException which gets converted to 404 + assert response.status_code == 404 + + +@pytest.mark.asyncio +async def test_update_message_handles_crud_value_error( + client: TestClient, db_session: AsyncSession, sample_data: tuple[Workspace, Peer] +): + """Test that ValueError from CRUD is properly handled in update_message""" + test_workspace, test_peer = sample_data + + # Create a test session and message + test_session = models.Session( + workspace_name=test_workspace.name, name=str(generate_nanoid()) + ) + db_session.add(test_session) + await db_session.commit() + + test_message = models.Message( + session_name=test_session.name, + content="Test message", + workspace_name=test_workspace.name, + peer_name=test_peer.name, + ) + db_session.add(test_message) + await db_session.commit() + + # Mock the CRUD function to raise ValueError + with patch("src.routers.messages.crud.update_message") as mock_update: + mock_update.side_effect = ValueError("Test CRUD error") + + response = client.put( + f"/v2/workspaces/{test_workspace.name}/sessions/{test_session.name}/messages/{test_message.public_id}", + json={"metadata": {"key": "value"}}, + ) + + # Should raise ResourceNotFoundException which gets converted to 404 + assert response.status_code == 404 + + +@pytest.mark.asyncio +async def test_create_messages_with_file_too_large( + client: TestClient, db_session: AsyncSession, sample_data: tuple[Workspace, Peer] +): + """Test that FileTooLargeError is properly handled in create_messages_with_file""" + test_workspace, test_peer = sample_data + + # Create a test session + test_session = models.Session( + workspace_name=test_workspace.name, name=str(generate_nanoid()) + ) + db_session.add(test_session) + await db_session.commit() + + # Create a large file that exceeds the size limit + import io + + large_content = b"x" * (10 * 1024 * 1024) # 10MB file + file_data = io.BytesIO(large_content) + + # Mock the settings to make the test deterministic + with patch( + "src.routers.messages.settings.MAX_FILE_SIZE", 5 * 1024 * 1024 + ): # 5MB limit + files = {"file": ("large_file.txt", file_data, "text/plain")} + form_data = {"peer_id": test_peer.name} + + response = client.post( + f"/v2/workspaces/{test_workspace.name}/sessions/{test_session.name}/messages/upload", + files=files, + data=form_data, + ) + + # Should raise FileTooLargeError which gets converted to 413 + assert response.status_code == 413 diff --git a/tests/routes/test_peers.py b/tests/routes/test_peers.py index 44aaaf8b..00eab6e5 100644 --- a/tests/routes/test_peers.py +++ b/tests/routes/test_peers.py @@ -506,3 +506,56 @@ def test_search_peer_with_semantic_search_true_disabled( assert "Semantic search requires EMBED_MESSAGES flag to be enabled" in data.get( "detail", "" ) + + +def test_get_peers_with_complex_filter( + client: TestClient, sample_data: tuple[Workspace, Peer] +): + """Test peer listing with complex filters""" + test_workspace, _ = sample_data + + # Create peers with different metadata + for i in range(3): + client.post( + f"/v2/workspaces/{test_workspace.name}/peers", + json={ + "name": str(generate_nanoid()), + "metadata": {"index": i, "type": "test"}, + }, + ) + + # Test complex filter combination + response = client.post( + f"/v2/workspaces/{test_workspace.name}/peers/list", + json={ + "filter": { + "AND": [ + {"metadata": {"type": "test"}}, + {"metadata": {"index": {"gte": 1}}}, + ] + } + }, + ) + assert response.status_code == 200 + data = response.json() + assert "items" in data + + +def test_update_peer_all_fields( + client: TestClient, sample_data: tuple[Workspace, Peer] +): + """Test peer update with all possible fields""" + test_workspace, test_peer = sample_data + + # Test updating both metadata and configuration + metadata = {"updated": True, "version": 2} + configuration = {"features": {"new_feature": True}} + + response = client.put( + f"/v2/workspaces/{test_workspace.name}/peers/{test_peer.name}", + json={"metadata": metadata, "configuration": configuration}, + ) + assert response.status_code == 200 + data = response.json() + assert data["metadata"] == metadata + assert data["configuration"] == configuration diff --git a/tests/test_llm_mock.py b/tests/test_llm_mock.py new file mode 100644 index 00000000..7bdab32a --- /dev/null +++ b/tests/test_llm_mock.py @@ -0,0 +1,65 @@ +from datetime import datetime, timezone +from typing import cast +from unittest.mock import MagicMock + +import pytest + +from src.models import Message + + +@pytest.mark.asyncio +async def test_generic_honcho_llm_call_mock(): + """Test that the generic honcho_llm_call mock is working for existing decorated functions""" + # Import a function that we know is decorated with honcho_llm_call + from src.deriver.deriver import critical_analysis_call + + # Call the decorated function - this should use our mock + result = await critical_analysis_call( + peer_name="test_peer", + message_created_at=datetime(2023, 1, 1, 0, 0, 0, tzinfo=timezone.utc), + context="test context", + history="test history", + new_turn="test new turn", + ) + + # Verify that we get a mock result, not an actual LLM call + assert result is not None + # The result should have the attributes we expect from our mock + assert hasattr(result, "explicit") + assert hasattr(result, "deductive") + assert hasattr(result, "_response") + + +@pytest.mark.asyncio +async def test_summarizer_decorated_functions_with_mock(): + """Test that summarizer decorated functions work with our mock""" + # Import functions that we know are decorated with honcho_llm_call + from src.utils.summarizer import create_long_summary, create_short_summary + + # Create mock messages for testing + mock_message = MagicMock(spec=Message) + mock_message.content = "Test message content" + mock_message.peer_name = "test_peer" + mock_messages = cast(list[Message], [mock_message]) + + # Call the decorated functions - these should use our mock + short_result = await create_short_summary( + messages=mock_messages, input_tokens=100, previous_summary="Previous summary" + ) + + long_result = await create_long_summary( + messages=mock_messages, previous_summary="Previous summary" + ) + + # Verify that we get mock results, not actual LLM calls + assert short_result is not None + assert long_result is not None + # For functions with return_call_response=True, we should get a string or object with content + # The existing mock returns a string, so we check if it's a string + assert isinstance(short_result, str | object) + assert isinstance(long_result, str | object) + # If it's not a string, check for content attribute + if not isinstance(short_result, str): + assert hasattr(short_result, "content") + if not isinstance(long_result, str): + assert hasattr(long_result, "content")