File size: 3,446 Bytes
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 | import asyncio
import pydantic
import pytest
from app.errors import AnalysisNotFoundError, InvalidTranscriptError, LLMCompletionError
from app.ports import LLm
from app.repositories import InMemoryTranscriptAnalysisRepository
from app.services import TranscriptAnalysisService
class FakeLLM(LLm):
def __init__(self) -> None:
self.user_prompts: list[str] = []
self.async_user_prompts: list[str] = []
def run_completion(
self,
system_prompt: str,
user_prompt: str,
dto: type[pydantic.BaseModel],
) -> pydantic.BaseModel:
self.user_prompts.append(user_prompt)
return dto(summary="A concise summary.", action_items=["Follow up", "Share notes"])
async def run_completion_async(
self,
system_prompt: str,
user_prompt: str,
dto: type[pydantic.BaseModel],
) -> pydantic.BaseModel:
self.async_user_prompts.append(user_prompt)
return dto(summary="A concise summary.", action_items=["Follow up", "Share notes"])
class FailingLLM(LLm):
def run_completion(
self,
system_prompt: str,
user_prompt: str,
dto: type[pydantic.BaseModel],
) -> pydantic.BaseModel:
raise RuntimeError("provider unavailable")
async def run_completion_async(
self,
system_prompt: str,
user_prompt: str,
dto: type[pydantic.BaseModel],
) -> pydantic.BaseModel:
raise RuntimeError("provider unavailable")
def build_service(llm: LLm | None = None) -> TranscriptAnalysisService:
return TranscriptAnalysisService(llm or FakeLLM(), InMemoryTranscriptAnalysisRepository())
def test_analyze_returns_and_persists_result() -> None:
llm = FakeLLM()
service = build_service(llm)
analysis = service.analyze(" Discuss launch plan. ")
assert analysis.id
assert analysis.summary == "A concise summary."
assert analysis.action_items == ("Follow up", "Share notes")
assert service.get(analysis.id) == analysis
assert "Discuss launch plan." in llm.user_prompts[0]
def test_analyze_rejects_empty_transcript() -> None:
service = build_service()
with pytest.raises(InvalidTranscriptError):
service.analyze(" ")
def test_get_raises_when_analysis_is_missing() -> None:
service = build_service()
with pytest.raises(AnalysisNotFoundError):
service.get("missing-id")
def test_analyze_many_processes_and_persists_all_results() -> None:
llm = FakeLLM()
service = build_service(llm)
analyses = asyncio.run(service.analyze_many(["First transcript", "Second transcript"]))
assert len(analyses) == 2
assert len({analysis.id for analysis in analyses}) == 2
assert [service.get(analysis.id) for analysis in analyses] == analyses
assert len(llm.async_user_prompts) == 2
assert llm.user_prompts == []
def test_analyze_many_rejects_empty_list() -> None:
service = build_service()
with pytest.raises(InvalidTranscriptError):
asyncio.run(service.analyze_many([]))
def test_llm_errors_are_wrapped() -> None:
service = build_service(FailingLLM())
with pytest.raises(LLMCompletionError):
service.analyze("Discuss launch plan.")
def test_analyze_many_wraps_llm_errors() -> None:
service = build_service(FailingLLM())
with pytest.raises(LLMCompletionError):
asyncio.run(service.analyze_many(["Discuss launch plan."]))
|