aiturk-hermes-ide/apps/desktop/assets/codex-bridge/test_combo_tekrar.py

"""A 9Router combo (Fusion-*) must not be sent twice once it may be running upstream.

22 Sep 2026: after a Cloudflare 524 the bridge resent 9r/Fusion-Orta three times and the whole
Fusion ran (and billed) again. Failures before the work starts may still be retried.
"""
import ast
import io
import json
from pathlib import Path
import types
import unittest
from unittest.mock import Mock
import urllib.error
import urllib.request

source = ast.parse(Path(__file__).with_name("kopru.py").read_text(encoding="utf-8"))
pick = lambda name: next(n for n in source.body if isinstance(n, ast.FunctionDef) and n.name == name)
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")


def http_error(code, body=b"{}"):
    return urllib.error.HTTPError("https://example.invalid", code, "x", {}, io.BytesIO(body))


class ComboNotResent(unittest.TestCase):
    def setUp(self):
        self.upstream = Mock()
        self.env = {
            "json": json, "re": __import__("re"), "os": __import__("os"), "http": __import__("http.client"),
            "time": types.SimpleNamespace(sleep=lambda s: None),
            "urllib": types.SimpleNamespace(
                request=types.SimpleNamespace(Request=urllib.request.Request, urlopen=self.upstream),
                error=urllib.error),
            "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": False}},
            "fix_tool_ids": lambda req: 0, "drop_foreign_items": lambda req: 0,
            "SseRewriter": Mock(), "log": lambda message: None,
            "DEBUG": False, "UA": "test", "RETRIES": 6, "UPSTREAM_IDLE_TIMEOUT": 3600,
            "NO_RETRY_PATTERNS": [], "pencere": __import__("pencere"), "IDE_BILGI": {}, "yarim": __import__("yarim"), "jev_denetim": __import__("jev_denetim"), "internet": __import__("internet"), "is_hatti": __import__("is_hatti"), "GORULEN_ISTEMCI": set(), "arac_semalarini_duzelt": lambda req: 0, "nokta_mesaji_mi": lambda req: False, "EXTRA_MODELS": [],
        }
        exec(compile(ast.Module(body=[pick("ek_model_mu"), pick("yonlendirilen"), pick("flatten_request"), pick("claude_sonu_duzelt"),
                                      pick("is_timeout"), pick("kopma_mi"), responses],
                                type_ignores=[]), "bridge.py", "exec"), self.env)

    def send(self, model):
        body = json.dumps({"model": model, "input": [{"role": "user", "content": [{"type": "input_text", "text": "8 vezir"}]}]}).encode()
        fake = Mock(headers={})
        self.env["_responses"](fake, 1, "/responses", body)
        return fake

    def test_combo_timeout_is_not_resent(self):
        self.upstream.side_effect = urllib.error.URLError(TimeoutError("timed out"))
        fake = self.send("9r/Fusion-Orta")
        self.assertEqual(self.upstream.call_count, 1)
        fake._stream.assert_not_called()

    def test_combo_504_and_524_are_not_resent(self):
        for code in (504, 524):
            with self.subTest(code=code):
                self.upstream.reset_mock()
                self.upstream.side_effect = http_error(code)
                self.send("9r/Fusion-Agir")
                self.assertEqual(self.upstream.call_count, 1)

    def test_combo_cut_mid_stream_is_not_resent(self):
        self.upstream.side_effect = None
        self.upstream.return_value = Mock(status=200)
        fake = self.send("9r/Fusion-Orta")
        fake._stream.side_effect = TimeoutError("read timed out")
        self.upstream.reset_mock()
        with self.assertRaises(TimeoutError):
            self.send_with(fake, "9r/Fusion-Orta")
        self.assertEqual(self.upstream.call_count, 1)

    def send_with(self, fake, model):
        body = json.dumps({"model": model, "input": [{"role": "user", "content": [{"type": "input_text", "text": "8 vezir"}]}]}).encode()
        self.env["_responses"](fake, 1, "/responses", body)

    def test_combo_failure_before_work_starts_is_retried(self):
        self.upstream.side_effect = [http_error(503), urllib.error.URLError(ConnectionRefusedError("refused")), Mock(status=200)]
        fake = self.send("9r/Fusion-Orta")
        self.assertEqual(self.upstream.call_count, 3)
        fake._stream.assert_called_once()

    def test_combo_closed_by_the_peer_before_headers_is_not_resent(self):
        import http.client
        for error in (http.client.RemoteDisconnected("closed"), urllib.error.URLError(ConnectionResetError("reset"))):
            with self.subTest(error=error):
                self.upstream.reset_mock()
                self.upstream.side_effect = error
                fake = self.send("9r/Fusion-Orta")
                self.assertEqual(self.upstream.call_count, 1)
                fake._stream.assert_not_called()

    def test_single_model_timeout_is_still_retried(self):
        self.upstream.side_effect = [urllib.error.URLError(TimeoutError("timed out")), Mock(status=200)]
        fake = self.send("9r/cc/claude-opus-5-5")
        self.assertEqual(self.upstream.call_count, 2)
        fake._stream.assert_called_once()


if __name__ == "__main__":
    unittest.main()