114 lines
5.4 KiB
Python
114 lines
5.4 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", "path": "/responses"},
|
|
{"seq": 2, "request_id": "req_b", "path": "/v1/responses"},
|
|
]
|
|
}
|
|
rows = mod.normalize_request_rows(payload)
|
|
self.assertEqual([row["request_id"] for row in rows], ["req_b", "req_a"])
|
|
|
|
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",
|
|
)
|
|
self.assertIn("updated", mod.render_request_detail({"seq": 1, "request_id": "r", "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": 4, "stream_chunk_count": 1, "started_at": "2026-06-30T00:00:00Z", "finished_at": "2026-06-30T00:00:01Z", "usage_last_updated_at": "2026-06-30T00:00:01Z"}))
|
|
|
|
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()
|