"""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()