""" Tests for conversation history management. """ import pytest from fastapi.testclient import TestClient from src.responses.history import ConversationHistory from src.responses.context import ContextWindow from src.responses.schemas import ResponseRequest from src.responses import service @pytest.mark.unit @pytest.mark.asyncio async def test_conversation_id_from_metadata(): """Test conversation ID extraction from metadata.""" history = ConversationHistory() request = ResponseRequest( model="lorem-tester", input=[{"role": "user", "content": "Hello"}], metadata={"conversation_id": "conv_123"} ) conv_id = await history.get_conversation_id(request) assert conv_id == "conv_123" @pytest.mark.unit @pytest.mark.asyncio async def test_conversation_id_generation(): """Test conversation ID generation from first message.""" history = ConversationHistory() request = ResponseRequest( model="lorem-tester", input=[{"role": "user", "content": "Hello"}] # No metadata provided ) conv_id = await history.get_conversation_id(request) assert isinstance(conv_id, str) assert len(conv_id) == 16 # 16 character hex # Same first message should generate same ID request2 = ResponseRequest( model="lorem-tester", input=[{"role": "user", "content": "Hello"}] ) conv_id2 = await history.get_conversation_id(request2) assert conv_id == conv_id2 @pytest.mark.unit def test_conversation_history_tracking(client: TestClient): """Test that conversations are tracked server-side.""" # First request with conversation ID request_data = { "model": "lorem-tester", "input": [{"role": "user", "content": "Hello"}], "metadata": {"conversation_id": "test_conv_001"}, "stream": False } response = client.post("/v1/responses", json=request_data) assert response.status_code == 200 # Second request with same conversation ID request_data2 = { "model": "lorem-tester", "input": [ {"role": "user", "content": "Hello"}, {"role": "assistant", "content": "Hi there!"}, {"role": "user", "content": "How are you?"} ], "metadata": {"conversation_id": "test_conv_001"}, "stream": False } response2 = client.post("/v1/responses", json=request_data2) assert response2.status_code == 200 # Both responses should be successful assert response.json()["status"] == "completed" assert response2.json()["status"] == "completed" @pytest.mark.unit @pytest.mark.asyncio async def test_conversation_history_retrieval(): """Test retrieving conversation history.""" history = ConversationHistory() request = ResponseRequest( model="lorem-tester", input=[{"role": "user", "content": "Test"}], metadata={"conversation_id": "test_retrieve"} ) conv_id = await history.get_conversation_id(request) # Initially empty retrieved = await history.get_history(conv_id) assert len(retrieved) == 0 # Add mock response from src.responses.schemas import Response, ResponseUsage, MessageOutputItem, OutputTextContent mock_response = Response( id="resp_123", created_at=1234567890, model="lorem-tester", status="completed", output=[ MessageOutputItem( id="msg_1", content=[OutputTextContent(text="Test response")] ) ], usage=ResponseUsage( input_tokens=10, output_tokens=5, reasoning_tokens=0, total_tokens=15 ) ) await history.add_response(conv_id, mock_response) # Should now have 1 response retrieved = await history.get_history(conv_id) assert len(retrieved) == 1 assert retrieved[0].id == "resp_123" @pytest.mark.unit @pytest.mark.asyncio async def test_conversation_history_trimming(): """Test that history is trimmed to max_turns.""" history = ConversationHistory(max_turns=3) from src.responses.schemas import Response, ResponseUsage, MessageOutputItem, OutputTextContent conv_id = "test_trim" # Add 5 responses (more than max_turns) for i in range(5): response = Response( id=f"resp_{i}", created_at=1234567890 + i, model="lorem-tester", status="completed", output=[ MessageOutputItem( id=f"msg_{i}", content=[OutputTextContent(text=f"Response {i}")] ) ], usage=ResponseUsage( input_tokens=10, output_tokens=5, reasoning_tokens=0, total_tokens=15 ) ) await history.add_response(conv_id, response) # Should only keep last 3 retrieved = await history.get_history(conv_id) assert len(retrieved) == 3 assert retrieved[0].id == "resp_2" # Oldest kept assert retrieved[1].id == "resp_3" assert retrieved[2].id == "resp_4" # Most recent @pytest.mark.unit @pytest.mark.asyncio async def test_clear_conversation(): """Test clearing conversation history.""" history = ConversationHistory() from src.responses.schemas import Response, ResponseUsage, MessageOutputItem, OutputTextContent conv_id = "test_clear" # Add a response response = Response( id="resp_123", created_at=1234567890, model="lorem-tester", status="completed", output=[ MessageOutputItem( id="msg_1", content=[OutputTextContent(text="Test")] ) ], usage=ResponseUsage( input_tokens=10, output_tokens=5, reasoning_tokens=0, total_tokens=15 ) ) await history.add_response(conv_id, response) # Verify it exists assert len(await history.get_history(conv_id)) == 1 # Clear it cleared = await history.clear_conversation(conv_id) assert cleared is True # Should be empty now assert len(await history.get_history(conv_id)) == 0 # Clearing non-existent conversation should return False cleared_again = await history.clear_conversation(conv_id) assert cleared_again is False @pytest.mark.unit @pytest.mark.asyncio async def test_conversation_count(): """Test conversation count tracking.""" history = ConversationHistory() initial_count = await history.get_conversation_count() # Add responses to 3 different conversations from src.responses.schemas import Response, ResponseUsage, MessageOutputItem, OutputTextContent for i in range(3): response = Response( id=f"resp_{i}", created_at=1234567890, model="lorem-tester", status="completed", output=[ MessageOutputItem( id=f"msg_{i}", content=[OutputTextContent(text="Test")] ) ], usage=ResponseUsage( input_tokens=10, output_tokens=5, reasoning_tokens=0, total_tokens=15 ) ) await history.add_response(f"conv_{i}", response) new_count = await history.get_conversation_count() assert new_count == initial_count + 3 @pytest.mark.unit @pytest.mark.asyncio async def test_context_window_token_counting(): """Test token counting in context window.""" context = ContextWindow(max_tokens=4096) # Test string counting text = "Hello world! " * 100 # ~1200 characters tokens = await context.count_tokens([text]) assert tokens > 0 assert tokens == len(text) // 4 # Approximate # Test dict counting message = {"role": "user", "content": "Test message"} tokens = await context.count_tokens([message]) assert tokens > 0 @pytest.mark.unit @pytest.mark.asyncio async def test_context_window_trimming(): """Test context window trimming to fit.""" context = ContextWindow(max_tokens=100) # Create items that exceed limit items = [ "This is a long message " * 20, # ~480 chars = ~120 tokens "Another message " * 10, # ~160 chars = ~40 tokens "Short message" # ~13 chars = ~3 tokens ] # Trim with 10 token reserve trimmed = await context.trim_to_fit(items, reserve_tokens=10) # Should keep only items that fit (90 tokens available) # Most recent first: "Short message" (3 tokens) + "Another message..." (40 tokens) = 43 tokens assert len(trimmed) >= 1 # At least the short message assert "Short message" in trimmed # Most recent always kept if it fits @pytest.mark.unit @pytest.mark.asyncio async def test_context_window_fits_check(): """Test checking if items fit in context.""" context = ContextWindow(max_tokens=100) small_items = ["Hello", "World"] large_items = ["Very long text " * 200] # ~3200 chars = ~800 tokens # Small items should fit assert await context.fits_in_context(small_items, reserve_tokens=10) is True # Large items should not fit assert await context.fits_in_context(large_items, reserve_tokens=10) is False @pytest.mark.unit @pytest.mark.asyncio async def test_context_window_usage_stats(): """Test context window usage statistics.""" context = ContextWindow(max_tokens=1000) items = ["Test message " * 50] # ~650 chars = ~162 tokens stats = await context.get_usage_stats(items, reserve_tokens=100) assert "total_tokens" in stats assert "max_tokens" in stats assert "reserved_tokens" in stats assert "available_tokens" in stats assert "usage_percent" in stats assert "fits" in stats assert stats["max_tokens"] == 1000 assert stats["reserved_tokens"] == 100 assert stats["available_tokens"] == 900 assert isinstance(stats["usage_percent"], (int, float)) @pytest.mark.unit @pytest.mark.asyncio async def test_service_conversation_helpers(): """Test service helper functions for conversation history.""" # Get stats stats = await service.get_conversation_stats() assert "total_conversations" in stats assert "max_turns_per_conversation" in stats # Get context window context = service.get_context_window() assert context.max_tokens == 4096 # Test clearing (should handle non-existent gracefully) result = await service.clear_conversation("non_existent_conv") assert result is False @pytest.mark.unit @pytest.mark.asyncio async def test_streaming_tracks_history(async_client): """Test that streaming responses also track conversation history.""" from httpx import AsyncClient import json request_data = { "model": "lorem-tester", "input": [{"role": "user", "content": "Hello"}], "metadata": {"conversation_id": "stream_test_001"}, "stream": True } async with async_client.stream( "POST", "/v1/responses", json=request_data, timeout=20.0, ) as response: assert response.status_code == 200 # Consume the stream async for line in response.aiter_lines(): if line.startswith("data: ") and "[DONE]" not in line: continue # History should be tracked # (We can't easily verify this without exposing a GET endpoint, # but the integration is tested in the non-streaming tests)