237 lines
8.9 KiB
Python
237 lines
8.9 KiB
Python
"""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]
|