364 lines
14 KiB
Python
364 lines
14 KiB
Python
"""Jev HTTP 어댑터의 단위 계약."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import copy
|
|
import json
|
|
import unittest
|
|
from types import SimpleNamespace
|
|
from unittest.mock import patch
|
|
|
|
import httpx
|
|
from pydantic import SecretStr
|
|
|
|
from .services import jev_client as jev_module
|
|
from .services.jev_client import (
|
|
EMOTION_DIMENSIONS,
|
|
OPENROUTER_JEV_ENDPOINT,
|
|
TYPESAFE_JEV_ENDPOINT,
|
|
JevClient,
|
|
JevError,
|
|
)
|
|
|
|
|
|
def _answer(score: float = 2.0) -> dict[str, object]:
|
|
return {
|
|
"type": "score",
|
|
"score": score,
|
|
"confidence": 0.8,
|
|
"legend": {str(index): f"level {index}" for index in range(5)},
|
|
"probabilities": {
|
|
"0": 0.0,
|
|
"1": 0.1,
|
|
"2": 0.8,
|
|
"3": 0.1,
|
|
"4": 0.0,
|
|
},
|
|
}
|
|
|
|
|
|
def _response(model: str = "typesafe/jev-1.13-20260917") -> dict[str, object]:
|
|
return {
|
|
"model": model,
|
|
"answers": {dimension: _answer() for dimension in EMOTION_DIMENSIONS},
|
|
"usage": {"input_tokens": 120, "output_tokens": 45, "cost": 0.000019992},
|
|
}
|
|
|
|
|
|
class JevClientTest(unittest.IsolatedAsyncioTestCase):
|
|
def setUp(self) -> None:
|
|
self.calls = 0
|
|
|
|
async def _client(self, handler) -> JevClient:
|
|
client = JevClient(
|
|
provider="openrouter",
|
|
api_key="test-key",
|
|
model="~typesafe/jev-latest",
|
|
timeout_seconds=0.05,
|
|
transport=httpx.MockTransport(handler),
|
|
)
|
|
await client.startup()
|
|
self.addAsyncCleanup(client.shutdown)
|
|
return client
|
|
|
|
async def test_appraise_posts_one_typed_request_and_normalizes_scores(self) -> None:
|
|
async def handler(request: httpx.Request) -> httpx.Response:
|
|
self.calls += 1
|
|
self.assertEqual("POST", request.method)
|
|
self.assertEqual(OPENROUTER_JEV_ENDPOINT, str(request.url))
|
|
self.assertEqual("Bearer test-key", request.headers["Authorization"])
|
|
body = json.loads(request.content)
|
|
self.assertEqual("~typesafe/jev-latest", body["model"])
|
|
self.assertEqual(set(EMOTION_DIMENSIONS), set(body["questions"]))
|
|
for dimension, question in body["questions"].items():
|
|
self.assertEqual("score", question["type"])
|
|
self.assertEqual(5, len(question["criteria"]))
|
|
self.assertIn(dimension, question["instructions"])
|
|
self.assertIn("counselor_utterance", question["instructions"])
|
|
self.assertIn("pinned facts", question["instructions"])
|
|
self.assertLessEqual(len(question["instructions"].split()), 50)
|
|
return httpx.Response(200, json=_response())
|
|
|
|
client = await self._client(handler)
|
|
result = await client.appraise({"turn": "I hear you."})
|
|
|
|
self.assertEqual(1, self.calls)
|
|
self.assertEqual("typesafe/jev-1.13-20260917", result.model)
|
|
self.assertEqual("openrouter", result.provider)
|
|
self.assertEqual(0.000019992, result.cost_usd)
|
|
self.assertEqual(120, result.input_tokens)
|
|
self.assertEqual(45, result.output_tokens)
|
|
self.assertEqual(0.5, result.emotions["anxiety"].score)
|
|
self.assertEqual(0.8, result.emotions["trust"].confidence)
|
|
self.assertEqual(
|
|
(0.0, 0.1, 0.8, 0.1, 0.0),
|
|
result.emotions["anxiety"].probabilities,
|
|
)
|
|
self.assertGreaterEqual(result.latency_ms, 0)
|
|
|
|
async def test_startup_does_not_issue_a_request(self) -> None:
|
|
async def handler(request: httpx.Request) -> httpx.Response:
|
|
self.calls += 1
|
|
return httpx.Response(500)
|
|
|
|
client = await self._client(handler)
|
|
self.assertTrue(client.configured)
|
|
self.assertEqual(0, self.calls)
|
|
|
|
async def test_empty_key_fails_without_external_call(self) -> None:
|
|
async def handler(request: httpx.Request) -> httpx.Response:
|
|
self.calls += 1
|
|
return httpx.Response(200, json=_response())
|
|
|
|
client = JevClient(
|
|
api_key="",
|
|
transport=httpx.MockTransport(handler),
|
|
)
|
|
await client.startup()
|
|
self.addAsyncCleanup(client.shutdown)
|
|
|
|
with self.assertRaisesRegex(JevError, "not_configured"):
|
|
await client.appraise({})
|
|
self.assertEqual(0, self.calls)
|
|
|
|
async def test_status_failures_have_safe_codes_and_no_retry(self) -> None:
|
|
failures = (
|
|
(401, "unauthorized"),
|
|
(402, "insufficient_credits"),
|
|
(403, "forbidden"),
|
|
(404, "model_unavailable"),
|
|
(429, "rate_limited"),
|
|
(529, "overloaded"),
|
|
)
|
|
for status, code in failures:
|
|
with self.subTest(status=status):
|
|
self.calls = 0
|
|
|
|
async def handler(request: httpx.Request, status: int = status) -> httpx.Response:
|
|
self.calls += 1
|
|
return httpx.Response(status, text="sensitive response body")
|
|
|
|
client = await self._client(handler)
|
|
with self.assertRaisesRegex(JevError, code):
|
|
await client.appraise({})
|
|
self.assertEqual(1, self.calls)
|
|
|
|
async def test_timeout_and_transport_failures_are_typed(self) -> None:
|
|
async def delayed(request: httpx.Request) -> httpx.Response:
|
|
await asyncio.sleep(1)
|
|
return httpx.Response(200, json=_response())
|
|
|
|
client = await self._client(delayed)
|
|
with self.assertRaisesRegex(JevError, "timeout"):
|
|
await client.appraise({})
|
|
|
|
async def unavailable(request: httpx.Request) -> httpx.Response:
|
|
raise httpx.ConnectError("network unavailable", request=request)
|
|
|
|
client = await self._client(unavailable)
|
|
with self.assertRaisesRegex(JevError, "transport"):
|
|
await client.appraise({})
|
|
|
|
async def test_cancellation_propagates(self) -> None:
|
|
async def cancelled(request: httpx.Request) -> httpx.Response:
|
|
raise asyncio.CancelledError()
|
|
|
|
client = await self._client(cancelled)
|
|
with self.assertRaises(asyncio.CancelledError):
|
|
await client.appraise({})
|
|
|
|
async def test_rejects_malformed_score_responses(self) -> None:
|
|
invalid_payloads: list[dict[str, object]] = []
|
|
|
|
missing_dimension = _response()
|
|
del missing_dimension["answers"]["trust"]
|
|
invalid_payloads.append(missing_dimension)
|
|
|
|
non_finite = _response()
|
|
non_finite["answers"]["anxiety"]["score"] = float("nan")
|
|
invalid_payloads.append(non_finite)
|
|
|
|
out_of_range = _response()
|
|
out_of_range["answers"]["anxiety"]["score"] = 4.1
|
|
invalid_payloads.append(out_of_range)
|
|
|
|
invalid_probabilities = _response()
|
|
invalid_probabilities["answers"]["anxiety"]["probabilities"]["2"] = 0.7
|
|
invalid_payloads.append(invalid_probabilities)
|
|
|
|
invalid_high_probabilities = _response()
|
|
invalid_high_probabilities["answers"]["anxiety"]["probabilities"]["2"] = 0.9
|
|
invalid_payloads.append(invalid_high_probabilities)
|
|
|
|
missing_legend = _response()
|
|
del missing_legend["answers"]["anxiety"]["legend"]["4"]
|
|
invalid_payloads.append(missing_legend)
|
|
|
|
bad_usage = _response()
|
|
bad_usage["usage"]["input_tokens"] = -1
|
|
invalid_payloads.append(bad_usage)
|
|
|
|
bad_cost = _response()
|
|
bad_cost["usage"]["cost"] = -0.01
|
|
invalid_payloads.append(bad_cost)
|
|
|
|
for payload in invalid_payloads:
|
|
with self.subTest(payload=payload):
|
|
async def handler(request: httpx.Request, payload: dict[str, object] = payload) -> httpx.Response:
|
|
return httpx.Response(
|
|
200,
|
|
content=json.dumps(copy.deepcopy(payload), allow_nan=True),
|
|
headers={"Content-Type": "application/json"},
|
|
)
|
|
|
|
client = await self._client(handler)
|
|
with self.assertRaisesRegex(JevError, "malformed_response"):
|
|
await client.appraise({})
|
|
|
|
async def test_accepts_two_decimal_probability_sum_rounding(self) -> None:
|
|
for probability, expected_sum in ((0.79, 0.99), (0.81, 1.01)):
|
|
with self.subTest(expected_sum=expected_sum):
|
|
payload = _response()
|
|
payload["answers"]["anxiety"]["probabilities"]["2"] = probability
|
|
|
|
async def handler(request: httpx.Request) -> httpx.Response:
|
|
return httpx.Response(200, json=payload)
|
|
|
|
client = await self._client(handler)
|
|
result = await client.appraise({})
|
|
self.assertEqual(0.5, result.emotions["anxiety"].score)
|
|
self.assertEqual(
|
|
(0.0, 0.1, probability, 0.1, 0.0),
|
|
result.emotions["anxiety"].probabilities,
|
|
)
|
|
|
|
async def test_rejects_a_response_from_a_different_model(self) -> None:
|
|
payload = _response()
|
|
payload["model"] = "jev-unknown"
|
|
|
|
async def handler(request: httpx.Request) -> httpx.Response:
|
|
return httpx.Response(200, json=payload)
|
|
|
|
client = await self._client(handler)
|
|
with self.assertRaisesRegex(JevError, "model_mismatch"):
|
|
await client.appraise({})
|
|
|
|
async def test_explicit_typesafe_alias_records_the_resolved_version(self) -> None:
|
|
payload = _response("jev-1.13.0")
|
|
|
|
async def handler(request: httpx.Request) -> httpx.Response:
|
|
self.assertEqual("jev-latest", json.loads(request.content)["model"])
|
|
return httpx.Response(200, json=payload)
|
|
|
|
client = JevClient(
|
|
provider="typesafe",
|
|
api_key="test-key",
|
|
model="jev-latest",
|
|
timeout_seconds=0.05,
|
|
transport=httpx.MockTransport(handler),
|
|
)
|
|
await client.startup()
|
|
self.addAsyncCleanup(client.shutdown)
|
|
result = await client.appraise({})
|
|
|
|
self.assertEqual("jev-1.13.0", result.model)
|
|
|
|
async def test_exact_openrouter_model_slug_is_preserved(self) -> None:
|
|
async def handler(request: httpx.Request) -> httpx.Response:
|
|
self.assertEqual("typesafe/jev-1.13", json.loads(request.content)["model"])
|
|
return httpx.Response(200, json=_response("typesafe/jev-1.13"))
|
|
|
|
client = JevClient(
|
|
provider="openrouter",
|
|
api_key="openrouter-key",
|
|
model="typesafe/jev-1.13",
|
|
timeout_seconds=0.05,
|
|
transport=httpx.MockTransport(handler),
|
|
)
|
|
await client.startup()
|
|
self.addAsyncCleanup(client.shutdown)
|
|
result = await client.appraise({})
|
|
|
|
self.assertEqual("typesafe/jev-1.13", result.model)
|
|
|
|
async def test_typesafe_uses_only_its_explicit_route_and_key(self) -> None:
|
|
payload = _response("jev-1.13.0")
|
|
del payload["usage"]["cost"]
|
|
|
|
async def handler(request: httpx.Request) -> httpx.Response:
|
|
self.assertEqual(TYPESAFE_JEV_ENDPOINT, str(request.url))
|
|
self.assertEqual("Bearer typesafe-key", request.headers["Authorization"])
|
|
return httpx.Response(200, json=payload)
|
|
|
|
client = JevClient(
|
|
provider="typesafe",
|
|
api_key="typesafe-key",
|
|
model="jev-1.13.0",
|
|
timeout_seconds=0.05,
|
|
transport=httpx.MockTransport(handler),
|
|
)
|
|
await client.startup()
|
|
self.addAsyncCleanup(client.shutdown)
|
|
result = await client.appraise({})
|
|
|
|
self.assertEqual("typesafe", result.provider)
|
|
self.assertIsNone(result.cost_usd)
|
|
|
|
async def test_provider_uses_only_its_configured_key(self) -> None:
|
|
configured = SimpleNamespace(
|
|
openrouter_api_key=SecretStr("openrouter-key"),
|
|
typesafe_api_key=SecretStr("typesafe-key"),
|
|
jev_model="~typesafe/jev-latest",
|
|
jev_timeout_seconds=0.05,
|
|
)
|
|
seen_headers: list[str] = []
|
|
|
|
async def handler(request: httpx.Request) -> httpx.Response:
|
|
seen_headers.append(request.headers["Authorization"])
|
|
response_model = (
|
|
"typesafe/jev-1.13-20260917"
|
|
if str(request.url) == OPENROUTER_JEV_ENDPOINT
|
|
else "jev-1.13.0"
|
|
)
|
|
return httpx.Response(200, json=_response(response_model))
|
|
|
|
with patch.object(jev_module, "settings", configured):
|
|
openrouter = JevClient(
|
|
provider="openrouter",
|
|
transport=httpx.MockTransport(handler),
|
|
)
|
|
typesafe = JevClient(
|
|
provider="typesafe",
|
|
model="jev-1.13.0",
|
|
transport=httpx.MockTransport(handler),
|
|
)
|
|
|
|
await openrouter.startup()
|
|
await typesafe.startup()
|
|
self.addAsyncCleanup(openrouter.shutdown)
|
|
self.addAsyncCleanup(typesafe.shutdown)
|
|
await openrouter.appraise({})
|
|
await typesafe.appraise({})
|
|
|
|
self.assertEqual(["Bearer openrouter-key", "Bearer typesafe-key"], seen_headers)
|
|
|
|
async def test_openrouter_allows_optional_score_metadata(self) -> None:
|
|
payload = _response()
|
|
for answer in payload["answers"].values():
|
|
del answer["confidence"]
|
|
del answer["legend"]
|
|
del answer["probabilities"]
|
|
|
|
async def handler(request: httpx.Request) -> httpx.Response:
|
|
return httpx.Response(200, json=payload)
|
|
|
|
client = await self._client(handler)
|
|
result = await client.appraise({})
|
|
|
|
self.assertIsNone(result.emotions["anxiety"].confidence)
|
|
self.assertIsNone(result.emotions["anxiety"].probabilities)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|