573 lines
26 KiB
Python
573 lines
26 KiB
Python
from __future__ import annotations
|
|
|
|
import datetime as dt
|
|
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("text 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_image_profile_rows_show_active_first(self) -> None:
|
|
mod = load_module()
|
|
payload = {
|
|
"image_profiles": [
|
|
{"name": "beta", "active": False, "summary": {"base_url": "https://images-b.example/v1", "auth_mode": "fixed_bearer", "auth_source": "env"}},
|
|
{"name": "alpha", "active": True, "summary": {"base_url": "https://images-a.example/v1", "auth_mode": "manual_bearer", "auth_source": "manual_file"}},
|
|
]
|
|
}
|
|
rows = mod.normalize_image_profile_rows(payload)
|
|
self.assertEqual([row["name"] for row in rows], ["alpha", "beta"])
|
|
self.assertEqual(rows[0]["base_url"], "https://images-a.example/v1")
|
|
|
|
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_status_labels_follow_lifecycle_and_age(self) -> None:
|
|
mod = load_module()
|
|
now = dt.datetime(2026, 7, 10, 12, 0, tzinfo=dt.timezone.utc)
|
|
waiting = {
|
|
"lifecycle_state": "sent",
|
|
"started_at": "2026-07-10T11:59:50Z",
|
|
}
|
|
streaming = {
|
|
"lifecycle_state": "receive_first",
|
|
"started_at": "2026-07-10T11:59:00Z",
|
|
"first_response_at": "2026-07-10T11:59:55Z",
|
|
"last_activity_at": "2026-07-10T11:59:55Z",
|
|
}
|
|
retry_streaming = {
|
|
"lifecycle_state": "sent",
|
|
"started_at": "2026-07-10T11:59:00Z",
|
|
"reasoning_retry_current_firsts": [
|
|
{
|
|
"round": 1,
|
|
"slot": 1,
|
|
"first_response_at": "2026-07-10T11:59:58Z",
|
|
"first_delay_ms": 2200,
|
|
}
|
|
],
|
|
}
|
|
complete = {"lifecycle_state": "finish", "status_code": 200}
|
|
discarded = {
|
|
"lifecycle_state": "finish",
|
|
"status_code": 502,
|
|
"error": "client disconnected before reasoning retry completed",
|
|
}
|
|
slow_waiting = {
|
|
"lifecycle_state": "sent",
|
|
"started_at": "2026-07-10T11:59:30Z",
|
|
}
|
|
stalled = {
|
|
"lifecycle_state": "streaming",
|
|
"first_response_at": "2026-07-10T11:57:50Z",
|
|
"last_activity_at": "2026-07-10T11:57:50Z",
|
|
}
|
|
|
|
self.assertEqual(mod.request_status_label(waiting, now), "waiting")
|
|
self.assertEqual(mod.request_status_label(streaming, now), "streaming")
|
|
self.assertEqual(mod.request_status_label(retry_streaming, now), "streaming")
|
|
self.assertEqual(mod.request_activity_at(retry_streaming), "2026-07-10T11:59:58Z")
|
|
self.assertEqual(mod.request_status_label(complete, now), "HTTP 200")
|
|
self.assertEqual(mod.request_status_label(discarded, now), "discarded")
|
|
self.assertEqual(mod.request_status_label(slow_waiting, now), "waiting 30.0s")
|
|
self.assertEqual(mod.request_status_label(stalled, now), "stalled 130s")
|
|
self.assertNotEqual(mod.request_status_style(waiting, now), mod.request_status_style(streaming, now))
|
|
self.assertEqual(mod.request_status_style(discarded, now), "bold #ff5f5f")
|
|
self.assertEqual(mod.request_status_style(stalled, now), "bold #ff5f5f")
|
|
|
|
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_visible_cells_colorize_thread_identity_columns(self) -> None:
|
|
mod = load_module()
|
|
try:
|
|
from rich.text import Text
|
|
except ImportError:
|
|
self.skipTest("rich is not installed")
|
|
row = {
|
|
"seq": 1,
|
|
"request_id": "req_1",
|
|
"response_id": "resp_1",
|
|
"thread_id": "thread_same",
|
|
"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)
|
|
visible = mod.request_visible_cells(row, cells, ["seq", "req_id", "resp_id", "thread", "status"])
|
|
self.assertEqual(visible[0], "1")
|
|
self.assertIsInstance(visible[1], Text)
|
|
self.assertIsInstance(visible[2], Text)
|
|
self.assertIsInstance(visible[3], Text)
|
|
self.assertEqual(visible[1].plain, "req_1")
|
|
self.assertEqual(visible[2].plain, "resp_1")
|
|
self.assertEqual(visible[3].plain, "thread_same")
|
|
self.assertIsInstance(visible[4], Text)
|
|
self.assertEqual(visible[4].plain, "HTTP 200")
|
|
self.assertEqual(mod.request_thread_color("thread_same"), mod.request_thread_color("thread_same"))
|
|
self.assertIsNone(mod.request_thread_color(""))
|
|
|
|
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_text_profile_editor_payload_does_not_carry_image_fields(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": {
|
|
"image_base_url": "https://images.example/v1",
|
|
"image_auth_mode": "manual_bearer",
|
|
"image_manual_secret_file": "/run/secrets/images-token",
|
|
"image_manual_secret_configured": True,
|
|
"image_auth_json_key": "IMAGE_API_KEY",
|
|
"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.assertNotIn("image_base_url", payload)
|
|
self.assertNotIn("image_base_url", text)
|
|
self.assertNotIn("test-image-profile-secret", text)
|
|
saved_payload = mod.profile_payload_from_row(row)
|
|
self.assertNotIn("image_base_url", saved_payload)
|
|
self.assertIn(
|
|
"stream disconnected before completion: Concurrency limit exceeded for account, please retry later",
|
|
payload["retryable_error_messages"],
|
|
)
|
|
|
|
def test_image_profile_editor_payload_round_trip(self) -> None:
|
|
mod = load_module()
|
|
row = {
|
|
"name": "images",
|
|
"base_url": "https://images.example/v1",
|
|
"auth_mode": "manual_bearer",
|
|
"auth_source": "manual_file",
|
|
"raw": {
|
|
"form": {
|
|
"base_url": "https://images.example/v1",
|
|
"auth_mode": "manual_bearer",
|
|
"manual_secret_file": "/run/secrets/images-token",
|
|
"manual_secret_configured": True,
|
|
"auth_json_key": "IMAGE_API_KEY",
|
|
}
|
|
},
|
|
}
|
|
text = mod.image_profile_editor_document(row)
|
|
payload = mod.image_profile_payload_from_editor_text(text)
|
|
self.assertEqual(payload["name"], "images")
|
|
self.assertEqual(payload["base_url"], "https://images.example/v1")
|
|
self.assertEqual(payload["auth_mode"], "manual_bearer")
|
|
self.assertEqual(payload["manual_secret_file"], "/run/secrets/images-token")
|
|
self.assertTrue(payload["manual_secret_configured"])
|
|
self.assertEqual(payload["manual_secret"], "")
|
|
self.assertNotIn("test-image-profile-secret", text)
|
|
saved_payload = mod.image_profile_payload_from_row(row)
|
|
self.assertEqual(saved_payload["base_url"], "https://images.example/v1")
|
|
self.assertEqual(saved_payload["auth_mode"], "manual_bearer")
|
|
self.assertEqual(saved_payload["manual_secret_file"], "/run/secrets/images-token")
|
|
self.assertEqual(saved_payload["manual_secret"], "")
|
|
|
|
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()
|