File size: 4,189 Bytes
62151d3 4b624a7 62151d3 4b624a7 1379bdf 62151d3 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 | from pydantic import BaseModel
import pytest
from fastapi.testclient import TestClient
from app.errors import ConfigurationError
from app.main import app, get_analysis_service
from app.ports import LLm
from app.repositories import InMemoryTranscriptAnalysisRepository
from app.services import TranscriptAnalysisService
class FakeLLM(LLm):
def run_completion(
self,
system_prompt: str,
user_prompt: str,
dto: type[BaseModel],
) -> BaseModel:
return dto(summary="API summary.", action_items=["Confirm owner", "Set deadline"])
async def run_completion_async(
self,
system_prompt: str,
user_prompt: str,
dto: type[BaseModel],
) -> BaseModel:
return dto(summary="API summary.", action_items=["Confirm owner", "Set deadline"])
@pytest.fixture
def client() -> TestClient:
service = TranscriptAnalysisService(FakeLLM(), InMemoryTranscriptAnalysisRepository())
app.dependency_overrides[get_analysis_service] = lambda: service
with TestClient(app) as test_client:
yield test_client
app.dependency_overrides.clear()
def test_analyze_transcript_success(client: TestClient) -> None:
response = client.get("/analyses", params={"transcript": "Discuss rollout plan."})
assert response.status_code == 200
payload = response.json()
assert payload["id"]
assert payload["summary"] == "API summary."
assert payload["action_items"] == ["Confirm owner", "Set deadline"]
def test_analyze_transcript_rejects_empty_query(client: TestClient) -> None:
response = client.get("/analyses", params={"transcript": " "})
assert response.status_code == 400
assert response.json()["detail"] == "Transcript cannot be empty."
def test_get_transcript_analysis_success(client: TestClient) -> None:
created = client.get("/analyses", params={"transcript": "Discuss rollout plan."}).json()
response = client.get(f"/analyses/{created['id']}")
assert response.status_code == 200
assert response.json() == created
def test_get_transcript_analysis_returns_404(client: TestClient) -> None:
response = client.get("/analyses/missing-id")
assert response.status_code == 404
assert response.json()["detail"] == "Transcript analysis 'missing-id' was not found."
def test_analyze_batch_success(client: TestClient) -> None:
response = client.post(
"/analyses/batch",
json={"transcripts": ["Discuss roadmap.", "Review hiring plan."]},
)
assert response.status_code == 200
payload = response.json()
assert len(payload["items"]) == 2
assert all(item["summary"] == "API summary." for item in payload["items"])
def test_analyze_batch_rejects_empty_list(client: TestClient) -> None:
response = client.post("/analyses/batch", json={"transcripts": []})
assert response.status_code == 400
assert response.json()["detail"] == "At least one transcript is required."
def test_analyze_batch_rejects_empty_transcript(client: TestClient) -> None:
response = client.post("/analyses/batch", json={"transcripts": ["Discuss roadmap.", " "]})
assert response.status_code == 400
assert response.json()["detail"] == "Transcript cannot be empty."
def test_configuration_error_returns_503() -> None:
def raise_configuration_error() -> TranscriptAnalysisService:
raise ConfigurationError(
"OPENAI_API_KEY is not configured. Set it in .env or as an environment variable."
)
app.dependency_overrides[get_analysis_service] = raise_configuration_error
try:
with TestClient(app) as test_client:
response = test_client.get("/analyses", params={"transcript": "Discuss rollout plan."})
finally:
app.dependency_overrides.clear()
assert response.status_code == 503
assert response.json()["detail"] == (
"OPENAI_API_KEY is not configured. Set it in .env or as an environment variable."
)
def test_gradio_ui_is_mounted(client: TestClient) -> None:
response = client.get("/", follow_redirects=True)
assert response.status_code == 200
assert "text/html" in response.headers["content-type"]
|