yilsemandClaude Opus 5.5
8a612beba7
Measured in a live voice chat: asked whether it could change the model, Opus answered "I cannot change this chat's model here". Every request now carries one short developer note, and "modeli değiştir" without a name is answered by the bridge with how to say it. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
442 lines
25 KiB
Python
442 lines
25 KiB
Python
"""Audit of 11 Oct 2026, remaining findings: each fix has a test here (handler methods run through a fake handler)."""
|
||
import ast
|
||
import copy
|
||
import gzip
|
||
import hashlib
|
||
import io
|
||
import json
|
||
import tempfile
|
||
import time
|
||
import types
|
||
import unittest
|
||
import urllib.error
|
||
import urllib.request
|
||
from pathlib import Path
|
||
from unittest.mock import Mock
|
||
|
||
import ide_catalog
|
||
import is_hatti
|
||
import model_gecis
|
||
import ozet_kopru
|
||
import pencere
|
||
import resim_context
|
||
import yarim
|
||
|
||
SOURCE = Path(__file__).with_name("bridge.py").read_text(encoding="utf-8")
|
||
tree = ast.parse(SOURCE)
|
||
handler = next(n for n in tree.body if isinstance(n, ast.ClassDef) and n.name == "H")
|
||
|
||
|
||
def top(*names):
|
||
nodes = [n for n in tree.body if isinstance(n, (ast.FunctionDef, ast.ClassDef)) and n.name in names]
|
||
for node in tree.body:
|
||
if isinstance(node, ast.Assign) and any(isinstance(t, ast.Name) and t.id in names for t in node.targets):
|
||
nodes.insert(0, node)
|
||
return nodes
|
||
|
||
|
||
def method(name):
|
||
return next(n for n in handler.body if isinstance(n, ast.FunctionDef) and n.name == name)
|
||
|
||
|
||
def build(nodes, env):
|
||
exec(compile(ast.Module(body=nodes, type_ignores=[]), "bridge.py", "exec"), env)
|
||
return env
|
||
|
||
|
||
def user(text):
|
||
return {"type": "message", "role": "user", "content": [{"type": "input_text", "text": text}]}
|
||
|
||
|
||
def developer(text):
|
||
return {"type": "message", "role": "developer", "content": [{"type": "input_text", "text": text}]}
|
||
|
||
|
||
MODELS = [("gpt-6-sol", "GPT-6-Sol"), ("gpt-6-astra", "GPT-6-Astra"), ("gpt-6.1-sol", "GPT-6.1-Sol"),
|
||
("ide/9router-ag-gemini-3.8-flash-low", "agy/gemini-3.8-flash-low"),
|
||
("ide/9router-cc-claude-opus-5-5", "cc/opus-5-5")]
|
||
CALL = {"type": "function_call", "call_id": "c", "name": "shell", "arguments": "{}"}
|
||
OUT = {"type": "function_call_output", "call_id": "c", "output": "x"}
|
||
|
||
|
||
class SesliKomut(unittest.TestCase):
|
||
def pick(self, text):
|
||
command = model_gecis.komut_coz(text)
|
||
return model_gecis.eslestir(command["ad"], MODELS) if command and command["tur"] == "gec" else None
|
||
|
||
def test_numbers_said_as_words_and_a_continue_tail_still_switch(self):
|
||
for text, slug in [("modeli opus beş beş yap", "ide/9router-cc-claude-opus-5-5"),
|
||
("modeli gemini üç nokta sekiz flash low yap", "ide/9router-ag-gemini-3.8-flash-low"),
|
||
("modeli gpt altı nokta bir sol yap", "gpt-6.1-sol"),
|
||
("modeli gpt altı astra yap ve devam et", "gpt-6-astra"),
|
||
("modeli opus 5 5 yap sonra devam et", "ide/9router-cc-claude-opus-5-5")]:
|
||
self.assertEqual((self.pick(text) or {}).get("slug"), slug, text)
|
||
|
||
def test_a_task_after_the_command_is_not_swallowed(self):
|
||
for text in ("modeli opus 5 5 yap ve şu dosyayı düzelt", "bir model seç", "modeli iki kere değiştir"):
|
||
self.assertIsNone(self.pick(text), text)
|
||
|
||
def test_last_command_is_noted_only_when_it_really_took_effect(self):
|
||
opus = "ide/9router-cc-claude-opus-5-5"
|
||
refused = {"input": [user("modeli opus 5 5 yap"), {"type": "compaction_trigger"}]}
|
||
self.assertEqual(model_gecis.gecmisi_temizle(refused, MODELS, "gpt-6-sol"), 0) # e.g. "bağlantı kapalı"
|
||
self.assertEqual(refused["input"][0]["content"][0]["text"], "modeli opus 5 5 yap")
|
||
applied = {"input": [user("modeli opus 5 5 yap"), CALL, OUT]}
|
||
self.assertEqual(model_gecis.gecmisi_temizle(applied, MODELS, opus), 1) # mid-turn handover
|
||
self.assertIn("cc/opus-5-5 olarak değiştirdi", applied["input"][0]["content"][0]["text"])
|
||
|
||
def test_a_route_in_use_does_not_expire(self):
|
||
y = model_gecis.Yonlendirme(None)
|
||
y.koy("chat", "gpt-6-sol", "ide/x", simdi=1000)
|
||
day = 24 * 3600
|
||
for n in range(1, 12): # used every 1.5 days for 16 days
|
||
self.assertEqual(y.al("chat", "gpt-6-sol", simdi=1000 + n * 1.5 * day), "ide/x", n)
|
||
self.assertIsNone(y.al("chat", "gpt-6-sol", simdi=1000 + 16.5 * day + model_gecis.OMUR_SN + 1)) # unused for a week
|
||
|
||
|
||
class Pencere(unittest.TestCase):
|
||
def test_guard_runs_for_a_combo_whose_small_member_is_known(self):
|
||
self.assertFalse(pencere.bekci_calisir("varsayilan", 128_000))
|
||
self.assertFalse(pencere.bekci_calisir("", 128_000))
|
||
self.assertFalse(pencere.bekci_calisir("uyeler:varsayilan", 128_000))
|
||
self.assertTrue(pencere.bekci_calisir("uyeler:nanogpt,varsayilan", 8192)) # the 8K member is real
|
||
self.assertFalse(pencere.bekci_calisir("uyeler:nanogpt,varsayilan", 128_000)) # the minimum is the guess
|
||
for known in ("yerel", "nanogpt", "ogrenilen", "uyeler:nanogpt,yerel"):
|
||
self.assertTrue(pencere.bekci_calisir(known, 200_000), known)
|
||
|
||
def test_combo_with_a_claude_member_uses_the_claude_ratio(self):
|
||
combo = {"topology": {"kind": "combo", "strategy": "priority", "members": [
|
||
{"kind": "model", "model": "cx/gpt-6-astra"},
|
||
{"kind": "combo", "strategy": "priority", "members": [{"kind": "model", "model": "cc/claude-opus-5-5"}]}]}}
|
||
plain = {"topology": {"kind": "combo", "members": [{"kind": "model", "model": "ag/gemini-3.8-flash"}]}}
|
||
self.assertTrue(pencere.claude_var("combo", "Opus-5", combo))
|
||
self.assertFalse(pencere.claude_var("combo", "Hizli", plain))
|
||
self.assertTrue(pencere.claude_var("model", "no-think/claude-opus-5-5"))
|
||
self.assertFalse(pencere.claude_var("model", "nano/z-ai/glm-5.3"))
|
||
self.assertEqual(pencere.oran_sec("Opus-5", True), pencere.CLAUDE_BAYT_BOLU_TOKEN)
|
||
self.assertEqual(pencere.oran_sec("Opus-5"), pencere.BAYT_BOLU_TOKEN)
|
||
|
||
def test_a_corrupt_nano_cache_does_not_block_the_refresh(self):
|
||
home = tempfile.mkdtemp()
|
||
Path(home, pencere.NANO_DOSYA).write_text(json.dumps({"zaman": "dun", "modeller": {"eski": {"pencere": 9000}}}),
|
||
encoding="utf-8")
|
||
fresh = {"data": [{"id": "Yeni/M", "context_length": 65536, "capabilities": {}}]}
|
||
self.assertIn("yeni/m", pencere.nano_yukle(home, simdi=1000.0, getir=lambda: fresh))
|
||
|
||
def test_trimming_keeps_the_real_last_user_message_and_the_newest_tool_output(self):
|
||
marker = "(Önceki cevabın yarıda kaldı ya da sesli konuşma kapandı; bağlama göre devam et.)"
|
||
self.assertIn(marker, SOURCE) # the text bridge.py appends
|
||
self.assertTrue(marker.startswith(pencere.KOPRU_DEVAM_BASI))
|
||
req = {"input": [{"type": "function_call_output", "call_id": "a", "output": "eski " * 30000},
|
||
user("TALIMAT " * 5000),
|
||
{"type": "function_call_output", "call_id": "b", "output": "yeni " * 4000},
|
||
user(marker)]}
|
||
trimmed, fits, _ = pencere.sigdir(req, 30_000)
|
||
self.assertEqual((trimmed, fits), (1, True))
|
||
self.assertIn("çıkarıldı (köprü)", req["input"][0]["output"])
|
||
self.assertEqual(req["input"][1]["content"][0]["text"], "TALIMAT " * 5000) # the user's real last message
|
||
self.assertEqual(req["input"][2]["output"], "yeni " * 4000) # newest tool result untouched
|
||
# When nothing else is left, the newest tool output is trimmed rather than the request refused.
|
||
only = {"input": [user("soru"), {"type": "function_call_output", "call_id": "b", "output": "yeni " * 40000}]}
|
||
self.assertEqual(pencere.sigdir(only, 30_000)[:2], (1, True))
|
||
|
||
|
||
class Ozet(unittest.TestCase):
|
||
def test_a_cut_stream_is_not_returned_as_a_whole_summary(self):
|
||
delta = b'data: {"type":"response.output_text.delta","item_id":"m","delta":"yarim oz"}\n\n'
|
||
self.assertIsNone(ozet_kopru.sse_metni(delta))
|
||
done = delta + b'data: {"type":"response.completed","response":{}}\n\n'
|
||
self.assertEqual(ozet_kopru.sse_metni(done), "yarim oz")
|
||
|
||
def test_the_in_a_row_brake_lifts_after_the_window(self):
|
||
f = ozet_kopru.DonguFreni(sinir=8, pencere_sn=600, art_arda=3)
|
||
self.assertEqual([f.izin("s", t) for t in (0, 10, 20, 30)], [True, True, True, False])
|
||
self.assertFalse(f.izin("s", 500)) # still inside the window
|
||
self.assertTrue(f.izin("s", 24 * 3600)) # a day later the chat is not locked any more
|
||
|
||
def test_cleanup_survives_an_empty_record(self):
|
||
f = ozet_kopru.DonguFreni()
|
||
f.kayit = {"k%d" % i: [float(i)] for i in range(500)}
|
||
f.kayit["bos"] = []
|
||
self.assertTrue(f.izin("yeni", 1000.0))
|
||
self.assertLess(len(f.kayit), 300)
|
||
|
||
|
||
class Akis(unittest.TestCase):
|
||
env = build(top("fix_tool_ids", "SseRewriter", "phase_damgala", "asistan_mesaji_mi", "ID_PREFIX", "ARAC_OGELERI"),
|
||
{"json": json, "re": __import__("re"), "time": time, "log": lambda m: None, "pencere": pencere})
|
||
|
||
def events(self, raw):
|
||
return [json.loads(line[6:]) for line in raw.decode().splitlines() if line.startswith("data: ")]
|
||
|
||
def test_crlf_separated_events_are_translated(self):
|
||
flat = {"type": "error", "code": "bad_request", "message": "upstream 402"}
|
||
raw = b"event: error\r\ndata: " + json.dumps(flat).encode() + b"\r\n\r\n"
|
||
adapter = self.env["SseRewriter"](set())
|
||
out = self.events(adapter.feed(raw[:20]) + adapter.feed(raw[20:]) + adapter.flush())
|
||
self.assertEqual(out[-1]["type"], "response.completed")
|
||
self.assertIn("kredi ya da kota", out[1]["delta"])
|
||
watcher = yarim.Izleyici("k", "m")
|
||
watcher.feed(b'data: {"type":"response.output_text.delta","delta":"merhaba"}\r\n\r\n'
|
||
b'data: {"type":"response.completed"}\r\n\r\n')
|
||
self.assertEqual((watcher.metin, watcher.bitti), ("merhaba", True))
|
||
|
||
def test_a_non_object_event_body_passes_through(self):
|
||
adapter = self.env["SseRewriter"](set())
|
||
self.assertEqual(adapter.feed(b"data: null\n\ndata: [1]\n\n"), b"data: null\n\ndata: [1]\n\n")
|
||
|
||
def test_a_waiting_message_is_released_before_the_error_answer(self):
|
||
message = {"type": "response.output_item.done", "output_index": 0,
|
||
"item": {"id": "m1", "type": "message", "role": "assistant", "content": []}}
|
||
adapter = self.env["SseRewriter"](set(), dolu=True)
|
||
first = adapter.feed(b"data: " + json.dumps(message).encode() + b"\n\n")
|
||
self.assertEqual(first, b"") # held until what follows is known
|
||
out = self.events(adapter.feed(b'data: {"type":"error","message":"upstream 400"}\n\n') + adapter.flush())
|
||
self.assertEqual([e["type"] for e in out], ["response.output_item.done", "response.failed"])
|
||
self.assertEqual(out[0]["item"]["id"], "m1")
|
||
self.assertIn("nedenini bildirmedi", out[1]["response"]["error"]["message"]) # no false "did not fit"
|
||
self.assertIn("olabilir", out[1]["response"]["error"]["message"])
|
||
|
||
|
||
class Resim(unittest.TestCase):
|
||
def test_a_new_image_in_a_message_without_type_is_kept(self):
|
||
img = {"type": "input_image", "image_url": "data:image/png;base64," + "A" * 2000}
|
||
req = {"input": [{"role": "user", "content": [{"type": "input_text", "text": "bak"}, dict(img)]}]}
|
||
self.assertEqual(resim_context.resimleri_sadelestir(req), (0, 0))
|
||
self.assertEqual(req["input"][0]["content"][1]["type"], "input_image")
|
||
|
||
|
||
class Katalog(unittest.TestCase):
|
||
def test_names_stay_distinct_when_the_id_tail_collides_too(self):
|
||
names = ide_catalog.adlandir([("a-ortakson", "Ad"), ("b-ortakson", "Ad"), ("c-ortakson", "Ad")], dolu={"ad"})
|
||
self.assertEqual(len({v.lower() for v in names.values()}), 3)
|
||
|
||
def test_an_original_listed_as_extra_keeps_its_place(self):
|
||
models = [{"slug": "gpt-6-sol", "display_name": "GPT-6-Sol", "priority": 1},
|
||
{"slug": "gpt-6-astra", "display_name": "GPT-6-Astra", "priority": 2},
|
||
{"slug": "ide/a", "display_name": "Aaa", "priority": 0}]
|
||
extras = [{"slug": "gpt-6-astra"}, {"slug": "ide/a", "kind": "combo"}]
|
||
ide_catalog.sirala_codex(models, extras, {"gpt-6-sol", "gpt-6-astra"})
|
||
self.assertEqual([(m["slug"], m["priority"]) for m in models],
|
||
[("gpt-6-sol", 1), ("gpt-6-astra", 2), ("ide/a", 3)])
|
||
|
||
|
||
class Sahte:
|
||
"""A handler stand-in: real bridge methods, recorded output."""
|
||
|
||
def __init__(self, headers=None, command="POST"):
|
||
self.headers = headers or {}
|
||
self.command = command
|
||
self.path = "/backend-api/codex/responses"
|
||
self.wfile = io.BytesIO()
|
||
self.status = None
|
||
self.sent = []
|
||
self._passthrough = Mock()
|
||
self._yerel_yanit = Mock()
|
||
self._send_json = Mock()
|
||
self._stream = Mock()
|
||
self._tasma_ogren = Mock()
|
||
self._aracsiz_ogren = Mock()
|
||
|
||
def send_response(self, status):
|
||
self.status = status
|
||
|
||
def send_header(self, k, v):
|
||
self.sent.append((k, v))
|
||
|
||
def end_headers(self):
|
||
pass
|
||
|
||
|
||
class Zincir(unittest.TestCase):
|
||
"""_responses + _model_komutu with a conversation key, the way Codex sends it."""
|
||
|
||
def setUp(self):
|
||
self.env = build(top("nokta_mesaji_mi", "NOKTA_ISARETLERI", "is_timeout", "kopma_mi", "flatten_request",
|
||
"claude_sonu_duzelt") + [method("_responses"), method("_model_komutu")], {
|
||
"json": json, "re": __import__("re"), "os": __import__("os"), "copy": copy, "http": __import__("http.client"),
|
||
"time": types.SimpleNamespace(sleep=lambda s: None, time=time.time),
|
||
"urllib": types.SimpleNamespace(request=types.SimpleNamespace(Request=urllib.request.Request, urlopen=Mock()),
|
||
error=urllib.error),
|
||
"decompress": lambda body, enc: body, "MODEL_MAP": {}, "ROUTES": {}, "CFG": {"model_bilgi": False}, "log": lambda m: None,
|
||
"model_gecis": model_gecis, "YONLENDIRME": model_gecis.Yonlendirme(None), "bilinen_modeller": lambda: MODELS,
|
||
"model_adi": lambda slug: dict(MODELS).get(slug, slug), "is_hatti": is_hatti, "GORULEN_ISTEMCI": set(),
|
||
"yarim": yarim, "ozet_kopru": ozet_kopru, "OZET_FRENI": ozet_kopru.DonguFreni(), "pencere": pencere,
|
||
"fix_tool_ids": lambda req: 0, "drop_foreign_items": lambda req: 0, "IDE_BILGI": {}, "EXTRA_MODELS": [],
|
||
"arac_semalarini_duzelt": lambda req: 0, "SseRewriter": Mock(), "DEBUG": False, "UA": "test", "RETRIES": 6,
|
||
"UPSTREAM_IDLE_TIMEOUT": 1, "NO_RETRY_PATTERNS": [], "pano_blogunu_ulastir": lambda req: (0, ""),
|
||
"resimleri_sadelestir": lambda req, gorsel=False: (0, 0),
|
||
})
|
||
Sahte._responses = self.env["_responses"]
|
||
Sahte._model_komutu = self.env["_model_komutu"]
|
||
|
||
def send(self, model, items, key="sohbet-1", path="/responses"):
|
||
fake = Sahte()
|
||
body = json.dumps({"model": model, "prompt_cache_key": key, "input": items}).encode()
|
||
fake._responses(1, path, body)
|
||
return fake, body
|
||
|
||
def test_mid_turn_note_reaches_the_subscription_model_behind_hook_messages(self):
|
||
items = [user("işe başla"), CALL, OUT, user("modeli sol yap"), developer("k1"), developer("k2"), developer("k3")]
|
||
fake, body = self.send("gpt-6-astra", items)
|
||
fake._yerel_yanit.assert_not_called()
|
||
args, kwargs = fake._passthrough.call_args
|
||
self.assertTrue(kwargs["rewritten"])
|
||
self.assertNotEqual(args[2], body)
|
||
sent = json.loads(args[2])
|
||
self.assertIn("ad belirsizdi", sent["input"][3]["content"][0]["text"])
|
||
self.assertEqual(sent["model"], "gpt-6-astra")
|
||
|
||
def test_idle_switch_says_how_to_return_and_routes_the_next_request(self):
|
||
fake, _ = self.send("gpt-6-sol", [user("modeli gpt 6 astra yap")])
|
||
answer = fake._yerel_yanit.call_args.args[1]
|
||
self.assertIn("Sohbetin modeli artık GPT-6-Astra", answer)
|
||
self.assertIn('"modeli GPT-6-Sol yap"', answer)
|
||
final = {"type": "message", "role": "assistant", "phase": "final_answer", "content": "ok"}
|
||
fake, _ = self.send("gpt-6-sol", [user("modeli gpt 6 astra yap"), final, user("devam")])
|
||
sent = json.loads(fake._passthrough.call_args.args[2])
|
||
self.assertEqual(sent["model"], "gpt-6-astra")
|
||
self.assertIn("GPT-6-Astra olarak değiştirdi", sent["input"][0]["content"][0]["text"])
|
||
back, _ = self.send("gpt-6-sol", [user("modeli gpt 6 astra yap"), final, user("modeli gpt 6 sol yap")])
|
||
self.assertNotIn("dönmek için", back._yerel_yanit.call_args.args[1]) # already back on the picked model
|
||
|
||
def test_a_refused_switch_is_not_told_as_done_after_compaction(self):
|
||
items = [user("modeli opus 5 5 yap"), {"type": "compaction_trigger"}] # the ide/ route is closed here
|
||
fake, body = self.send("gpt-6-sol", items)
|
||
self.assertEqual(fake._passthrough.call_args.args[2], body)
|
||
|
||
def test_lone_dot_answer_says_selected_not_ready(self):
|
||
self.env["CFG"]["nokta_karsila"] = True
|
||
fake, _ = self.send("gpt-6-sol", [user(".")])
|
||
self.assertIn("GPT-6-Sol seçildi", fake._yerel_yanit.call_args.args[1])
|
||
self.assertNotIn("hazır", fake._yerel_yanit.call_args.args[1])
|
||
|
||
def test_old_router_name_without_a_route_is_refused_locally(self):
|
||
fake, _ = self.send("9r/cc/claude-opus-5-5", [user("selam")])
|
||
self.assertEqual(fake._send_json.call_args.args[0], 503)
|
||
fake._passthrough.assert_not_called()
|
||
|
||
def test_a_route_removed_while_the_request_is_prepared_gives_503(self):
|
||
self.env["MODEL_MAP"]["ide/x"] = ("aiturk-ide", "x", None)
|
||
fake, _ = self.send("ide/x", [user("selam")])
|
||
self.assertEqual(fake._send_json.call_args.args[0], 503)
|
||
self.env["urllib"].request.urlopen.assert_not_called()
|
||
|
||
def external(self, error):
|
||
self.env["ROUTES"]["r"] = {"url": "https://example.invalid/v1/responses", "key": "test-only"}
|
||
self.env["MODEL_MAP"].update({"x/combo": ("r", "Takim", None), "x/tek": ("r", "cc/claude", None)})
|
||
self.env["urllib"].request.urlopen.side_effect = error
|
||
|
||
def test_combo_dropped_after_sending_is_not_resent_but_a_single_model_is(self):
|
||
import http.client
|
||
self.external(http.client.RemoteDisconnected("closed"))
|
||
self.send("x/combo", [user("selam")])
|
||
self.assertEqual(self.env["urllib"].request.urlopen.call_count, 1)
|
||
self.env["urllib"].request.urlopen.reset_mock()
|
||
self.external([urllib.error.URLError(ConnectionResetError("reset")), Mock(status=200)])
|
||
fake, _ = self.send("x/tek", [user("selam")])
|
||
self.assertEqual(self.env["urllib"].request.urlopen.call_count, 2)
|
||
fake._stream.assert_called_once()
|
||
self.env["urllib"].request.urlopen.reset_mock()
|
||
self.external([urllib.error.URLError(ConnectionRefusedError("refused")), Mock(status=200)])
|
||
self.send("x/combo", [user("selam")]) # never connected: safe to retry
|
||
self.assertEqual(self.env["urllib"].request.urlopen.call_count, 2)
|
||
|
||
def test_error_body_reaches_the_client_whole_with_retry_after(self):
|
||
whole = json.dumps({"error": {"message": "limit " + "x" * 900, "code": "rate_limit_exceeded"}}).encode()
|
||
self.external(urllib.error.HTTPError("https://example.invalid", 429, "x", {"Retry-After": "17"}, io.BytesIO(whole)))
|
||
fake, _ = self.send("x/tek", [user("selam")])
|
||
self.assertEqual(fake.status, 429)
|
||
self.assertEqual(fake.wfile.getvalue(), whole)
|
||
self.assertIn(("Retry-After", "17"), fake.sent)
|
||
self.assertIn(("Content-Length", str(len(whole))), fake.sent)
|
||
|
||
def test_image_in_an_orphan_tool_output_is_removed_before_flattening(self):
|
||
self.env["resimleri_sadelestir"] = resim_context.resimleri_sadelestir
|
||
self.env["CFG"]["nine_router_catalog"] = {"enabled": True, "route": "r", "prefix": "9r/"}
|
||
self.external(None)
|
||
self.env["ROUTES"]["r"]["flatten_tools"] = True
|
||
self.env["urllib"].request.urlopen.return_value = Mock(status=200)
|
||
picture = {"type": "input_image", "image_url": "data:image/png;base64," + "QUJD" * 500}
|
||
self.send("x/tek", [{"type": "function_call_output", "name": "image_gen", "output": [picture]}, user("devam")])
|
||
sent = self.env["urllib"].request.urlopen.call_args.args[0].data.decode()
|
||
self.assertNotIn("QUJDQUJD", sent)
|
||
self.assertIn("resim", sent)
|
||
|
||
|
||
class Govde(unittest.TestCase):
|
||
read = build([method("_read")], {})["_read"]
|
||
|
||
def test_chunked_upload_cut_short_ends_instead_of_spinning(self):
|
||
fake = types.SimpleNamespace(headers={"Transfer-Encoding": "chunked"}, rfile=io.BytesIO(b"5\r\nhello\r\n"))
|
||
self.assertEqual(Govde.read(fake), b"hello")
|
||
fake = types.SimpleNamespace(headers={"Transfer-Encoding": "chunked"}, rfile=io.BytesIO(b"5\r\nhello\r\n0\r\n\r\n"))
|
||
self.assertEqual(Govde.read(fake), b"hello")
|
||
|
||
|
||
class Liste(unittest.TestCase):
|
||
def env(self, **extra):
|
||
env = {"json": json, "copy": copy, "hashlib": hashlib, "time": time, "log": lambda m: None, "CFG": {},
|
||
"decompress": lambda raw, enc: raw, "EXTRA_MODELS": [], "IDE_PREFIX": "ide/", "SLIM_CLONES": False,
|
||
"SLIM_FIELDS": [], "ide_catalog": ide_catalog, "LAST_MODELS": {}, "save_last_models": Mock(),
|
||
"RESP_HOP": {"content-length", "content-encoding"}, "REQ_HOP": {"host"}, "CHATGPT": "https://example.invalid",
|
||
"UPSTREAM_IDLE_TIMEOUT": 1,
|
||
"urllib": types.SimpleNamespace(request=types.SimpleNamespace(Request=urllib.request.Request, urlopen=Mock()),
|
||
error=urllib.error)}
|
||
env.update(extra)
|
||
return build([method("_models"), method("_passthrough"), method("_son_listeyi_ver")], env)
|
||
|
||
UPSTREAM = {"models": [{"slug": "gpt-6-sol", "display_name": "GPT-6-Sol", "priority": 1},
|
||
{"slug": "gpt-6-astra", "display_name": "GPT-6-Astra", "priority": 2}]}
|
||
|
||
def test_gzip_answer_is_still_cached_plain_and_originals_keep_name_and_place(self):
|
||
extras = [{"slug": "gpt-6-astra", "display_name": "Baska Ad"}, {"slug": "ide/a", "display_name": "Aaa", "kind": "combo"}]
|
||
env = self.env(EXTRA_MODELS=extras, CFG={"display_map": {"gpt-6-sol": "Yeni Ad"}})
|
||
fake = Sahte({"Accept-Encoding": "gzip", "chatgpt-account-id": "hesap"}, "GET")
|
||
resp = types.SimpleNamespace(read=lambda: json.dumps(self.UPSTREAM).encode(), headers={"ETag": "x"})
|
||
env["_models"](fake, resp, "[t]")
|
||
listed = json.loads(gzip.decompress(fake.wfile.getvalue()))["models"]
|
||
self.assertEqual([(m["slug"], m["display_name"]) for m in listed],
|
||
[("gpt-6-sol", "GPT-6-Sol"), ("gpt-6-astra", "GPT-6-Astra"), ("ide/a", "Aaa")])
|
||
cached = env["LAST_MODELS"]["hesap"]
|
||
self.assertEqual(json.loads(cached["body"])["models"], listed) # plain JSON, usable as a fallback
|
||
self.assertNotIn("Content-Encoding", dict(cached["headers"]))
|
||
self.assertIn(("Content-Encoding", "gzip"), fake.sent)
|
||
env["save_last_models"].assert_called_once()
|
||
|
||
def test_no_network_still_serves_the_last_list(self):
|
||
cached = {"headers": [("Content-Type", "application/json")], "body": b'{"models": []}', "at": "x"}
|
||
env = self.env(LAST_MODELS={"hesap": cached})
|
||
env["urllib"].request.urlopen.side_effect = urllib.error.URLError(OSError("no network"))
|
||
fake = Sahte({"chatgpt-account-id": "hesap"}, "GET")
|
||
fake._son_listeyi_ver = types.MethodType(env["_son_listeyi_ver"], fake)
|
||
env["_passthrough"](fake, 1, "/backend-api/codex/models?client_version=1", b"")
|
||
self.assertEqual((fake.status, fake.wfile.getvalue()), (200, b'{"models": []}'))
|
||
other = Sahte({"chatgpt-account-id": "baska"}, "GET")
|
||
other._son_listeyi_ver = types.MethodType(env["_son_listeyi_ver"], other)
|
||
env["_passthrough"](other, 1, "/backend-api/codex/models", b"")
|
||
self.assertEqual(other._send_json.call_args.args[0], 502)
|
||
|
||
|
||
class Kucukler(unittest.TestCase):
|
||
def test_log_line_has_the_date_and_no_token_tail_and_status_reports_the_code(self):
|
||
self.assertIn('time.strftime("%Y-%m-%d %H:%M:%S ") + msg', SOURCE)
|
||
self.assertNotIn("token=..", SOURCE)
|
||
self.assertNotIn("DISPLAY_MAP", SOURCE)
|
||
self.assertIn('"kod": KOD', SOURCE)
|
||
|
||
def test_responses_path_with_a_query_is_still_routed(self):
|
||
env = build([method("_handle")], {"lock": __import__("threading").Lock(), "seq": [0], "DEBUG": False,
|
||
"log": lambda m: None})
|
||
fake = Sahte({})
|
||
fake.path = "/backend-api/codex/responses?stream=true"
|
||
fake._read = lambda: b"{}"
|
||
fake._responses = Mock()
|
||
env["_handle"](fake)
|
||
fake._responses.assert_called_once()
|
||
fake._passthrough.assert_not_called()
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|