178 lines
7.7 KiB
Python
178 lines
7.7 KiB
Python
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_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()
|