yilsemandClaude Opus 5.5
1e2b8e2b9c
Since bridge 7 external models get Codex's classic tools. One desktop tool has a top-level union schema,
which Claude rejects ("input_schema does not support oneOf, allOf, or anyOf at the top level"), so every
Opus request from the desktop app failed with a bare "upstream 400". The earlier context-size diagnosis
was wrong: the request still failed after compaction (239 KB).
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
84 lines
4.3 KiB
Python
84 lines
4.3 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("bridge.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")
|
|
|
|
|
|
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=[flatten, claude_sonu, responses], 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={})
|
|
self.env["_responses"](fake, 1, "/responses", body)
|
|
self.assertEqual(fake._passthrough.call_args.args[2], body)
|
|
self.upstream.assert_not_called()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|