import json import unittest from unittest.mock import AsyncMock, patch from app.contracts.engine_gateway import EngineMessage, GenerateRequest from engine_gateway import provider_registry class _FakeStreamReader: def __init__(self, lines: list[bytes] | None = None, body: bytes = b""): self.lines = list(lines or []) self.body = body async def readline(self) -> bytes: return self.lines.pop(0) if self.lines else b"" async def read(self) -> bytes: return self.body class _FakeAgyProcess: def __init__(self, events: list[dict]): self.stdout = _FakeStreamReader( [(json.dumps(event, ensure_ascii=False) + "\n").encode("utf-8") for event in events] ) self.stderr = _FakeStreamReader() self.returncode = None async def wait(self) -> int: if self.returncode is None: self.returncode = 0 return self.returncode def kill(self) -> None: self.returncode = -9 class ProviderRegistryTest(unittest.IsolatedAsyncioTestCase): def setUp(self): provider_registry.clear_capability_cache() def tearDown(self): provider_registry.clear_capability_cache() async def test_codex_catalog_uses_live_models_with_terra_medium_default(self): payload = { "data": [ { "id": "gpt-5.6-sol", "model": "gpt-5.6-sol", "displayName": "GPT-5.6-Sol", "description": "Frontier", "hidden": False, "isDefault": True, "defaultReasoningEffort": "low", "supportedReasoningEfforts": [ {"reasoningEffort": "low"}, {"reasoningEffort": "medium"}, ], }, { "id": "gpt-5.6-terra", "model": "gpt-5.6-terra", "displayName": "GPT-5.6-Terra", "description": "Balanced", "hidden": False, "isDefault": False, "defaultReasoningEffort": "medium", "supportedReasoningEfforts": [ {"reasoningEffort": "low"}, {"reasoningEffort": "medium"}, {"reasoningEffort": "high"}, ], }, ] } with ( patch.object(provider_registry, "_binary", return_value="codex.exe"), patch.object( provider_registry, "_codex_model_list", AsyncMock(return_value=payload), ), ): result = await provider_registry.discover_capabilities("codex_cli") self.assertTrue(result.available) self.assertEqual(result.source, "live_cli") self.assertEqual(result.default_model, "gpt-5.6-terra") self.assertEqual(result.default_reasoning_effort, "medium") terra = next(model for model in result.models if model.id == "gpt-5.6-terra") self.assertTrue(terra.is_default) self.assertEqual(terra.reasoning_efforts, ["low", "medium", "high"]) async def test_agy_catalog_uses_cli_list_with_flash_high_default(self): stdout = "\n".join( [ "gemini-3.6-flash-high", "gemini-3.6-flash-medium", "claude-sonnet-4-6", ] ) with ( patch.object(provider_registry, "_binary", return_value="agy.exe"), patch.object( provider_registry, "_run_process", AsyncMock(return_value=(stdout, "")), ), ): result = await provider_registry.discover_capabilities("agy_cli") self.assertTrue(result.available) self.assertEqual(result.default_model, "gemini-3.6-flash-high") self.assertEqual(result.default_reasoning_effort, "high") selected = next(model for model in result.models if model.is_default) self.assertEqual(selected.reasoning_efforts, ["high"]) self.assertEqual(selected.label, "Gemini 3.6 Flash (High)") async def test_claude_cli_catalog_is_explicit_static_alias_fallback(self): with patch.object(provider_registry, "_binary", return_value="claude.exe"): result = await provider_registry.discover_capabilities("claude_cli") self.assertTrue(result.available) self.assertEqual(result.source, "static_cli") self.assertEqual(result.default_model, "gateway-default") self.assertEqual([model.id for model in result.models], ["gateway-default", "opus", "sonnet", "fable"]) async def test_anthropic_catalog_fails_closed_without_api_key(self): with patch.dict(provider_registry.os.environ, {}, clear=True): result = await provider_registry.discover_capabilities( "claude_api", force=True ) self.assertFalse(result.available) self.assertEqual(result.source, "unavailable") self.assertEqual(result.models, []) self.assertIn("ANTHROPIC_API_KEY", result.detail) async def test_codex_generation_uses_model_and_reasoning_from_selection(self): capabilities = provider_registry.EngineCapabilitiesResponse( provider="codex_cli", available=True, source="live_cli", models=[ provider_registry.EngineModelOption( id="gpt-5.6-terra", label="GPT-5.6-Terra", reasoning_efforts=["low", "medium", "high"], default_reasoning_effort="medium", is_default=True, ) ], default_model="gpt-5.6-terra", default_reasoning_effort="medium", fetched_at=1, ) stdout = "\n".join( [ json.dumps( { "type": "item.completed", "item": {"type": "agent_message", "text": "OK"}, } ), json.dumps( { "type": "turn.completed", "usage": { "input_tokens": 12, "cached_input_tokens": 2, "output_tokens": 3, }, } ), ] ) runner = AsyncMock(return_value=(stdout, "")) request = GenerateRequest( provider="codex_cli", model="gpt-5.6-terra", reasoning_effort="medium", messages=[EngineMessage(role="user", content="hello")], ) with ( patch.object(provider_registry, "_binary", return_value="codex.exe"), patch.object( provider_registry, "discover_capabilities", AsyncMock(return_value=capabilities), ), patch.object(provider_registry, "_run_process", runner), ): result = await provider_registry.generate_with_provider( request, system_prompt="system", user_payload="hello", ) self.assertEqual(result.text, "OK") self.assertEqual(result.tokens_in, 12) self.assertEqual(result.tokens_out, 3) self.assertEqual(result.cost_usd, 0.0000705) args = runner.await_args.args[0] self.assertIn("gpt-5.6-terra", args) self.assertIn('model_reasoning_effort="medium"', args) self.assertEqual(args[-1], "-") self.assertIn("[시스템 지침]", runner.await_args.kwargs["input_text"]) async def test_agy_generation_uses_stream_json_usage_and_reference_cost(self): capabilities = provider_registry.EngineCapabilitiesResponse( provider="agy_cli", available=True, source="live_cli", models=[ provider_registry.EngineModelOption( id="gemini-3.6-flash-high", label="Gemini 3.6 Flash (High)", reasoning_efforts=["high"], default_reasoning_effort="high", ) ], default_model="gemini-3.6-flash-high", default_reasoning_effort="high", fetched_at=1, ) request = GenerateRequest( provider="agy_cli", model="gemini-3.6-flash-high", reasoning_effort="high", messages=[EngineMessage(role="user", content="hello")], ) process = _FakeAgyProcess( [ { "event": "step_update", "step_update": { "step_type": "agent_response", "state": "DONE", "text_delta": "OK", }, }, { "event": "result", "result": { "status": "SUCCESS", "response": "OK", "usage": { "input_tokens": 12, "cache_read_tokens": 2, "output_tokens": 2, }, }, }, ] ) captured: list[tuple] = [] async def fake_create_subprocess_exec(*args, **kwargs): captured.append(args) return process with ( patch.object(provider_registry, "_binary", return_value="agy"), patch.object( provider_registry, "discover_capabilities", AsyncMock(return_value=capabilities), ), patch.object( provider_registry.asyncio, "create_subprocess_exec", fake_create_subprocess_exec, ), ): result = await provider_registry.generate_with_provider( request, system_prompt="system", user_payload="hello", ) self.assertEqual(result.text, "OK") self.assertEqual(result.tokens_in, 12) self.assertEqual(result.tokens_out, 2) self.assertEqual(result.cost_usd, 0.0000303) args = captured[0] print_index = args.index("--print") self.assertEqual(print_index, len(args) - 2) self.assertIn("[시스템 지침]", args[-1]) self.assertEqual(args[args.index("--output-format") + 1], "stream-json") async def test_agy_stream_forwards_live_deltas_without_repeating_final_response(self): capabilities = provider_registry.EngineCapabilitiesResponse( provider="agy_cli", available=True, source="live_cli", models=[ provider_registry.EngineModelOption( id="gemini-3.6-flash-high", label="Gemini 3.6 Flash (High)", reasoning_efforts=["high"], default_reasoning_effort="high", is_default=True, ) ], default_model="gemini-3.6-flash-high", default_reasoning_effort="high", fetched_at=1, ) request = GenerateRequest( provider="agy_cli", model="gemini-3.6-flash-high", reasoning_effort="high", messages=[EngineMessage(role="user", content="hello")], ) process = _FakeAgyProcess( [ { "event": "step_update", "step_update": { "step_type": "agent_response", "state": "ACTIVE", "text_delta": "안", }, }, { "event": "step_update", "step_update": { "step_type": "agent_response", "state": "DONE", "text_delta": "녕", }, }, { "event": "result", "result": { "status": "SUCCESS", "response": "안녕", "usage": { "input_tokens": 12, "cache_read_tokens": 2, "output_tokens": 2, }, }, }, ] ) captured: list[tuple] = [] async def fake_create_subprocess_exec(*args, **kwargs): captured.append(args) return process with ( patch.object(provider_registry, "_binary", return_value="agy.exe"), patch.object( provider_registry, "discover_capabilities", AsyncMock(return_value=capabilities), ), patch.object( provider_registry.asyncio, "create_subprocess_exec", fake_create_subprocess_exec, ), ): events = [ event async for event in provider_registry.stream_with_provider( request, system_prompt="system", user_payload="hello", ) ] self.assertEqual([event.type for event in events], ["delta", "delta", "done"]) self.assertEqual("".join(event.text for event in events), "안녕") self.assertEqual(events[-1].result.text, "안녕") self.assertEqual(events[-1].result.tokens_in, 12) self.assertEqual(events[-1].result.tokens_out, 2) self.assertEqual(events[-1].result.cost_usd, 0.0000303) args = captured[0] self.assertIn("--output-format", args) self.assertEqual(args[args.index("--output-format") + 1], "stream-json") self.assertEqual(args.index("--print"), len(args) - 2) async def test_generation_rejects_model_effort_not_returned_by_provider(self): capabilities = provider_registry.EngineCapabilitiesResponse( provider="agy_cli", available=True, source="live_cli", models=[ provider_registry.EngineModelOption( id="gemini-3.6-flash-high", label="Gemini 3.6 Flash (High)", reasoning_efforts=["high"], default_reasoning_effort="high", ) ], default_model="gemini-3.6-flash-high", default_reasoning_effort="high", fetched_at=1, ) request = GenerateRequest( provider="agy_cli", model="gemini-3.6-flash-high", reasoning_effort="low", messages=[EngineMessage(role="user", content="hello")], ) with patch.object( provider_registry, "discover_capabilities", AsyncMock(return_value=capabilities), ): with self.assertRaisesRegex( provider_registry.ProviderError, "사용할 수 없는 추론 강도" ): await provider_registry._resolve_selection(request, "agy_cli") if __name__ == "__main__": unittest.main()