from __future__ import annotations import importlib.util import json import os import sys from pathlib import Path import tempfile import unittest from unittest import mock def load_module(): module_path = Path(__file__).resolve().parents[1] / "codex_retry_gateway_tui.py" spec = importlib.util.spec_from_file_location("codex_retry_gateway_tui", module_path) assert spec and spec.loader module = importlib.util.module_from_spec(spec) sys.modules[spec.name] = module spec.loader.exec_module(module) return module class CodexRetryGatewayTUITests(unittest.TestCase): def test_summary_line_uses_api_snapshot(self) -> None: mod = load_module() payload = { "listen": "127.0.0.1:4610", "config": {"profile_name": "pc", "upstream_base_url": "https://example.com/v1"}, "metrics": {"total_proxy_request_count": 11, "inspected_response_count": 7, "matched_response_count": 2, "reasoning_516_count": 1}, } self.assertIn("profile pc", mod.summary_line(payload)) self.assertIn("req 11", mod.summary_line(payload)) def test_request_rows_sort_desc_and_keep_request_id(self) -> None: mod = load_module() payload = { "entries": [ {"seq": 1, "request_id": "req_a", "response_id": "resp_a", "thread_id": "thread_a", "path": "/responses"}, {"seq": 2, "request_id": "req_b", "response_id": "resp_b", "thread_id": "thread_b", "path": "/v1/responses"}, ] } rows = mod.normalize_request_rows(payload) self.assertEqual([row["request_id"] for row in rows], ["req_b", "req_a"]) self.assertEqual(rows[0]["response_id"], "resp_b") self.assertEqual(rows[0]["thread_id"], "thread_b") def test_profile_rows_show_active_first(self) -> None: mod = load_module() payload = { "profiles": [ {"name": "beta", "active": False, "summary": {"listen_host": "127.0.0.1", "listen_port": 4611, "upstream_base_url": "u1", "auth_mode": "passthrough", "auth_source": "passthrough", "request_history_limit": 10}}, {"name": "alpha", "active": True, "summary": {"listen_host": "127.0.0.1", "listen_port": 4610, "upstream_base_url": "u0", "auth_mode": "manual_bearer", "auth_source": "manual_file", "request_history_limit": 20}}, ] } rows = mod.normalize_profile_rows(payload) self.assertEqual([row["name"] for row in rows], ["alpha", "beta"]) def test_version_update_message_only_for_newer_versions(self) -> None: mod = load_module() self.assertIn("0.1.0 -> 0.1.1", mod.version_update_message("0.1.1", "0.1.0")) self.assertEqual(mod.version_update_message("0.1.0", "0.1.0"), "") def test_gateway_url_helpers_and_request_age(self) -> None: mod = load_module() self.assertEqual( mod.gateway_admin_url("http://127.0.0.1:4610"), "http://127.0.0.1:4610/__codex_retry_gateway", ) self.assertEqual( mod.normalize_gateway_url("http://127.0.0.1:4610/__codex_retry_gateway/api/status"), "http://127.0.0.1:4610/__codex_retry_gateway", ) self.assertEqual( mod.gateway_status_url("http://127.0.0.1:4610/__codex_retry_gateway"), "http://127.0.0.1:4610/__codex_retry_gateway/api/status", ) row = { "seq": 1, "request_id": "r", "response_id": "resp_1", "thread_id": "thread_1", "method": "POST", "path": "/responses", "status_code": 200, "upstream_status_code": 200, "upstream_attempt_count": 1, "first_response_delay_ms": 10, "duration_ms": 20, "request_body_bytes": 3, "response_bytes_received": 4096, "stream_chunk_count": 7, "response_stream": True, "started_at": "2026-06-30T00:00:00Z", "finished_at": "2026-06-30T00:00:01Z", "usage_last_updated_at": "2026-06-30T00:00:01Z", } self.assertEqual(mod.primary_request_id(row), "resp_1") self.assertIn("thread thread_1", mod.render_request_detail(row)) self.assertIn("updated 1.0s", mod.render_request_detail(row)) self.assertEqual(mod.request_updated_elapsed(row), "1.0s") self.assertEqual(mod.request_chunk_progress(row), "7 / 4.1KB") def test_request_usage_summary_shows_cached_ratio(self) -> None: mod = load_module() row = { "input_tokens": 100, "output_tokens": 40, "cached_tokens": 25, } self.assertEqual(mod.request_usage_summary(row), "in 75 | out 40 | cached 25 (25%)") def test_request_row_cells_expand_with_wider_limits(self) -> None: mod = load_module() row = { "seq": 1, "request_id": "req_abcdefghijklmnopqrstuvwxyz0123456789", "response_id": "resp_abcdefghijklmnopqrstuvwxyz0123456789", "thread_id": "thread_abcdefghijklmnopqrstuvwxyz0123456789", "started_at": "2026-06-30T12:00:00Z", "status_code": 200, "path": "/v1/responses/this/is/a/very/long/path/for/testing/truncation", "model": "gpt-very-long-model-name-for-testing", "requested_model": "", "forwarded_model": "", "reasoning_tokens": 516, "input_tokens": 1200, "output_tokens": 340, "cached_tokens": 128, "response_bytes_received": 363800, "response_stream": True, "stream_chunk_count": 587, "first_response_delay_ms": 1500, "duration_ms": 1542200, "upstream_attempt_count": 2, "error": "Selected model is at capacity. Please try a different model.", "finished_at": "2026-06-30T12:25:00Z", } compact = mod.request_row_cells(row, {"id": 18, "thread": 18, "path": 22, "model": 16, "usage": 24, "note": 22}) wide = mod.request_row_cells(row, {"id": 36, "thread": 36, "path": 40, "model": 24, "usage": 72, "note": 48}) self.assertLess(len(compact[1]), len(wide[1])) self.assertLess(len(compact[2]), len(wide[2])) self.assertLess(len(compact[5]), len(wide[5])) self.assertLess(len(compact[8]), len(wide[8])) self.assertLess(len(compact[14]), len(wide[14])) def test_request_rows_read_usage_from_nested_usage_object(self) -> None: mod = load_module() payload = { "entries": [ { "seq": 1, "request_id": "req_nested", "path": "/responses", "usage": { "input_tokens": 120, "output_tokens": 33, "total_tokens": 153, "cached_tokens": 20, }, } ] } rows = mod.normalize_request_rows(payload) self.assertEqual(rows[0]["input_tokens"], 120) self.assertEqual(rows[0]["output_tokens"], 33) self.assertEqual(rows[0]["total_tokens"], 153) self.assertEqual(rows[0]["cached_tokens"], 20) def test_format_duration_ms_as_seconds(self) -> None: mod = load_module() self.assertEqual(mod.format_duration_ms_as_seconds(1532), "1.5s") self.assertEqual(mod.format_duration_ms_as_seconds(120000), "120s") def test_profile_editor_payload_round_trip(self) -> None: mod = load_module() row = { "name": "pc", "listen_host": "100.115.235.115", "listen_port": "4610", "upstream_base_url": "https://example.com/v1", "auth_mode": "passthrough", "auth_json_key": "OPENAI_API_KEY", "request_history_limit": 0, "model_remap": "a=b", "reasoning_equals": [516, 1034], "raw": { "form": { "retryable_error_messages": [ "Selected model is at capacity. Please try a different model.", "stream disconnected before completion: Concurrency limit exceeded for account, please retry later", ], } }, } text = mod.profile_editor_document(row) payload = mod.profile_payload_from_editor_text(text) self.assertEqual(payload["name"], "pc") self.assertIn( "stream disconnected before completion: Concurrency limit exceeded for account, please retry later", payload["retryable_error_messages"], ) def test_default_api_url_discovers_gateway_state(self) -> None: mod = load_module() with tempfile.TemporaryDirectory() as tmpdir: state_path = Path(tmpdir) / "state.json" config_path = Path(tmpdir) / "config.json" api_url_path = Path(tmpdir) / "api-url" state_path.write_text(json.dumps({"gateway_base_url": "http://100.115.235.115:4610"}), encoding="utf-8") config_path.write_text("{}", encoding="utf-8") mod.DEFAULT_GATEWAY_STATE_FILE = str(state_path) mod.DEFAULT_GATEWAY_JSON_CONFIG_FILE = str(config_path) with mock.patch.dict(os.environ, {"CODEX_RETRY_GATEWAY_TUI_API_URL_FILE": str(api_url_path)}, clear=False): self.assertEqual( mod.default_api_url(), "http://100.115.235.115:4610/__codex_retry_gateway", ) def test_default_api_url_discovers_gateway_config_when_state_missing(self) -> None: mod = load_module() with tempfile.TemporaryDirectory() as tmpdir: state_path = Path(tmpdir) / "missing-state.json" config_path = Path(tmpdir) / "config.json" api_url_path = Path(tmpdir) / "api-url" config_path.write_text( json.dumps({"listen_host": "0.0.0.0", "listen_port": 4610}), encoding="utf-8", ) mod.DEFAULT_GATEWAY_STATE_FILE = str(state_path) mod.DEFAULT_GATEWAY_JSON_CONFIG_FILE = str(config_path) with mock.patch.dict(os.environ, {"CODEX_RETRY_GATEWAY_TUI_API_URL_FILE": str(api_url_path)}, clear=False): self.assertEqual( mod.default_api_url(), "http://127.0.0.1:4610/__codex_retry_gateway", ) if __name__ == "__main__": unittest.main()