""" Tests for accumulated context management. Tests context accumulation, merging, and limits. """ import json class TestAccumulatedContextInitialization: """Tests for accumulated context initialization.""" def test_new_session_has_empty_accumulated_context( self, authenticated_client ): """Test that new sessions have initialized empty context.""" response = authenticated_client.post( "/api/chat/sessions", json={"initial_query": "Test"}, content_type="application/json", ) session_id = json.loads(response.data)["session_id"] get_resp = authenticated_client.get(f"/api/chat/sessions/{session_id}") data = json.loads(get_resp.data) ctx = data["session"].get("accumulated_context", {}) # Should have initialized fields if ctx: # If API returns it assert isinstance(ctx.get("key_entities", []), list) assert isinstance(ctx.get("topics", []), list) class TestAccumulatedContextStructure: """Tests for accumulated context data structure.""" def test_context_structure_from_manager(self): """Test context structure created by ChatContextManager.""" from src.local_deep_research.chat.context import ChatContextManager messages = [ {"role": "user", "content": "Test query", "message_type": "query"} ] accumulated = { "key_entities": ["entity1", "entity2"], "topics": ["topic1"], "summary": "Previous summary", } manager = ChatContextManager("test-session", messages, accumulated) # Entities and topics should be accessible entities = manager._get_key_entities() topics = manager._get_topics() assert "entity1" in entities assert "entity2" in entities assert "topic1" in topics def test_context_limits_entities(self): """Test that entities are limited to 20.""" from src.local_deep_research.chat.context import ChatContextManager # Create more than 20 entities many_entities = [f"entity{i}" for i in range(30)] accumulated = {"key_entities": many_entities} manager = ChatContextManager("test-session", [], accumulated) entities = manager._get_key_entities() assert len(entities) <= 20 def test_context_limits_topics(self): """Test that topics are limited to 10.""" from src.local_deep_research.chat.context import ChatContextManager # Create more than 10 topics many_topics = [f"topic{i}" for i in range(15)] accumulated = {"topics": many_topics} manager = ChatContextManager("test-session", [], accumulated) topics = manager._get_topics() assert len(topics) <= 10 class TestContextMerging: """Tests for merging context updates.""" def test_merge_entities_deduplicates(self): """Test that merging entities removes duplicates.""" from src.local_deep_research.chat.context import ChatContextManager accumulated = {"key_entities": ["entity1", "entity2"]} manager = ChatContextManager("test-session", [], accumulated) # Get updates with overlapping and new entities updates = manager.extract_context_updates("Content about entity1") # new_entities would normally contain extracted entities; # verify the structure is present assert "new_entities" in updates class TestContextSummary: """Tests for context summary handling.""" def test_summary_created_from_content(self): """Test that summaries are created from content.""" from src.local_deep_research.chat.context import ChatContextManager manager = ChatContextManager("test-session", [], {}) content = """First paragraph with some introduction. Second paragraph with main content that should be captured. Third paragraph with more details.""" summary = manager._create_summary(content) # Should capture meaningful content assert len(summary) > 0 assert len(summary) <= 310 # Max 300 + "..." def test_summary_skips_headers(self): """Test that summary skips markdown headers.""" from src.local_deep_research.chat.context import ChatContextManager manager = ChatContextManager("test-session", [], {}) content = """# Header Title This is the actual content paragraph that should be included in the summary. ## Another Header More content here.""" summary = manager._create_summary(content) # Should skip headers and get content assert not summary.startswith("#") def test_summary_handles_short_paragraphs(self): """Test that summary handles content with short paragraphs.""" from src.local_deep_research.chat.context import ChatContextManager manager = ChatContextManager("test-session", [], {}) content = """Short. Another short. This is a longer paragraph that meets the minimum length requirement for inclusion in the summary.""" summary = manager._create_summary(content) # Should get the longer paragraph assert "longer paragraph" in summary class TestContextWithRealMessages: """Tests with realistic message patterns.""" def test_context_with_multiple_turns(self, authenticated_client): """Test context building with multiple conversation turns.""" # Create session create_resp = authenticated_client.post( "/api/chat/sessions", json={"initial_query": "What is machine learning?"}, content_type="application/json", ) session_id = json.loads(create_resp.data)["session_id"] # Simulate multi-turn conversation turns = [ "Tell me about supervised learning", "How does neural network training work?", "Explain backpropagation", "What are common activation functions?", "Compare ReLU and sigmoid", ] for turn in turns: authenticated_client.post( f"/api/chat/sessions/{session_id}/messages", json={"content": turn, "trigger_research": False}, content_type="application/json", ) # Get all messages messages_resp = authenticated_client.get( f"/api/chat/sessions/{session_id}/messages" ) data = json.loads(messages_resp.data) # All messages should be retrievable assert len(data["messages"]) == 5 def test_context_preserves_message_order(self, authenticated_client): """Test that context preserves message ordering.""" # Create session create_resp = authenticated_client.post( "/api/chat/sessions", json={"initial_query": "Test"}, content_type="application/json", ) session_id = json.loads(create_resp.data)["session_id"] # Add numbered messages for i in range(5): authenticated_client.post( f"/api/chat/sessions/{session_id}/messages", json={ "content": f"Message number {i}", "trigger_research": False, }, content_type="application/json", ) # Get messages messages_resp = authenticated_client.get( f"/api/chat/sessions/{session_id}/messages" ) data = json.loads(messages_resp.data) # Verify order for i, msg in enumerate(data["messages"]): assert f"Message number {i}" in msg["content"] assert msg["sequence_number"] == i + 1 class TestEdgeCases: """Edge case tests for accumulated context.""" def test_empty_accumulated_context_handled(self): """Test handling of None/empty accumulated context.""" from src.local_deep_research.chat.context import ChatContextManager manager1 = ChatContextManager("test", [], None) manager2 = ChatContextManager("test", [], {}) # Both should work without errors ctx1 = manager1.build_research_context() ctx2 = manager2.build_research_context() assert ctx1["key_entities"] == [] assert ctx2["key_entities"] == [] def test_special_characters_in_context(self): """Test handling of special characters in context.""" from src.local_deep_research.chat.context import ChatContextManager accumulated = { "key_entities": ["