shujakuin/codex-retry-gateway-tui
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
aac381b53959705da164ccb62baf4690f17b78c2
codex-retry-gateway-tui/tests/test_payload.py
T

305 lines
13 KiB
Python
Raw Blame History

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, "usage": 24, "note": 22})
wide = mod.request_row_cells(row, {"req_id": 36, "resp_id": 36, "thread": 36, "path": 40, "model": 24, "usage": 72, "note": 48})
self.assertEqual(compact[10], "0B")
self.assertEqual(wide[10], "0B")
self.assertEqual(compact[8], "-")
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[9]), len(wide[9]))
self.assertLess(len(compact[16]), len(wide[16]))
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[10], "1.5KB")
self.assertEqual(cells[11], "4.1KB")
self.assertEqual(cells[16], "-")
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_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()
Reference in New Issue View Git Blame Copy Permalink