Spaces:
Sleeping
Sleeping
File size: 5,358 Bytes
cc678b9 | 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 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 | import pytest
from app.config import Settings
from app.providers import get_providers
from app.providers.base import ASRProvider, GlossaryEntry, MTProvider, ProviderOutputError
from app.providers.claude_mt import ClaudeMTProvider, parse_mt_output
from app.providers.claude_reviewer import ClaudeReviewerProvider, parse_review_output
from app.providers.openai_asr import OpenAIASRProvider, confidence_from_logprobs
class TextBlock:
def __init__(self, text: str) -> None:
self.text = text
class Message:
def __init__(self, text: str) -> None:
self.content = [TextBlock(text)]
class FakeMessages:
def __init__(self, text: str) -> None:
self.text = text
self.kwargs = {}
def create(self, **kwargs):
self.kwargs = kwargs
return Message(self.text)
class FakeAnthropicClient:
def __init__(self, text: str) -> None:
self.messages = FakeMessages(text)
class FakeTranscriptions:
def __init__(self) -> None:
self.calls = 0
self.kwargs = {}
def create(self, **kwargs):
self.calls += 1
self.kwargs = kwargs
return {"text": "xin chao", "logprobs": [{"logprob": -0.1}, {"logprob": -0.3}]}
class FakeAudio:
def __init__(self) -> None:
self.transcriptions = FakeTranscriptions()
class FakeOpenAIClient:
def __init__(self) -> None:
self.audio = FakeAudio()
def test_registry_returns_cloud_contracts() -> None:
providers = get_providers(
Settings(provider_mode="cloud", openai_api_key="openai", anthropic_api_key="anthropic")
)
assert isinstance(providers.asr, ASRProvider)
assert isinstance(providers.mt, MTProvider)
def test_openai_asr_uses_language_hint_and_logprob_confidence() -> None:
client = FakeOpenAIClient()
provider = OpenAIASRProvider(api_key="", client=client)
result = provider.transcribe(b"audio", "vi")
assert result.text == "xin chao"
expected_confidence = confidence_from_logprobs([{"logprob": -0.1}, {"logprob": -0.3}])
assert round(result.confidence, 3) == round(expected_confidence, 3)
assert client.audio.transcriptions.kwargs["model"] == "gpt-4o-transcribe"
assert client.audio.transcriptions.kwargs["language"] == "vi"
assert client.audio.transcriptions.kwargs["include"] == ["logprobs"]
def test_openai_asr_retries_then_raises() -> None:
class FailingTranscriptions:
def __init__(self) -> None:
self.calls = 0
def create(self, **kwargs):
del kwargs
self.calls += 1
raise TimeoutError("slow")
class FailingClient:
def __init__(self) -> None:
self.audio = type("Audio", (), {"transcriptions": FailingTranscriptions()})()
client = FailingClient()
provider = OpenAIASRProvider(api_key="", client=client, attempts=2)
with pytest.raises(RuntimeError, match="ASR transcription failed"):
provider.transcribe(b"audio", "en")
assert client.audio.transcriptions.calls == 2
def test_claude_mt_strict_json_and_glossary_prompt() -> None:
client = FakeAnthropicClient('{"translation":"Take Augmentin after food","confidence":0.93}')
provider = ClaudeMTProvider(api_key="", client=client)
result = provider.translate(
"Uống Augmentin sau ăn",
"vi",
"en",
[GlossaryEntry(term_vi="Augmentin", term_en="Augmentin", kind="drug")],
)
assert result.text == "Take Augmentin after food"
assert result.confidence == 0.93
prompt = client.messages.kwargs["messages"][0]["content"]
assert "Augmentin" in prompt
assert "Never add advice" in client.messages.kwargs["system"]
@pytest.mark.parametrize(
"text",
[
"Take it. You should drink water.",
'{"translation":"x","confidence":2}',
'{"translation":"x","confidence":0.8,"extra":"no"}',
],
)
def test_claude_mt_rejects_malformed_output(text: str) -> None:
with pytest.raises(ProviderOutputError):
parse_mt_output(text)
def test_claude_reviewer_parses_entities() -> None:
client = FakeAnthropicClient(
'{"back_translation":"Uống 0.5 viên",'
'"entities":[{"kind":"dose","source_text":"nửa viên","translated_text":"half a tablet"}],'
'"flags":[]}'
)
provider = ClaudeReviewerProvider(api_key="", client=client)
review = provider.review("Uống nửa viên", "Take half a tablet", "vi", "en")
assert review.back_translation == "Uống 0.5 viên"
assert review.entities[0].kind == "dose"
assert review.entities[0].source_text == "nửa viên"
assert review.entities[0].translated_text == "half a tablet"
@pytest.mark.parametrize(
"text",
[
"{}",
'{"back_translation":"","entities":[],"flags":[]}',
'{"back_translation":"x","entities":[{"kind":"dose","source_span":[2],"translated_span":[0,1]}],"flags":[]}',
'{"back_translation":"x","entities":[{"kind":"dose","source_text":2,"translated_text":"x"}],"flags":[]}',
],
)
def test_claude_reviewer_rejects_malformed_output(text: str) -> None:
with pytest.raises(ProviderOutputError):
parse_review_output(text)
@pytest.mark.live
def test_openai_asr_live_fixture_is_not_ci_default() -> None:
pytest.skip("live golden-audio fixture runs only with real API keys and committed audio")
|