"""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, patch import ide_catalog import is_hatti import model_gecis import ozet_kopru import pencere import resim_context import yarim SOURCE = Path(__file__).with_name("kopru.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"), {"WATCH_OWNER": None, "WatchResponse": lambda response, accept: response, "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", "ek_model_mu", "yonlendirilen") + [method("_responses"), method("_model_komutu"), method("_ozgun")], { "gecit_modelleri": lambda: [], "WATCH_OWNER": None, "WatchResponse": lambda response, accept: response, "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, "jev_denetim": __import__("jev_denetim"), "internet": __import__("internet"), "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"] Sahte._ozgun = self.env["_ozgun"] 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_a_request_for_codex_own_model_is_passed_through_untouched(self): """User decision, 11 Oct 2026: Codex's own models are one side, the added models the other; the bridge does not touch a request for Codex's own model (no command, dot, note, work line or compaction handling).""" self.env["CFG"].update({"nokta_karsila": True, "model_bilgi": True, "yarim_cevap": True, "ozet_kopru": {"enabled": True, "kapsam": "hepsi"}, "aiturk_hat": {"enabled": True, "komut": "aiturk is"}}) mid_turn = [user("işe başla"), CALL, OUT, user("modeli opus 5 5 yap"), developer("k1")] for items in ([user("modeli opus 5 5 yap")], [user(".")], [user("seçili model hangisi")], mid_turn, [user("uzun iş"), {"type": "compaction_trigger"}]): with self.subTest(items=items): fake = Sahte(headers={"originator": "Codex Desktop"}) body = json.dumps({"model": "gpt-6-astra", "prompt_cache_key": "sohbet-1", "input": items}).encode() fake._responses(1, "/responses", body) fake._yerel_yanit.assert_not_called() args, kwargs = fake._passthrough.call_args self.assertEqual(args[2], body) self.assertFalse(kwargs["rewritten"]) self.assertEqual(self.env["YONLENDIRME"].veri, {}) def test_the_bridge_does_not_even_help_codex_own_model(self): """'If you pick Codex and write to Codex, our bridge cannot be beside it even to help.' Traces an added model or the bridge left earlier in the chat are not cleaned either: the body goes out byte for byte.""" self.env["CFG"]["ozet_kopru"] = {"enabled": True, "kapsam": "hepsi"} self.env["fix_tool_ids"] = Mock(return_value=2) self.env["drop_foreign_items"] = Mock(return_value=3) summary = ozet_kopru.ozet_ogesi("eski köprü özeti", "t") fake, body = self.send("gpt-6-astra", [summary, {"type": "reasoning", "id": "rs_chatcmpl-x"}, user("devam")]) args, kwargs = fake._passthrough.call_args self.assertEqual(args[2], body) self.assertFalse(kwargs["rewritten"]) self.env["fix_tool_ids"].assert_not_called() self.env["drop_foreign_items"].assert_not_called() fake._yerel_yanit.assert_not_called() def test_voice_switch_stays_among_the_added_models(self): opus, flash = "ide/9router-cc-claude-opus-5-5", "ide/9router-ag-gemini-3.8-flash-low" self.env["ROUTES"]["aiturk-ide"] = {"url": "https://example.invalid/v1/responses", "key": "k"} fake = Sahte() answer = fake._model_komutu(1, {"input": [user("modeli gemini flash low yap")]}, "sohbet-1", opus) self.assertIn("Sohbetin modeli artık agy/gemini-3.8-flash-low", answer) self.assertEqual(self.env["yonlendirilen"]("sohbet-1", opus), flash) # Codex's own model is not a voice target, from an added model either. answer = fake._model_komutu(1, {"input": [user("modeli gpt 6 astra yap")]}, "sohbet-1", opus) self.assertIn("GPT-6-Astra Codex'in kendi modeli", answer) self.assertIn("sesle geçilmiyor", answer) self.assertIn('"."', answer) self.assertEqual(self.env["yonlendirilen"]("sohbet-1", opus), flash) # In the middle of a turn the running model is told to pass that on; the work is not cut. req = {"input": [user("işe başla"), CALL, OUT, user("modeli gpt 6 astra yap")]} self.assertIsNone(fake._model_komutu(1, req, "sohbet-1", opus)) self.assertIn("sesle yapılamıyor", req["input"][3]["content"][0]["text"]) # A record left from before the rule that points at Codex's own model is ignored. self.env["YONLENDIRME"].koy("sohbet-2", opus, "gpt-6-astra") self.assertIsNone(self.env["yonlendirilen"]("sohbet-2", opus)) 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_is_answered_for_an_added_model_and_left_alone_for_codex_own(self): self.env["CFG"]["nokta_karsila"] = True fake, _ = self.send("ide/9router-cc-claude-opus-5-5", [user(".")]) self.assertIn("cc/opus-5-5 seçildi", fake._yerel_yanit.call_args.args[1]) self.assertNotIn("hazır", fake._yerel_yanit.call_args.args[1]) fake, body = self.send("gpt-6-sol", [user(".")]) fake._yerel_yanit.assert_not_called() self.assertEqual(fake._passthrough.call_args.args[2], body) 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_resolved_voice_target_and_codex_instructions_reach_the_stream_guard(self): self.env["ROUTES"]["aiturk-ide"] = { "url": "https://example.invalid/v1/responses", "key": "test-only", "flatten_tools": True} self.env["MODEL_MAP"]["ide/voice-alias"] = ("aiturk-ide", "Sesli-Hizli-Zeki", None) req = {"model": "ide/voice-alias", "instructions": "Codex system instructions unchanged", "input": [developer("Codex developer instructions unchanged"), user("selam")]} for failure in (None, urllib.error.URLError(ConnectionResetError("reset")), urllib.error.HTTPError("https://example.invalid", 400, "bad", {}, io.BytesIO(b"bad"))): with self.subTest(failure=type(failure).__name__): opener = self.env["urllib"].request.urlopen opener.reset_mock() opener.side_effect = failure opener.return_value = Mock(status=200) fake = Sahte() fake._denetimli_akis = Mock() with patch.object(self.env["jev_denetim"], "anahtar", return_value="test-only"): fake._responses(1, "/responses", json.dumps(req).encode()) fake._denetimli_akis.assert_called_once() self.assertEqual(fake._denetimli_akis.call_args.kwargs, {"hedef": "Sesli-Hizli-Zeki"}) self.assertEqual(fake._denetimli_akis.call_args.args[5], list(self.env["jev_denetim"].KATMAN)) sent = json.loads(opener.call_args.args[0].data) self.assertEqual(sent["model"], "Sesli-Hizli-Zeki") self.assertEqual(sent["instructions"], req["instructions"]) self.assertEqual(sent["input"][0], req["input"][0]) def test_nonvoice_json_still_reaches_original_jev_with_upstream_identity(self): self.env["ROUTES"]["aiturk-ide"] = { "url": "https://example.invalid/v1/responses", "key": "test-only", "flatten_tools": True} self.env["MODEL_MAP"]["ide/scope"] = ("aiturk-ide", "Other-Model", None) response = Mock(status=200, headers={"Content-Type": "application/json"}) self.env["urllib"].request.urlopen.return_value = response watch = self.env["WatchResponse"] = Mock(side_effect=AssertionError("watch scope escaped")) fake = Sahte(); fake._denetimli_akis = Mock() with patch.object(self.env["jev_denetim"], "anahtar", return_value="test-only"): fake._responses(1, "/responses", json.dumps({"model":"ide/scope", "input":[user("task")]}).encode()) fake._denetimli_akis.assert_called_once() self.assertIs(fake._denetimli_akis.call_args.args[0], response) self.assertEqual(fake._denetimli_akis.call_args.kwargs, {"hedef":"Other-Model"}) response.read.assert_not_called() watch.assert_not_called(); fake._send_json.assert_not_called(); fake._stream.assert_not_called() 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 = {"WATCH_OWNER": None, "WatchResponse": lambda response, accept: response, "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()