import asyncio import math import unittest from fastapi.testclient import TestClient from jev_adapter.__main__ import positive_temperature, temperature_map from jev_adapter.backend import AdapterError, ScoringResult from jev_adapter.protocol import SystemOneRequest, probabilities_from_logprobs from jev_adapter.server import create_app from jev_adapter.service import SystemOneService def payload(): return { "model": "test-model", "state": {"text": "결제가 두 번 되었어요. 환불해 주세요."}, "questions": { "route": { "type": "choice", "instructions": "담당 부서", "criteria": {"billing": "청구", "technical": "기술"}, }, "refund": {"type": "noul", "instructions": "환불 요청인가?"}, "urgency": { "type": "score", "instructions": "긴급도", "criteria": ["낮음", "중간", "높음"], }, }, } class FakeBackend: model = "test-model" def __init__(self, delay=0, failure=False): self.calls = [] self.active = self.peak = self.cancelled = 0 self.started = self.closed = False self.delay, self.failure = delay, failure async def start(self): self.started = True async def close(self): self.closed = True def labels(self, count): return tuple(chr(65 + i) for i in range(count)), tuple(range(65, 65 + count)) async def evaluate(self, prompt, images, labels, token_ids, assistant_prefix): self.calls.append((prompt, images, labels, token_ids, assistant_prefix)) self.active += 1 self.peak = max(self.peak, self.active) try: if self.failure: raise AdapterError( "backend_unavailable", "Engine unavailable.", status=502 ) await asyncio.sleep(self.delay) return ScoringResult( tuple(math.log(0.8 if i == 0 else 0.1) for i in range(len(labels))), 20 ) except asyncio.CancelledError: self.cancelled += 1 raise finally: self.active -= 1 class TestService(unittest.IsolatedAsyncioTestCase): async def test_mixed_questions_keep_images_and_build_typed_response(self): backend = FakeBackend() body = payload() body["images"] = ["data:image/png;base64,aGVsbG8="] result = await SystemOneService(backend).score( SystemOneRequest.model_validate(body) ) self.assertEqual(result["answers"]["route"]["choice"], "billing") self.assertAlmostEqual(result["answers"]["refund"]["noul"], 8 / 9) self.assertAlmostEqual(result["answers"]["urgency"]["score"], 0.3) self.assertEqual(result["usage"], {"input_tokens": 60, "output_tokens": 0}) self.assertEqual(result["metadata"]["evaluations"], 3) for call in backend.calls: self.assertEqual(call[1], body["images"]) self.assertNotIn("담당 부서", backend.calls[1][0]) async def test_rotations_preserve_canonical_option_mapping(self): body = payload() body["questions"] = {"route": body["questions"]["route"]} body["options"] = {"permutations": 2, "return_logprobs": True} result = await SystemOneService(FakeBackend()).score( SystemOneRequest.model_validate(body) ) route = result["answers"]["route"] self.assertAlmostEqual(route["probabilities"]["billing"], 0.5) self.assertAlmostEqual(route["confidence"], 0) self.assertEqual(len(route["logprobs"]), 2) async def test_concurrency_limit_applies_across_requests(self): backend = FakeBackend(delay=0.01) service = SystemOneService(backend, max_concurrency=2) await asyncio.gather( *( service.score(SystemOneRequest.model_validate(payload())) for _ in range(3) ) ) self.assertEqual(len(backend.calls), 9) self.assertEqual(backend.peak, 2) async def test_cancellation_closes_all_pending_work(self): backend = FakeBackend(delay=10) task = asyncio.create_task( SystemOneService(backend, max_concurrency=2).score( SystemOneRequest.model_validate(payload()) ) ) while backend.active != 2: await asyncio.sleep(0) task.cancel() with self.assertRaises(asyncio.CancelledError): await task self.assertEqual(backend.cancelled, 2) self.assertEqual(backend.active, 0) self.assertEqual(len(backend.calls), 2) async def test_wrong_model_rejected_before_inference(self): backend = FakeBackend() body = payload() body["model"] = "wrong-model" with self.assertRaises(AdapterError) as error: await SystemOneService(backend).score(SystemOneRequest.model_validate(body)) self.assertEqual(error.exception.status, 404) self.assertEqual(backend.calls, []) class TestDefaultTemperature(unittest.IsolatedAsyncioTestCase): """A server-side default fills an omitted options.temperature only.""" vector = tuple(math.log(0.8 if i == 0 else 0.1) for i in range(3)) async def score(self, body, **service_options): service = SystemOneService(FakeBackend(), **service_options) return await service.score(SystemOneRequest.model_validate(body)) def route_probabilities(self, result): return list(result["answers"]["urgency"]["probabilities"].values()) async def test_unset_default_keeps_request_default_of_one(self): result = await self.score(payload()) self.assertEqual(result["metadata"]["temperature"], 1.0) expected = probabilities_from_logprobs(self.vector, 1.0) for actual, wanted in zip(self.route_probabilities(result), expected): self.assertAlmostEqual(actual, wanted) async def test_omitted_temperature_takes_server_default(self): for body in (payload(), {**payload(), "options": {"permutations": 1}}): result = await self.score(body, default_temperature=2.4) self.assertEqual(result["metadata"]["temperature"], 2.4) expected = probabilities_from_logprobs(self.vector, 2.4) for actual, wanted in zip(self.route_probabilities(result), expected): self.assertAlmostEqual(actual, wanted) self.assertNotAlmostEqual( self.route_probabilities(result)[0], probabilities_from_logprobs(self.vector, 1.0)[0], ) async def test_explicit_request_temperature_wins_over_default(self): body = {**payload(), "options": {"temperature": 1.0}} result = await self.score(body, default_temperature=2.4) self.assertEqual(result["metadata"]["temperature"], 1.0) expected = probabilities_from_logprobs(self.vector, 1.0) for actual, wanted in zip(self.route_probabilities(result), expected): self.assertAlmostEqual(actual, wanted) body = {**payload(), "options": {"temperature": 3.0}} result = await self.score(body, default_temperature=2.4) self.assertEqual(result["metadata"]["temperature"], 3.0) async def test_disabled_scaling_ignores_default(self): body = {**payload(), "options": {"temperature_scaling": False}} result = await self.score(body, default_temperature=2.4) self.assertEqual(result["metadata"]["temperature"], 1.0) expected = probabilities_from_logprobs(self.vector, 1.0) for actual, wanted in zip(self.route_probabilities(result), expected): self.assertAlmostEqual(actual, wanted) async def test_default_does_not_mutate_the_request(self): request = SystemOneRequest.model_validate(payload()) service = SystemOneService(FakeBackend(), default_temperature=2.4) await service.score(request) self.assertEqual(request.options.temperature, 1.0) self.assertNotIn("temperature", request.options.model_fields_set) self.assertEqual(service.effective_options(request).temperature, 2.4) def test_invalid_default_temperature_rejected_at_construction(self): for value in (0, -1.0, float("inf"), float("nan"), True, "2.4", None): with self.assertRaises((ValueError, TypeError)): SystemOneService(FakeBackend(), default_temperature=value) self.assertEqual( SystemOneService(FakeBackend(), default_temperature=2).default_temperature, 2.0, ) async def test_temperature_by_type_applies_per_answer_type(self): result = await self.score( payload(), default_temperature=2.4, temperature_by_type={"score": 3.0, "noul": 1.5} ) urgency = list(result["answers"]["urgency"]["probabilities"].values()) for actual, wanted in zip(urgency, probabilities_from_logprobs(self.vector, 3.0)): self.assertAlmostEqual(actual, wanted) route = list(result["answers"]["route"]["probabilities"].values()) for actual, wanted in zip(route, probabilities_from_logprobs(self.vector[:2], 2.4)): self.assertAlmostEqual(actual, wanted) self.assertEqual(result["metadata"]["temperature"], 2.4) self.assertEqual(result["metadata"]["temperature_by_type"], {"score": 3.0, "noul": 1.5}) async def test_explicit_request_temperature_wins_over_temperature_by_type(self): body = {**payload(), "options": {"temperature": 1.0}} result = await self.score(body, temperature_by_type={"score": 3.0}) urgency = list(result["answers"]["urgency"]["probabilities"].values()) for actual, wanted in zip(urgency, probabilities_from_logprobs(self.vector, 1.0)): self.assertAlmostEqual(actual, wanted) self.assertNotIn("temperature_by_type", result["metadata"]) async def test_disabled_scaling_ignores_temperature_by_type(self): body = {**payload(), "options": {"temperature_scaling": False}} result = await self.score(body, temperature_by_type={"score": 3.0}) urgency = list(result["answers"]["urgency"]["probabilities"].values()) for actual, wanted in zip(urgency, probabilities_from_logprobs(self.vector, 1.0)): self.assertAlmostEqual(actual, wanted) self.assertNotIn("temperature_by_type", result["metadata"]) def test_invalid_temperature_by_type_rejected_at_construction(self): for value in ({"judge": 1.2}, {"score": 0}, {"score": float("nan")}, {"noul": True}): with self.assertRaises(ValueError): SystemOneService(FakeBackend(), temperature_by_type=value) self.assertEqual(SystemOneService(FakeBackend()).temperature_by_type, {}) def test_cli_temperature_map_type(self): self.assertEqual(temperature_map("choice=1.6, noul=1.2,score=1.8"), {"choice": 1.6, "noul": 1.2, "score": 1.8}) self.assertEqual(temperature_map(""), {}) import argparse for text in ("choice", "judge=1.2", "choice=0", "choice=1.2,choice=1.3", "noul=abc"): with self.assertRaises(argparse.ArgumentTypeError): temperature_map(text) def test_cli_default_temperature_type(self): self.assertEqual(positive_temperature("2.4"), 2.4) self.assertEqual(positive_temperature("1"), 1.0) import argparse for text in ("0", "-1", "inf", "nan", "abc", ""): with self.assertRaises(argparse.ArgumentTypeError): positive_temperature(text) class TestPromptWording(unittest.IsolatedAsyncioTestCase): """--prompt-wording is a server-wide setting (CLI/env, not per-request); the default keeps today's served text unchanged.""" async def test_default_served_wording_is_unchanged(self): backend = FakeBackend() body = payload() body["questions"] = {"route": body["questions"]["route"]} await SystemOneService(backend).score(SystemOneRequest.model_validate(body)) prompt = backend.calls[0][0] self.assertTrue(prompt.startswith("Context:\n")) async def test_native_wording_is_plumbed_into_build_prompt(self): backend = FakeBackend() body = payload() body["questions"] = {"route": body["questions"]["route"]} service = SystemOneService(backend, prompt_wording="native") await service.score(SystemOneRequest.model_validate(body)) prompt = backend.calls[0][0] self.assertTrue(prompt.startswith("Read the state and question.")) self.assertIn("\n\nState:\n", prompt) self.assertNotIn("Context:\n", prompt) async def test_native_wording_requires_canonical_az_labels(self): # FakeBackend.labels() already returns canonical A, B, C, ... so this # documents the happy path; build_prompt itself enforces the # requirement (see TestNativePromptWording in test_protocol.py) when a # backend's labels are not canonical. backend = FakeBackend() labels, _ = backend.labels(3) self.assertEqual(labels, ("A", "B", "C")) def test_invalid_prompt_wording_rejected_at_construction(self): for value in ("", "SERVED", "native ", None, 1): with self.assertRaises(ValueError): SystemOneService(FakeBackend(), prompt_wording=value) class TestHTTP(unittest.TestCase): def test_standalone_app_lifecycle_schema_and_alias(self): backend = FakeBackend() with TestClient(create_app(backend)) as client: self.assertTrue(backend.started) self.assertEqual(client.get("/health").json(), {"status": "ok"}) self.assertEqual( client.get("/v1/models").json()["data"][0]["id"], backend.model ) body = payload() body["model"] = "jev-latest" response = client.post("/v1/systemone", json=body) self.assertEqual(response.status_code, 200) self.assertEqual(response.json()["model"], backend.model) self.assertEqual( set(response.json()["answers"]), {"route", "refund", "urgency"} ) bad = client.post("/v1/systemone", json={"model": backend.model}) self.assertEqual(bad.status_code, 422) self.assertEqual(bad.json()["error"]["code"], "invalid_request") self.assertTrue(backend.closed) def test_app_default_temperature_applies_when_request_omits_it(self): with TestClient(create_app(FakeBackend(), default_temperature=2.4)) as client: response = client.post("/v1/systemone", json=payload()) self.assertEqual(response.status_code, 200) self.assertEqual(response.json()["metadata"]["temperature"], 2.4) body = {**payload(), "options": {"temperature": 1.5}} response = client.post("/v1/systemone", json=body) self.assertEqual(response.json()["metadata"]["temperature"], 1.5) with self.assertRaises(ValueError): create_app(FakeBackend(), default_temperature=0) def test_app_prompt_wording_native_reaches_the_backend_over_http(self): backend = FakeBackend() with TestClient(create_app(backend, prompt_wording="native")) as client: body = payload() body["questions"] = {"route": body["questions"]["route"]} response = client.post("/v1/systemone", json=body) self.assertEqual(response.status_code, 200) self.assertTrue(backend.calls[0][0].startswith("Read the state and question.")) def test_auth_guards_inference_and_model_discovery(self): backend = FakeBackend() with TestClient(create_app(backend, api_key="test-secret")) as client: self.assertEqual(client.get("/v1/models").status_code, 401) self.assertEqual( client.post("/v1/systemone", json=payload()).status_code, 401 ) self.assertEqual( client.post( "/v1/systemone", json=payload(), headers={b"Authorization": b"Bearer caf\xe9"}, ).status_code, 401, ) self.assertEqual(backend.calls, []) response = client.post( "/v1/systemone", json=payload(), headers={"Authorization": "Bearer test-secret"}, ) self.assertEqual(response.status_code, 200) def test_upstream_failure_is_not_returned_as_probabilities(self): with TestClient(create_app(FakeBackend(failure=True))) as client: response = client.post("/v1/systemone", json=payload()) self.assertEqual(response.status_code, 502) self.assertEqual(response.json()["error"]["code"], "backend_unavailable") self.assertNotIn("answers", response.json())