111 lines
3.5 KiB
Python
111 lines
3.5 KiB
Python
from __future__ import annotations
|
|
|
|
import pytest
|
|
|
|
from semantic.llm_client import AnthropicSemanticClient, EnrichmentUnavailable
|
|
|
|
RateLimitError = type("RateLimitError", (Exception,), {})
|
|
|
|
|
|
class FakeMessages:
|
|
def __init__(self, *, response: object | None = None, error: Exception | None = None):
|
|
self.response = response
|
|
self.error = error
|
|
self.calls: list[dict[str, object]] = []
|
|
|
|
def create(self, **kwargs: object) -> object:
|
|
self.calls.append(kwargs)
|
|
if self.error:
|
|
raise self.error
|
|
return self.response
|
|
|
|
|
|
class FakeTransport:
|
|
def __init__(self, messages: FakeMessages) -> None:
|
|
self.messages = messages
|
|
|
|
|
|
def _payload() -> dict[str, object]:
|
|
return {
|
|
"page": "Chat",
|
|
"intents": ["send a message"],
|
|
"widgets": [{"element_id": "send", "purpose": "send message"}],
|
|
}
|
|
|
|
|
|
def test_anthropic_semantic_client_requests_schema_output_and_prompt_cache() -> None:
|
|
messages = FakeMessages(response={"content": [{"parsed": _payload()}]})
|
|
client = AnthropicSemanticClient(model="test-model", transport=FakeTransport(messages))
|
|
|
|
result = client.enrich({"screen": {"width": 10, "height": 20}}, timeout=1.5)
|
|
|
|
assert result == _payload()
|
|
assert len(messages.calls) == 1
|
|
call = messages.calls[0]
|
|
assert call["model"] == "test-model"
|
|
assert call["timeout"] == 1.5
|
|
assert call["output_config"] == {
|
|
"format": {
|
|
"type": "json_schema",
|
|
"schema": {
|
|
"type": "object",
|
|
"additionalProperties": False,
|
|
"required": ["page", "intents", "widgets"],
|
|
"properties": {
|
|
"page": {"type": "string", "minLength": 1},
|
|
"intents": {
|
|
"type": "array",
|
|
"items": {"type": "string", "minLength": 1},
|
|
},
|
|
"widgets": {
|
|
"type": "array",
|
|
"items": {
|
|
"type": "object",
|
|
"additionalProperties": False,
|
|
"required": ["element_id", "purpose"],
|
|
"properties": {
|
|
"element_id": {"type": "string", "minLength": 1},
|
|
"purpose": {"type": "string", "minLength": 1},
|
|
},
|
|
},
|
|
},
|
|
},
|
|
},
|
|
}
|
|
}
|
|
assert call["system"][0]["cache_control"] == {"type": "ephemeral"} # type: ignore[index]
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"error",
|
|
[
|
|
TimeoutError("timed out"),
|
|
RateLimitError("rate limited"),
|
|
],
|
|
)
|
|
def test_anthropic_semantic_client_maps_transport_failures(
|
|
error: Exception,
|
|
) -> None:
|
|
messages = FakeMessages(error=error)
|
|
client = AnthropicSemanticClient(transport=FakeTransport(messages))
|
|
|
|
with pytest.raises(EnrichmentUnavailable):
|
|
client.enrich({"screen": {"width": 10, "height": 20}}, timeout=1)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"response",
|
|
[
|
|
{"content": [{"text": "not json"}]},
|
|
{"page": "Chat", "intents": ["send a message"]},
|
|
],
|
|
)
|
|
def test_anthropic_semantic_client_maps_malformed_or_schema_invalid_responses(
|
|
response: object,
|
|
) -> None:
|
|
messages = FakeMessages(response=response)
|
|
client = AnthropicSemanticClient(transport=FakeTransport(messages))
|
|
|
|
with pytest.raises(EnrichmentUnavailable):
|
|
client.enrich({"screen": {"width": 10, "height": 20}}, timeout=1)
|