"""Tests for History/Call Record API endpoints""" import pytest import time class TestHistoryAPI: """Test cases for History/Call Record endpoints""" def test_get_history_empty(self, client): """Test getting history when database is empty""" response = client.get("/api/history") assert response.status_code == 200 data = response.json() assert "total" in data assert "list" in data def test_create_call_record(self, client, sample_call_record_data): """Test creating a new call record""" response = client.post("/api/history", json=sample_call_record_data) assert response.status_code == 200 data = response.json() assert data["user_id"] == sample_call_record_data["user_id"] assert data["source"] == sample_call_record_data["source"] assert data["status"] == "connected" assert "id" in data assert "started_at" in data def test_create_call_record_with_assistant(self, client, sample_assistant_data, sample_call_record_data): """Test creating a call record associated with an assistant""" # Create assistant first assistant_response = client.post("/api/assistants", json=sample_assistant_data) assistant_id = assistant_response.json()["id"] # Create call record with assistant sample_call_record_data["assistant_id"] = assistant_id response = client.post("/api/history", json=sample_call_record_data) assert response.status_code == 200 assert response.json()["assistant_id"] == assistant_id def test_get_call_record_by_id(self, client, sample_call_record_data): """Test getting a specific call record by ID""" # Create first create_response = client.post("/api/history", json=sample_call_record_data) record_id = create_response.json()["id"] # Get by ID response = client.get(f"/api/history/{record_id}") assert response.status_code == 200 data = response.json() assert data["id"] == record_id def test_get_call_record_not_found(self, client): """Test getting a non-existent call record""" response = client.get("/api/history/non-existent-id") assert response.status_code == 404 def test_update_call_record(self, client, sample_call_record_data): """Test updating a call record""" # Create first create_response = client.post("/api/history", json=sample_call_record_data) record_id = create_response.json()["id"] # Update update_data = { "status": "completed", "summary": "Test summary", "duration_seconds": 120 } response = client.put(f"/api/history/{record_id}", json=update_data) assert response.status_code == 200 data = response.json() assert data["status"] == "completed" assert data["summary"] == "Test summary" assert data["duration_seconds"] == 120 def test_delete_call_record(self, client, sample_call_record_data): """Test deleting a call record""" # Create first create_response = client.post("/api/history", json=sample_call_record_data) record_id = create_response.json()["id"] # Delete response = client.delete(f"/api/history/{record_id}") assert response.status_code == 200 # Verify deleted get_response = client.get(f"/api/history/{record_id}") assert get_response.status_code == 404 def test_add_transcript(self, client, sample_call_record_data): """Test adding a transcript to a call record""" # Create call record first create_response = client.post("/api/history", json=sample_call_record_data) record_id = create_response.json()["id"] # Add transcript transcript_data = { "turn_index": 0, "speaker": "human", "content": "Hello, I need help", "start_ms": 0, "end_ms": 3000, "confidence": 0.95 } response = client.post( f"/api/history/{record_id}/transcripts", json=transcript_data ) assert response.status_code == 200 data = response.json() assert data["turn_index"] == 0 assert data["speaker"] == "human" assert data["content"] == "Hello, I need help" def test_add_multiple_transcripts(self, client, sample_call_record_data): """Test adding multiple transcripts""" # Create call record first create_response = client.post("/api/history", json=sample_call_record_data) record_id = create_response.json()["id"] # Add human transcript human_transcript = { "turn_index": 0, "speaker": "human", "content": "Hello", "start_ms": 0, "end_ms": 1000 } client.post(f"/api/history/{record_id}/transcripts", json=human_transcript) # Add AI transcript ai_transcript = { "turn_index": 1, "speaker": "ai", "content": "Hello! How can I help you?", "start_ms": 1500, "end_ms": 4000 } client.post(f"/api/history/{record_id}/transcripts", json=ai_transcript) # Verify both transcripts exist response = client.get(f"/api/history/{record_id}") assert response.status_code == 200 data = response.json() assert len(data["transcripts"]) == 2 def test_filter_history_by_status(self, client, sample_call_record_data): """Test filtering history by status""" # Create records with different statuses for i in range(2): data = sample_call_record_data.copy() data["status"] = "connected" if i % 2 == 0 else "missed" client.post("/api/history", json=data) # Filter by status response = client.get("/api/history?status=connected") assert response.status_code == 200 data = response.json() for record in data["list"]: assert record["status"] == "connected" def test_filter_history_by_source(self, client, sample_call_record_data): """Test filtering history by source""" sample_call_record_data["source"] = "external" client.post("/api/history", json=sample_call_record_data) response = client.get("/api/history?source=external") assert response.status_code == 200 data = response.json() for record in data["list"]: assert record["source"] == "external" def test_history_pagination(self, client, sample_call_record_data): """Test history pagination""" # Create multiple records for i in range(5): data = sample_call_record_data.copy() data["source"] = f"source-{i}" client.post("/api/history", json=data) # Test pagination response = client.get("/api/history?page=1&limit=3") assert response.status_code == 200 data = response.json() assert data["total"] == 5 assert len(data["list"]) == 3 def test_transcript_with_emotion(self, client, sample_call_record_data): """Test adding transcript with emotion""" # Create call record first create_response = client.post("/api/history", json=sample_call_record_data) record_id = create_response.json()["id"] # Add transcript with emotion transcript_data = { "turn_index": 0, "speaker": "ai", "content": "Great news!", "start_ms": 0, "end_ms": 2000, "emotion": "happy" } response = client.post( f"/api/history/{record_id}/transcripts", json=transcript_data ) assert response.status_code == 200 assert response.json()["emotion"] == "happy" def test_history_with_cost(self, client, sample_call_record_data): """Test creating history with cost""" sample_call_record_data["cost"] = 0.05 response = client.post("/api/history", json=sample_call_record_data) assert response.status_code == 200 assert response.json()["cost"] == 0.05 def test_history_search(self, client, sample_call_record_data): """Test searching history""" # Create record create_response = client.post("/api/history", json=sample_call_record_data) record_id = create_response.json()["id"] # Add transcript with searchable content transcript_data = { "turn_index": 0, "speaker": "human", "content": "I want to buy a product", "start_ms": 0, "end_ms": 3000 } client.post(f"/api/history/{record_id}/transcripts", json=transcript_data) # Search (this endpoint may not exist yet) response = client.get("/api/history/search?q=product") # This might return 404 if endpoint doesn't exist assert response.status_code in [200, 404]