29c406d89f
상담자 발화 판정(A)·감정(B)·표현(C) 20문항 질문 세트, 감쇠 없는 이번 턴 반응과 비대칭 기분 전이, 개방도 게이트로 생성 지시를 만들고 ccd.coping_strategy 전달 누락을 고친다. trace v2와 고정 문구 속마음 요약(migration 24, AI 경로 차단 RLS)을 같은 트랜잭션에 저장하고 피드백 정책이 켜진 경우에만 done·TurnResponse·음성 reply·리뷰로 노출한다. 회기 화면 속마음 보기 토글, 리뷰 접힘 블록, 관리자 감정 관측 v2 표시를 추가한다.
585 lines
24 KiB
Python
585 lines
24 KiB
Python
"""Jev HTTP 어댑터(질문 세트 v2)의 단위 계약."""
|
|
|
|
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 (
|
|
CHOICE_QUESTION_IDS,
|
|
EMOTION_DIMENSIONS,
|
|
MAX_SORE_SPOTS,
|
|
NOUL_QUESTION_IDS,
|
|
OPENROUTER_JEV_ENDPOINT,
|
|
SORE_SPOT_QUESTION_ID,
|
|
TYPESAFE_JEV_ENDPOINT,
|
|
JevClient,
|
|
JevError,
|
|
)
|
|
|
|
|
|
def _state(
|
|
*,
|
|
first_turn: bool = False,
|
|
sore_spots: tuple[str, ...] = ("성급한 조언", "능력 평가"),
|
|
forbidden: tuple[str, ...] = (),
|
|
) -> dict[str, object]:
|
|
recent_turns: list[dict[str, str]] = []
|
|
if not first_turn:
|
|
recent_turns.append({"speaker": "client", "text": "저도 몰라서 온 건 아니에요."})
|
|
return {
|
|
"counselor_utterance": "일단 긍정적으로 생각하고 운동부터 해보면 어떨까요?",
|
|
"recent_turns": recent_turns,
|
|
"client_profile": {
|
|
"presenting": "해결책보다 이해받길 바란다.",
|
|
"history": "노력 부족이라는 말을 반복해서 들었다.",
|
|
"core_belief": "실수하면 가치가 없다.",
|
|
"coping_strategy": "설명하거나 날카롭게 항의한다.",
|
|
"big5": {"O": 0.48, "C": 0.82, "E": 0.43, "A": 0.46, "N": 0.69},
|
|
"sore_spots": list(sore_spots),
|
|
"forbidden": list(forbidden),
|
|
"speech_style": {"register": "존댓말"},
|
|
},
|
|
"pinned_facts": ["유능하지 못하다는 평가에 민감하다."],
|
|
"recall_summary": "문제를 설명할 때마다 노력 부족이라는 말을 들었다.",
|
|
"relationship": {"stage": "탐색", "openness": "guarded", "resistance": "high"},
|
|
"previous_feelings": {dimension: "moderate" for dimension in EMOTION_DIMENSIONS},
|
|
}
|
|
|
|
|
|
def _score_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 _noul_answer(probability: float = 0.7) -> dict[str, object]:
|
|
# docs.typesafe.ai/primitives/noul(2026-09-29 확인): 응답은 {"type":"noul","noul":p}이고
|
|
# confidence 필드는 없다("There is no separate confidence field for Noul answers").
|
|
return {"type": "noul", "noul": probability}
|
|
|
|
|
|
def _choice_answer(criteria: dict[str, object]) -> dict[str, object]:
|
|
codes = list(criteria)
|
|
probabilities = {code: (0.7 if index == 0 else 0.3 / (len(codes) - 1)) for index, code in enumerate(codes)}
|
|
return {"type": "choice", "choice": codes[0], "probabilities": probabilities, "confidence": 0.6}
|
|
|
|
|
|
def _answers_for(questions: dict[str, dict[str, object]]) -> dict[str, object]:
|
|
answers: dict[str, object] = {}
|
|
for question_id, question in questions.items():
|
|
if question["type"] == "score":
|
|
answers[question_id] = _score_answer()
|
|
elif question["type"] == "noul":
|
|
answers[question_id] = _noul_answer()
|
|
else:
|
|
answers[question_id] = _choice_answer(question["criteria"]) # type: ignore[arg-type]
|
|
return answers
|
|
|
|
|
|
def _response(
|
|
state: dict[str, object],
|
|
*,
|
|
model: str = "typesafe/jev-1.13-20260917",
|
|
questions: dict[str, dict[str, object]] | None = None,
|
|
) -> tuple[dict[str, object], dict[str, dict[str, object]]]:
|
|
built_questions = questions if questions is not None else JevClient(
|
|
provider="openrouter", api_key="k", model="m"
|
|
)._questions(state)
|
|
payload = {
|
|
"model": model,
|
|
"answers": _answers_for(built_questions),
|
|
"usage": {"input_tokens": 120, "output_tokens": 45, "cost": 0.000019992},
|
|
}
|
|
return payload, built_questions
|
|
|
|
|
|
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_full_question_set_and_normalizes_scores(self) -> None:
|
|
state = _state()
|
|
payload, questions = _response(state)
|
|
|
|
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(questions), set(body["questions"]))
|
|
for dimension in EMOTION_DIMENSIONS:
|
|
question = body["questions"][dimension]
|
|
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"])
|
|
for question_id in NOUL_QUESTION_IDS:
|
|
if question_id not in body["questions"]:
|
|
continue
|
|
question = body["questions"][question_id]
|
|
self.assertEqual("noul", question["type"])
|
|
self.assertEqual({"true", "false"}, set(question["criteria"]))
|
|
for question_id in CHOICE_QUESTION_IDS:
|
|
question = body["questions"][question_id]
|
|
self.assertEqual("choice", question["type"])
|
|
self.assertGreaterEqual(len(question["criteria"]), 2)
|
|
sore_spot_question = body["questions"][SORE_SPOT_QUESTION_ID]
|
|
self.assertEqual("choice", sore_spot_question["type"])
|
|
self.assertEqual({"none", "spot_1", "spot_2"}, set(sore_spot_question["criteria"]))
|
|
return httpx.Response(200, json=payload)
|
|
|
|
client = await self._client(handler)
|
|
result = await client.appraise(state)
|
|
|
|
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.assertEqual(2, result.sore_spot_count)
|
|
self.assertEqual(set(NOUL_QUESTION_IDS), set(result.noul_judgments))
|
|
self.assertEqual(
|
|
set(CHOICE_QUESTION_IDS) | {SORE_SPOT_QUESTION_ID},
|
|
set(result.choice_judgments),
|
|
)
|
|
self.assertEqual(0.7, result.noul_judgments["a_judged"].probability)
|
|
# 공식 noul 응답에는 confidence 필드가 없다 — 응답에 없으면 None으로 보존한다.
|
|
self.assertIsNone(result.noul_judgments["a_judged"].confidence)
|
|
self.assertGreaterEqual(result.latency_ms, 0)
|
|
|
|
async def test_noul_confidence_is_preserved_when_the_response_includes_it(self) -> None:
|
|
state = _state()
|
|
payload, questions = _response(state)
|
|
payload["answers"]["a_judged"] = {"type": "noul", "noul": 0.9, "confidence": 0.55}
|
|
|
|
async def handler(request: httpx.Request) -> httpx.Response:
|
|
return httpx.Response(200, json=payload)
|
|
|
|
client = await self._client(handler)
|
|
result = await client.appraise(state)
|
|
|
|
self.assertEqual(0.9, result.noul_judgments["a_judged"].probability)
|
|
self.assertEqual(0.55, result.noul_judgments["a_judged"].confidence)
|
|
|
|
async def test_first_turn_excludes_a_understood(self) -> None:
|
|
state = _state(first_turn=True)
|
|
payload, questions = _response(state)
|
|
|
|
async def handler(request: httpx.Request) -> httpx.Response:
|
|
body = json.loads(request.content)
|
|
self.assertNotIn("a_understood", body["questions"])
|
|
return httpx.Response(200, json=payload)
|
|
|
|
client = await self._client(handler)
|
|
result = await client.appraise(state)
|
|
|
|
self.assertNotIn("a_understood", questions)
|
|
self.assertNotIn("a_understood", result.noul_judgments)
|
|
|
|
async def test_empty_sore_spots_excludes_a_sore_spot(self) -> None:
|
|
state = _state(sore_spots=(), forbidden=())
|
|
payload, questions = _response(state)
|
|
|
|
async def handler(request: httpx.Request) -> httpx.Response:
|
|
body = json.loads(request.content)
|
|
self.assertNotIn(SORE_SPOT_QUESTION_ID, body["questions"])
|
|
return httpx.Response(200, json=payload)
|
|
|
|
client = await self._client(handler)
|
|
result = await client.appraise(state)
|
|
|
|
self.assertNotIn(SORE_SPOT_QUESTION_ID, questions)
|
|
self.assertNotIn(SORE_SPOT_QUESTION_ID, result.choice_judgments)
|
|
self.assertEqual(0, result.sore_spot_count)
|
|
|
|
async def test_sore_spots_and_forbidden_are_combined_and_capped_at_twelve(self) -> None:
|
|
state = _state(
|
|
sore_spots=tuple(f"민감{i}" for i in range(8)),
|
|
forbidden=tuple(f"금기{i}" for i in range(8)),
|
|
)
|
|
payload, questions = _response(state)
|
|
|
|
async def handler(request: httpx.Request) -> httpx.Response:
|
|
body = json.loads(request.content)
|
|
criteria = body["questions"][SORE_SPOT_QUESTION_ID]["criteria"]
|
|
self.assertEqual(13, len(criteria)) # none + 최대 12개
|
|
self.assertEqual(
|
|
{"none", *(f"spot_{index}" for index in range(1, MAX_SORE_SPOTS + 1))},
|
|
set(criteria),
|
|
)
|
|
return httpx.Response(200, json=payload)
|
|
|
|
client = await self._client(handler)
|
|
result = await client.appraise(state)
|
|
|
|
self.assertEqual(MAX_SORE_SPOTS, result.sore_spot_count)
|
|
self.assertEqual(13, len(questions[SORE_SPOT_QUESTION_ID]["criteria"])) # type: ignore[arg-type]
|
|
|
|
async def test_startup_does_not_issue_a_request(self) -> None:
|
|
async def handler(request: httpx.Request) -> httpx.Response:
|
|
self.calls += 1
|
|
payload, _ = _response(_state())
|
|
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
|
|
payload, _ = _response(_state())
|
|
return httpx.Response(200, json=payload)
|
|
|
|
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(_state())
|
|
self.assertEqual(1, self.calls)
|
|
|
|
async def test_timeout_and_transport_failures_are_typed(self) -> None:
|
|
payload, _ = _response(_state())
|
|
|
|
async def delayed(request: httpx.Request) -> httpx.Response:
|
|
await asyncio.sleep(1)
|
|
return httpx.Response(200, json=payload)
|
|
|
|
client = await self._client(delayed)
|
|
with self.assertRaisesRegex(JevError, "timeout"):
|
|
await client.appraise(_state())
|
|
|
|
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(_state())
|
|
|
|
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(_state())
|
|
|
|
async def test_rejects_malformed_responses(self) -> None:
|
|
state = _state()
|
|
base_payload, questions = _response(state)
|
|
invalid_payloads: list[dict[str, object]] = []
|
|
|
|
missing_question = copy.deepcopy(base_payload)
|
|
del missing_question["answers"]["trust"]
|
|
invalid_payloads.append(missing_question)
|
|
|
|
extra_question = copy.deepcopy(base_payload)
|
|
extra_question["answers"]["unexpected_question"] = _noul_answer()
|
|
invalid_payloads.append(extra_question)
|
|
|
|
non_finite_score = copy.deepcopy(base_payload)
|
|
non_finite_score["answers"]["anxiety"]["score"] = float("nan")
|
|
invalid_payloads.append(non_finite_score)
|
|
|
|
out_of_range_score = copy.deepcopy(base_payload)
|
|
out_of_range_score["answers"]["anxiety"]["score"] = 4.1
|
|
invalid_payloads.append(out_of_range_score)
|
|
|
|
wrong_type = copy.deepcopy(base_payload)
|
|
wrong_type["answers"]["a_judged"]["type"] = "score"
|
|
invalid_payloads.append(wrong_type)
|
|
|
|
noul_out_of_range = copy.deepcopy(base_payload)
|
|
noul_out_of_range["answers"]["a_judged"]["noul"] = 1.5
|
|
invalid_payloads.append(noul_out_of_range)
|
|
|
|
noul_missing_field = copy.deepcopy(base_payload)
|
|
del noul_missing_field["answers"]["a_judged"]["noul"]
|
|
invalid_payloads.append(noul_missing_field)
|
|
|
|
noul_wrong_field_name = copy.deepcopy(base_payload)
|
|
del noul_wrong_field_name["answers"]["a_judged"]["noul"]
|
|
noul_wrong_field_name["answers"]["a_judged"]["probability"] = 0.7
|
|
invalid_payloads.append(noul_wrong_field_name)
|
|
|
|
noul_non_finite = copy.deepcopy(base_payload)
|
|
noul_non_finite["answers"]["a_judged"]["noul"] = float("nan")
|
|
invalid_payloads.append(noul_non_finite)
|
|
|
|
noul_boolean = copy.deepcopy(base_payload)
|
|
noul_boolean["answers"]["a_judged"]["noul"] = True
|
|
invalid_payloads.append(noul_boolean)
|
|
|
|
choice_unknown_code = copy.deepcopy(base_payload)
|
|
choice_unknown_code["answers"]["a_coping"]["choice"] = "not_a_code"
|
|
invalid_payloads.append(choice_unknown_code)
|
|
|
|
choice_missing_option = copy.deepcopy(base_payload)
|
|
del choice_missing_option["answers"]["a_coping"]["probabilities"]["overwhelming"]
|
|
invalid_payloads.append(choice_missing_option)
|
|
|
|
choice_bad_sum = copy.deepcopy(base_payload)
|
|
choice_bad_sum["answers"]["a_coping"]["probabilities"] = {
|
|
"nothing_asked": 0.5,
|
|
"manageable": 0.5,
|
|
"stretch": 0.5,
|
|
"overwhelming": 0.5,
|
|
}
|
|
invalid_payloads.append(choice_bad_sum)
|
|
|
|
bad_usage = copy.deepcopy(base_payload)
|
|
bad_usage["usage"]["input_tokens"] = -1
|
|
invalid_payloads.append(bad_usage)
|
|
|
|
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(payload, allow_nan=True),
|
|
headers={"Content-Type": "application/json"},
|
|
)
|
|
|
|
client = await self._client(handler)
|
|
with self.assertRaisesRegex(JevError, "malformed_response"):
|
|
await client.appraise(state)
|
|
|
|
async def test_choice_probability_tolerance_scales_with_option_count(self) -> None:
|
|
state = _state()
|
|
payload, questions = _response(state)
|
|
option_count = len(questions["c_display"]["criteria"]) # type: ignore[arg-type]
|
|
codes = list(questions["c_display"]["criteria"]) # type: ignore[arg-type]
|
|
# 허용오차 경계 바로 안쪽: N * 0.005 만큼 반올림된 합.
|
|
drift = option_count * 0.005 - 0.0005
|
|
probabilities = {code: 1.0 / option_count for code in codes}
|
|
probabilities[codes[0]] += drift
|
|
payload["answers"]["c_display"] = {
|
|
"type": "choice",
|
|
"choice": codes[0],
|
|
"probabilities": probabilities,
|
|
"confidence": 0.6,
|
|
}
|
|
|
|
async def handler(request: httpx.Request) -> httpx.Response:
|
|
return httpx.Response(200, json=payload)
|
|
|
|
client = await self._client(handler)
|
|
result = await client.appraise(state)
|
|
|
|
self.assertEqual(codes[0], result.choice_judgments["c_display"].choice)
|
|
|
|
async def test_choice_probability_order_follows_criteria_order(self) -> None:
|
|
state = _state()
|
|
payload, questions = _response(state)
|
|
codes = list(questions["c_display"]["criteria"]) # type: ignore[arg-type]
|
|
reordered = {code: (0.7 if code == codes[-1] else 0.1) for code in codes}
|
|
payload["answers"]["c_display"] = {
|
|
"type": "choice",
|
|
"choice": codes[-1],
|
|
"probabilities": reordered,
|
|
"confidence": 0.6,
|
|
}
|
|
|
|
async def handler(request: httpx.Request) -> httpx.Response:
|
|
return httpx.Response(200, json=payload)
|
|
|
|
client = await self._client(handler)
|
|
result = await client.appraise(state)
|
|
|
|
self.assertEqual(tuple(codes), tuple(result.choice_judgments["c_display"].probabilities))
|
|
|
|
async def test_rejects_a_response_from_a_different_model(self) -> None:
|
|
state = _state()
|
|
payload, _ = _response(state, 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(state)
|
|
|
|
async def test_explicit_typesafe_alias_records_the_resolved_version(self) -> None:
|
|
state = _state()
|
|
payload, _ = _response(state, model="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(state)
|
|
|
|
self.assertEqual("jev-1.13.0", result.model)
|
|
|
|
async def test_exact_openrouter_model_slug_is_preserved(self) -> None:
|
|
state = _state()
|
|
payload, _ = _response(state, model="typesafe/jev-1.13")
|
|
|
|
async def handler(request: httpx.Request) -> httpx.Response:
|
|
self.assertEqual("typesafe/jev-1.13", json.loads(request.content)["model"])
|
|
return httpx.Response(200, json=payload)
|
|
|
|
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(state)
|
|
|
|
self.assertEqual("typesafe/jev-1.13", result.model)
|
|
|
|
async def test_typesafe_uses_only_its_explicit_route_and_key(self) -> None:
|
|
state = _state()
|
|
payload, _ = _response(state, model="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(state)
|
|
|
|
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,
|
|
)
|
|
state = _state()
|
|
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"
|
|
)
|
|
payload, _ = _response(state, model=response_model)
|
|
return httpx.Response(200, json=payload)
|
|
|
|
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(state)
|
|
await typesafe.appraise(state)
|
|
|
|
self.assertEqual(["Bearer openrouter-key", "Bearer typesafe-key"], seen_headers)
|
|
|
|
async def test_openrouter_allows_optional_score_and_noul_metadata(self) -> None:
|
|
state = _state()
|
|
payload, _ = _response(state)
|
|
for answer in payload["answers"].values():
|
|
answer.pop("confidence", None)
|
|
answer.pop("legend", None)
|
|
|
|
async def handler(request: httpx.Request) -> httpx.Response:
|
|
return httpx.Response(200, json=payload)
|
|
|
|
client = await self._client(handler)
|
|
result = await client.appraise(state)
|
|
|
|
self.assertIsNone(result.emotions["anxiety"].confidence)
|
|
self.assertIsNone(result.noul_judgments["a_judged"].confidence)
|
|
self.assertIsNone(result.choice_judgments["a_coping"].confidence)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|