yilsemandClaude Opus 5.5
23e8aa8191
bridge.py is now a small splitter with no features: a request for one of Codex's own models goes to chatgpt.com byte for byte, a request for an added model goes to kopru.py, which runs as its own process. Codex's own models no longer pass through the feature code at all, and restarting or losing the bridge process does not touch them. Measured with real Codex on spare ports: own model answers; forced compaction on the own model completes (twice) while the bridge process is restarted; added model answers; no own-model request reaches the bridge process. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
88 lines
4.6 KiB
Python
88 lines
4.6 KiB
Python
"""Regression coverage for Codex -> external API access_programs compatibility."""
|
|
import ast
|
|
import copy
|
|
import json
|
|
from pathlib import Path
|
|
import unittest
|
|
from unittest.mock import Mock
|
|
import urllib.request
|
|
|
|
|
|
# Load only the actual adapter and routing method: no catalog/network/thread
|
|
# startup or credential loading is needed to exercise the production code.
|
|
source = ast.parse(Path(__file__).with_name("kopru.py").read_text(encoding="utf-8"))
|
|
flatten = next(n for n in source.body if isinstance(n, ast.FunctionDef) and n.name == "flatten_request")
|
|
claude_sonu = next(n for n in source.body if isinstance(n, ast.FunctionDef) and n.name == "claude_sonu_duzelt")
|
|
handler = next(n for n in source.body if isinstance(n, ast.ClassDef) and n.name == "H")
|
|
responses = next(n for n in handler.body if isinstance(n, ast.FunctionDef) and n.name == "_responses")
|
|
ozgun = next(n for n in handler.body if isinstance(n, ast.FunctionDef) and n.name == "_ozgun")
|
|
taraf = [n for n in source.body if isinstance(n, ast.FunctionDef) and n.name in ("ek_model_mu", "yonlendirilen")]
|
|
|
|
|
|
class AccessPrograms(unittest.TestCase):
|
|
def setUp(self):
|
|
self.upstream = Mock(return_value=Mock(status=200))
|
|
self.env = {
|
|
"json": json,
|
|
"urllib": Mock(request=Mock(Request=urllib.request.Request, urlopen=self.upstream)),
|
|
"decompress": lambda body, encoding: body,
|
|
"MODEL_MAP": {},
|
|
"CFG": {"nine_router_catalog": {"enabled": True, "route": "9router", "prefix": "9r/"},
|
|
"resim_sadelestir": False, "pano_kanca_user": False},
|
|
"ROUTES": {"9router": {"url": "https://example.invalid/v1/responses", "key": "test-only", "flatten_tools": True}},
|
|
"fix_tool_ids": lambda req: 0,
|
|
"drop_foreign_items": lambda req: 0,
|
|
"SseRewriter": Mock(),
|
|
"log": lambda message: None, "pencere": __import__("pencere"), "IDE_BILGI": {}, "yarim": __import__("yarim"), "is_hatti": __import__("is_hatti"), "GORULEN_ISTEMCI": set(), "arac_semalarini_duzelt": lambda req: 0, "nokta_mesaji_mi": lambda req: False, "EXTRA_MODELS": [],
|
|
"DEBUG": False, "UA": "test", "RETRIES": 1, "UPSTREAM_IDLE_TIMEOUT": 3900,
|
|
}
|
|
exec(compile(ast.Module(body=taraf + [flatten, claude_sonu, responses, ozgun], type_ignores=[]), "bridge.py", "exec"), self.env)
|
|
|
|
def request(self, model):
|
|
return {"model": model, "access_programs": {"cyber": "standard"},
|
|
"metadata": {"access_programs": "preserve nested data"},
|
|
"input": [{"role": "user", "content": [{"type": "input_text", "text": "hello"}]}],
|
|
"tools": [{"type": "function", "name": "echo", "parameters": {"type": "object", "properties": {}}}]}
|
|
|
|
def test_adapter_removes_only_top_level_field(self):
|
|
req = self.request("meta/muse-spark-1.3")
|
|
original = copy.deepcopy(req)
|
|
self.env["flatten_request"](req)
|
|
self.assertNotIn("access_programs", req)
|
|
self.assertEqual(req["metadata"], original["metadata"])
|
|
self.assertEqual(req["input"], original["input"])
|
|
self.assertEqual(req["tools"][0]["name"], "echo")
|
|
|
|
def test_absent_null_and_empty_values(self):
|
|
for value in (None, {}, {"cyber": "standard"}):
|
|
with self.subTest(value=value):
|
|
req = {"input": [], "access_programs": value}
|
|
self.env["flatten_request"](req)
|
|
self.assertNotIn("access_programs", req)
|
|
self.env["flatten_request"]({"input": []})
|
|
|
|
def test_model_and_combo_external_request_strips_field(self):
|
|
for model in ("9r/meta/muse-spark-1.3-contributor", "9r/Fusion-Hizli"):
|
|
with self.subTest(model=model):
|
|
body = json.dumps(self.request(model)).encode()
|
|
fake = Mock(headers={})
|
|
self.env["_responses"](fake, 1, "/responses", body)
|
|
sent = json.loads(self.upstream.call_args.args[0].data)
|
|
self.assertNotIn("access_programs", sent)
|
|
self.assertEqual(sent["model"], model.removeprefix("9r/"))
|
|
self.assertIn("access_programs", json.loads(body))
|
|
fake._stream.assert_called_once()
|
|
|
|
def test_chatgpt_passthrough_preserves_original_body(self):
|
|
body = json.dumps(self.request("gpt-6-astra")).encode()
|
|
fake = Mock(headers={})
|
|
fake._ozgun = lambda *a: self.env["_ozgun"](fake, *a)
|
|
self.env["_responses"](fake, 1, "/responses", body)
|
|
self.assertEqual(fake._passthrough.call_args.args[2], body)
|
|
self.assertFalse(fake._passthrough.call_args.kwargs["rewritten"])
|
|
self.upstream.assert_not_called()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|