"""Tests for the logging system.""" import json import pytest import tempfile import shutil from pathlib import Path from datetime import datetime, timezone from src.logging.logger import InteractionLogger, get_logger, InteractionTracker from src.logging.schema import ( InteractionLog, ToolCall, Citation, QualityRatings, Labels, ErrorDetail, ErrorCategory, Severity ) class TestLoggerSingleton: """Test logger singleton pattern.""" def test_logger_singleton(self): """Verify singleton pattern works.""" logger1 = get_logger() logger2 = get_logger() assert logger1 is logger2 def test_multiple_instances_same_object(self): """Verify multiple instances reference same object.""" logger1 = InteractionLogger() logger2 = InteractionLogger() assert logger1 is logger2 class TestBasicLogging: """Test basic logging functionality.""" def setup_method(self): """Set up test fixtures.""" self.temp_dir = tempfile.mkdtemp() self.logger = InteractionLogger() self.logger.log_dir = Path(self.temp_dir) def teardown_method(self): """Clean up test fixtures.""" shutil.rmtree(self.temp_dir, ignore_errors=True) def test_basic_interaction_log(self): """Test logging a basic interaction.""" self.logger.log_interaction( session_id="test-session-1", prompt="What is SBIR?", response="SBIR is the Small Business Innovation Research program.", model_version="gpt-5", lat_ms=1500 ) # Check file was created log_file = Path(self.temp_dir) / f"interactions_{datetime.now(timezone.utc).strftime('%Y%m%d')}.jsonl" assert log_file.exists() # Check log content with open(log_file, "r") as f: lines = f.readlines() assert len(lines) == 1 log_data = json.loads(lines[0]) assert log_data["session_id"] == "test-session-1" assert log_data["prompt"] == "What is SBIR?" assert log_data["response"] == "SBIR is the Small Business Innovation Research program." assert log_data["lat_ms"] == 1500 def test_interaction_with_tool_calls(self): """Test logging with tool calls.""" tool_calls = [ ToolCall(name="search_grants", ok=True, lat_ms=500, params={"query": "SBIR"}), ToolCall(name="check_eligibility", ok=True, lat_ms=200, params={"company_size": 50}) ] self.logger.log_interaction( session_id="test-session-2", prompt="Am I eligible?", response="Yes, you are eligible.", tool_calls=tool_calls ) log_file = Path(self.temp_dir) / f"interactions_{datetime.now(timezone.utc).strftime('%Y%m%d')}.jsonl" with open(log_file, "r") as f: log_data = json.loads(f.readlines()[-1]) assert len(log_data["tool_calls"]) == 2 assert log_data["tool_calls"][0]["name"] == "search_grants" def test_interaction_with_citations(self): """Test logging with citations.""" citations = [ Citation(doc_id="doc1", section="Section 1", confidence=0.95), Citation(doc_id="doc2", section="Section 2", confidence=0.85) ] self.logger.log_interaction( session_id="test-session-3", prompt="What is the eligibility criteria?", response="Based on the documents...", citations=citations ) log_file = Path(self.temp_dir) / f"interactions_{datetime.now(timezone.utc).strftime('%Y%m%d')}.jsonl" with open(log_file, "r") as f: log_data = json.loads(f.readlines()[-1]) assert log_data["citations_present"] is True assert len(log_data["citations"]) == 2 def test_interaction_with_error(self): """Test logging with errors.""" error = ErrorDetail( error_type="api_error", error_category=ErrorCategory.external_api, severity=Severity.high, message="API request failed" ) self.logger.log_interaction( session_id="test-session-4", prompt="Test query", error=error ) log_file = Path(self.temp_dir) / f"interactions_{datetime.now(timezone.utc).strftime('%Y%m%d')}.jsonl" with open(log_file, "r") as f: log_data = json.loads(f.readlines()[-1]) assert log_data["error"]["error_type"] == "api_error" assert log_data["error"]["severity"] == "high" class TestInteractionTracker: """Test InteractionTracker context manager.""" def setup_method(self): """Set up test fixtures.""" self.temp_dir = tempfile.mkdtemp() self.logger = InteractionLogger() self.logger.log_dir = Path(self.temp_dir) def teardown_method(self): """Clean up test fixtures.""" shutil.rmtree(self.temp_dir, ignore_errors=True) def test_tracker_context_manager(self): """Test tracker context manager.""" import time with self.logger.track_interaction("test-session-5", model_version="gpt-5") as tracker: tracker.set_response("This is the answer") tracker.add_retrieved_context("doc1") tracker.add_citation(Citation(doc_id="doc1", confidence=0.9)) tracker.add_tool_call(ToolCall(name="search", ok=True, lat_ms=100)) time.sleep(0.001) # Small delay to ensure measurable latency # Verify log was created log_file = Path(self.temp_dir) / f"interactions_{datetime.now(timezone.utc).strftime('%Y%m%d')}.jsonl" with open(log_file, "r") as f: log_data = json.loads(f.readlines()[-1]) assert log_data["response"] == "This is the answer" assert "doc1" in log_data["retrieved_ctx_ids"] assert len(log_data["citations"]) == 1 assert len(log_data["tool_calls"]) == 1 assert log_data["lat_ms"] > 0 class TestFeedbackLogging: """Test feedback logging.""" def setup_method(self): """Set up test fixtures.""" self.temp_dir = tempfile.mkdtemp() self.logger = InteractionLogger() self.logger.log_dir = Path(self.temp_dir) self.logger.feedback_dir = Path(self.temp_dir) def teardown_method(self): """Clean up test fixtures.""" shutil.rmtree(self.temp_dir, ignore_errors=True) def test_log_feedback(self): """Test logging feedback.""" quality = QualityRatings(correctness=5, clarity=4, citations=5) self.logger.log_feedback( session_id="test-session-6", rating_type="thumbs_up", quality_ratings=quality, reason="Good response" ) # Check feedback file was created feedback_file = Path(self.temp_dir) / f"feedback_{datetime.now(timezone.utc).strftime('%Y%m%d')}.jsonl" assert feedback_file.exists() with open(feedback_file, "r") as f: feedback_data = json.loads(f.readlines()[-1]) assert feedback_data["session_id"] == "test-session-6" assert feedback_data["rating_type"] == "thumbs_up" assert feedback_data["quality_ratings"]["correctness"] == 5 assert feedback_data["reason"] == "Good response" class TestLogSchema: """Test logging schema models.""" def test_interaction_log_to_jsonl(self): """Test InteractionLog to_jsonl() method.""" log = InteractionLog( session_id="test-123", prompt="Test query", response="Test response" ) jsonl = log.to_jsonl() assert isinstance(jsonl, str) # Verify it's valid JSON data = json.loads(jsonl) assert data["session_id"] == "test-123" def test_tool_call_model(self): """Test ToolCall model.""" tool_call = ToolCall( name="search", ok=True, lat_ms=500, params={"query": "test"} ) assert tool_call.name == "search" assert tool_call.ok is True assert tool_call.lat_ms == 500 def test_citation_model(self): """Test Citation model.""" citation = Citation( doc_id="doc-123", section="Introduction", confidence=0.95 ) assert citation.doc_id == "doc-123" assert citation.confidence == 0.95