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", "reasoning_effort": "high"}, {"seq": 2, "request_id": "req_b", "response_id": "resp_b", "thread_id": "thread_b", "path": "/v1/responses", "reasoning_effort": "xhigh", "reasoning_summary": "auto"}, ] } 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") self.assertEqual(rows[0]["reasoning_effort"], "xhigh") self.assertEqual(rows[0]["reasoning_summary"], "auto") 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, "reasoning_effort": "xhigh", "reasoning_summary": "auto", "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.request_id_text(row), "r") self.assertEqual(mod.response_id_text(row), "resp_1") self.assertIn("thread thread_1", mod.render_request_detail(row)) self.assertIn("req r", mod.render_request_detail(row)) self.assertIn("resp resp_1", mod.render_request_detail(row)) self.assertIn("updated 1.0s", mod.render_request_detail(row)) self.assertIn("effort xhigh", 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, "request_body_bytes": 0, "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, {"req_id": 18, "resp_id": 18, "thread": 18, "path": 22, "model": 16, "note": 22}) wide = mod.request_row_cells(row, {"req_id": 36, "resp_id": 36, "thread": 36, "path": 40, "model": 24, "note": 48}) self.assertEqual(compact[13], "0B") self.assertEqual(wide[13], "0B") self.assertEqual(compact[8], "-") self.assertEqual(compact[9], "516") self.assertEqual(compact[10], "1.1K") self.assertEqual(compact[11], "340") self.assertEqual(compact[12], "128") self.assertLess(len(compact[1]), len(wide[1])) self.assertLess(len(compact[2]), len(wide[2])) self.assertLess(len(compact[3]), len(wide[3])) self.assertLess(len(compact[6]), len(wide[6])) self.assertLess(len(compact[19]), len(wide[19])) def test_request_row_cells_show_request_bytes(self) -> None: mod = load_module() row = { "seq": 1, "request_id": "req_1", "response_id": "resp_1", "thread_id": "thread_1", "started_at": "2026-06-30T12:00:00Z", "status_code": 200, "path": "/responses", "model": "gpt-5", "requested_model": "", "forwarded_model": "", "reasoning_tokens": 7, "reasoning_effort": "xhigh", "input_tokens": 20, "output_tokens": 10, "cached_tokens": 5, "request_body_bytes": 1536, "response_bytes_received": 4096, "response_stream": True, "stream_chunk_count": 4, "first_response_delay_ms": 1200, "duration_ms": 3400, "upstream_attempt_count": 1, "error": "", "finished_at": "2026-06-30T12:00:03Z", } cells = mod.request_row_cells(row) self.assertEqual(cells[1], "req_1") self.assertEqual(cells[2], "resp_1") self.assertEqual(cells[8], "xhigh") self.assertEqual(cells[9], "7") self.assertEqual(cells[10], "15") self.assertEqual(cells[11], "10") self.assertEqual(cells[12], "5") self.assertEqual(cells[13], "1.5KB") self.assertEqual(cells[14], "4.1KB") self.assertEqual(cells[19], "-") def test_request_retry_note_only_shows_retry_reason(self) -> None: mod = load_module() single_attempt = { "upstream_attempt_count": 1, "error": "Selected model is at capacity. Please try a different model.", } retried = { "upstream_attempt_count": 3, "error": "Selected model is at capacity. Please try a different model.", } self.assertEqual(mod.request_retry_note(single_attempt), "") self.assertEqual( mod.request_retry_note(retried), "Selected model is at capacity. Please try a different model.", ) def test_request_rows_capture_retry_round_and_wave_firsts(self) -> None: mod = load_module() payload = { "entries": [ { "seq": 3, "request_id": "req_round", "response_id": "resp_round", "thread_id": "thread_round", "path": "/responses", "status_code": 200, "reasoning_retry_current_round": 3, "reasoning_retry_current_width": 2, "reasoning_retry_current_firsts": [ { "round": 3, "slot": 1, "first_response_delay_ms": 1200, "outcome": "ok", "status_code": 200, "reasoning_effort": "high", "reasoning_tokens": 516, }, { "round": 3, "slot": 2, "first_delay_ms": 1800, "outcome": "retry", "status_code": 429, "reason": "capacity", }, ], } ] } rows = mod.normalize_request_rows(payload) row = rows[0] self.assertEqual(row["reasoning_retry_current_round"], 3) self.assertEqual(row["reasoning_retry_current_width"], 2) self.assertEqual(len(row["reasoning_retry_current_firsts"]), 2) self.assertEqual(mod.request_retry_round_text(row), "3(2)") self.assertEqual(mod.request_first_text(row), "1.2s 1.8s") self.assertIn("round 3 slot 1", mod.request_retry_wave_summary(row)) self.assertIn("retry round 3(2)", mod.render_request_detail(row)) self.assertIn("first 1.2s 1.8s", mod.render_request_detail(row)) self.assertIn("current wave round 3 slot 1 first 1.2s", mod.render_request_detail(row)) self.assertEqual(mod.request_row_cells(row)[20], "3(2)") def test_request_table_preferences_round_trip(self) -> None: mod = load_module() with tempfile.TemporaryDirectory() as tmpdir: prefs_path = Path(tmpdir) / "request-table-columns.txt" with mock.patch.dict( os.environ, {"CODEX_RETRY_GATEWAY_TUI_REQUEST_TABLE_PREFERENCES_FILE": str(prefs_path)}, clear=False, ): saved_path = mod.write_request_table_preferences( { "columns": ["round", "seq", "status", "unknown"], "hidden_columns": ["status", "unknown"], "widths": {"round": 14, "status": 9, "unknown": 99}, } ) self.assertEqual(saved_path, prefs_path) loaded = mod.load_request_table_preferences() self.assertEqual(loaded["columns"][:3], ["round", "seq", "status"]) self.assertEqual(loaded["hidden_columns"], ["status"]) self.assertEqual(loaded["widths"], {"round": 14, "status": 9}) def test_request_table_preferences_from_text_hides_and_reorders_columns(self) -> None: mod = load_module() prefs = mod.request_table_preferences_from_text( "\n".join( [ "# comment", "thread 24", "seq", "# cache 9", "in", "out", "", ] ) ) self.assertEqual(prefs["columns"][:6], ["thread", "seq", "cache", "in", "out", "req_id"]) self.assertIn("cache", prefs["hidden_columns"]) self.assertEqual(prefs["widths"], {"thread": 24, "cache": 9}) def test_request_table_preferences_expand_legacy_usage_column(self) -> None: mod = load_module() prefs = mod.normalize_request_table_preferences({"visible_columns": ["seq", "usage", "status"]}) self.assertEqual(prefs["columns"][:5], ["seq", "in", "out", "cache", "status"]) self.assertEqual(prefs["hidden_columns"][:3], ["req_id", "resp_id", "thread"]) def test_sort_request_rows_supports_custom_column_and_missing_values(self) -> None: mod = load_module() rows = [ {"seq": 1, "request_id": "req_1", "duration_ms": 5000, "reasoning_retry_current_round": 1, "reasoning_retry_current_width": 1}, {"seq": 2, "request_id": "req_2", "duration_ms": None, "reasoning_retry_current_round": None, "reasoning_retry_current_width": None}, {"seq": 3, "request_id": "req_3", "duration_ms": 1200, "reasoning_retry_current_round": 3, "reasoning_retry_current_width": 2}, ] by_duration = mod.sort_request_rows(rows, "duration", reverse=False) self.assertEqual([row["request_id"] for row in by_duration], ["req_3", "req_1", "req_2"]) by_round = mod.sort_request_rows(rows, "round", reverse=True) self.assertEqual([row["request_id"] for row in by_round], ["req_3", "req_1", "req_2"]) 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()