generated from john/python-template
248 lines
9.4 KiB
Python
248 lines
9.4 KiB
Python
"""Tests for transcription.providers.openrouter."""
|
|
|
|
import asyncio
|
|
import logging
|
|
from types import SimpleNamespace
|
|
from typing import cast
|
|
|
|
import pytest
|
|
from openrouter import OpenRouter
|
|
|
|
from transcription.config import Settings
|
|
from transcription.providers.base import ProviderError
|
|
from transcription.providers.base import ProviderResponseError
|
|
from transcription.providers.openrouter import DEFAULT_OPENROUTER_MODEL
|
|
from transcription.providers.openrouter import OpenRouterTranscriptionProvider
|
|
|
|
|
|
class _FakeChat:
|
|
def __init__(self, response=None, error: BaseException | None = None):
|
|
self._response = response
|
|
self._error = error
|
|
self.calls = []
|
|
|
|
async def send_async(self, **kwargs):
|
|
self.calls.append(kwargs)
|
|
if self._error:
|
|
raise self._error
|
|
return self._response
|
|
|
|
|
|
class _FakeClient:
|
|
def __init__(self, response=None, error: BaseException | None = None):
|
|
self.chat = _FakeChat(response=response, error=error)
|
|
|
|
|
|
def _fake_client(response=None, error: BaseException | None = None) -> OpenRouter:
|
|
"""Return a stub typed as the SDK client the provider declares."""
|
|
return cast("OpenRouter", _FakeClient(response=response, error=error))
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestOpenRouterProviderInit:
|
|
"""Verify OpenRouter provider initialization behavior."""
|
|
|
|
def test_model_falls_back_to_default_when_unset(self):
|
|
"""Provider uses adapter default model when provider_model is None."""
|
|
settings = Settings(openrouter_api_key="test-key", provider_model=None)
|
|
provider = OpenRouterTranscriptionProvider(settings=settings, client=_fake_client())
|
|
assert provider.model == DEFAULT_OPENROUTER_MODEL
|
|
|
|
def test_model_uses_configured_value(self):
|
|
"""Provider uses configured provider_model when present."""
|
|
settings = Settings(openrouter_api_key="test-key", provider_model="vendor/custom-model")
|
|
provider = OpenRouterTranscriptionProvider(settings=settings, client=_fake_client())
|
|
assert provider.model == "vendor/custom-model"
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestOpenRouterProviderTranscribe:
|
|
"""Verify OpenRouter request construction and response parsing."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_includes_optional_referer_and_title_when_set(self):
|
|
"""Transcribe sends app attribution fields when configured."""
|
|
response = {"model": "vendor/model-a", "choices": [{"message": {"content": "Transcript text"}}]}
|
|
client = _FakeClient(response=response)
|
|
settings = Settings(
|
|
openrouter_api_key="test-key",
|
|
openrouter_http_referer="https://example.test",
|
|
openrouter_app_title="Transcription App",
|
|
)
|
|
provider = OpenRouterTranscriptionProvider(settings=settings, client=cast("OpenRouter", client))
|
|
|
|
result = await provider.transcribe(
|
|
prompt_text="Prompt body",
|
|
image_bytes=b"img-bytes",
|
|
mime_type="image/png",
|
|
)
|
|
|
|
send_call = client.chat.calls[0]
|
|
assert send_call["http_referer"] == "https://example.test"
|
|
assert send_call["x_open_router_title"] == "Transcription App"
|
|
assert result.text == "Transcript text"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_includes_temperature_and_top_p_when_provided(self):
|
|
"""Transcribe passes configured sampling parameters through to OpenRouter."""
|
|
response = {"model": "vendor/model-a", "choices": [{"message": {"content": "Transcript text"}}]}
|
|
client = _FakeClient(response=response)
|
|
provider = OpenRouterTranscriptionProvider(
|
|
settings=Settings(openrouter_api_key="test-key"),
|
|
client=cast("OpenRouter", client),
|
|
)
|
|
|
|
result = await provider.transcribe(
|
|
prompt_text="Prompt body",
|
|
image_bytes=b"img-bytes",
|
|
mime_type="image/png",
|
|
temperature=0.2,
|
|
top_p=0.85,
|
|
)
|
|
|
|
send_call = client.chat.calls[0]
|
|
assert send_call["temperature"] == 0.2
|
|
assert send_call["top_p"] == 0.85
|
|
assert result.temperature == 0.2
|
|
assert result.top_p == 0.85
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_parses_successful_response_text(self):
|
|
"""Transcribe returns normalized text from a valid response payload."""
|
|
response = {
|
|
"model": "vendor/model-b",
|
|
"choices": [
|
|
{
|
|
"message": {"content": [{"text": "Line 1"}, {"text": "Line 2"}]},
|
|
"finish_reason": "stop",
|
|
}
|
|
],
|
|
"usage": {"prompt_tokens": 10, "completion_tokens": 25, "total_tokens": 35},
|
|
}
|
|
provider = OpenRouterTranscriptionProvider(
|
|
settings=Settings(openrouter_api_key="test-key"),
|
|
client=_fake_client(response=response),
|
|
)
|
|
|
|
result = await provider.transcribe(
|
|
prompt_text="Prompt body",
|
|
image_bytes=b"img-bytes",
|
|
mime_type="image/jpeg",
|
|
)
|
|
|
|
assert result.text == "Line 1\nLine 2"
|
|
assert result.provider == "openrouter"
|
|
assert result.prompt_name is None
|
|
assert result.model == "vendor/model-b"
|
|
assert result.metadata_payload() == {
|
|
"finish_reason": "stop",
|
|
"usage": {"input_tokens": 10, "output_tokens": 25, "total_tokens": 35},
|
|
}
|
|
assert result.raw_api_response == response
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_maps_sdk_exception_to_provider_error(self):
|
|
"""Transcribe converts SDK failures to ProviderError."""
|
|
provider = OpenRouterTranscriptionProvider(
|
|
settings=Settings(openrouter_api_key="test-key"),
|
|
client=_fake_client(error=RuntimeError("network down")),
|
|
)
|
|
|
|
with pytest.raises(ProviderError):
|
|
await provider.transcribe(
|
|
prompt_text="Prompt body",
|
|
image_bytes=b"img-bytes",
|
|
mime_type="image/png",
|
|
)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_preserves_caller_cancellation(self):
|
|
"""Caller and shutdown cancellation must not be relabeled as a timeout."""
|
|
provider = OpenRouterTranscriptionProvider(
|
|
settings=Settings(openrouter_api_key="test-key"),
|
|
client=_fake_client(error=asyncio.CancelledError()),
|
|
)
|
|
|
|
with pytest.raises(asyncio.CancelledError):
|
|
await provider.transcribe(
|
|
prompt_text="Prompt body",
|
|
image_bytes=b"img-bytes",
|
|
mime_type="image/png",
|
|
)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_sends_pdf_as_file_content(self):
|
|
"""PDF payloads use OpenRouter's file content contract."""
|
|
response = {"model": "vendor/model-a", "choices": [{"message": {"content": "Transcript text"}}]}
|
|
client = _FakeClient(response=response)
|
|
provider = OpenRouterTranscriptionProvider(
|
|
settings=Settings(openrouter_api_key="test-key"),
|
|
client=cast("OpenRouter", client),
|
|
)
|
|
|
|
await provider.transcribe(
|
|
prompt_text="Prompt body",
|
|
image_bytes=b"pdf-bytes",
|
|
mime_type="application/pdf",
|
|
)
|
|
|
|
content = client.chat.calls[0]["messages"][0]["content"]
|
|
assert content[1]["type"] == "file"
|
|
assert content[1]["file"]["filename"] == "source.pdf"
|
|
assert content[1]["file"]["file_data"].startswith("data:application/pdf;base64,")
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_raises_on_empty_or_invalid_response(self):
|
|
"""Transcribe raises ProviderResponseError for missing completion text."""
|
|
provider = OpenRouterTranscriptionProvider(
|
|
settings=Settings(openrouter_api_key="test-key"),
|
|
client=_fake_client(response=SimpleNamespace(choices=[])),
|
|
)
|
|
|
|
with pytest.raises(ProviderResponseError):
|
|
await provider.transcribe(
|
|
prompt_text="Prompt body",
|
|
image_bytes=b"img-bytes",
|
|
mime_type="image/png",
|
|
)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ignores_invalid_token_metadata_without_discarding_transcript(self):
|
|
response = {
|
|
"model": "vendor/model-a",
|
|
"choices": [{"message": {"content": "Transcript text"}}],
|
|
"usage": {"prompt_tokens": -1},
|
|
}
|
|
provider = OpenRouterTranscriptionProvider(
|
|
settings=Settings(openrouter_api_key="test-key"),
|
|
client=_fake_client(response=response),
|
|
)
|
|
|
|
result = await provider.transcribe(
|
|
prompt_text="Prompt body",
|
|
image_bytes=b"img-bytes",
|
|
mime_type="image/png",
|
|
)
|
|
|
|
assert result.text == "Transcript text"
|
|
assert result.metadata_payload() is None
|
|
assert result.raw_api_response == response
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_logs_when_request_manifest_is_omitted_without_source_reference(self, caplog):
|
|
response = {"model": "vendor/model-a", "choices": [{"message": {"content": "Transcript text"}}]}
|
|
provider = OpenRouterTranscriptionProvider(
|
|
settings=Settings(openrouter_api_key="test-key"),
|
|
client=_fake_client(response=response),
|
|
)
|
|
|
|
with caplog.at_level(logging.WARNING):
|
|
result = await provider.transcribe(
|
|
prompt_text="Prompt body",
|
|
image_bytes=b"img-bytes",
|
|
mime_type="image/png",
|
|
)
|
|
|
|
assert result.request_manifest is None
|
|
assert "request manifest omitted" in caplog.text.lower()
|