Files
aiturk-hermes-ide/apps/desktop/assets/codex-bridge/test_ozet_kopru.py
T
yilsemandClaude Opus 5.5 1ce5ab82cf feat(codex-bridge): ship the local Codex bridge with the IDE
Adds the IDE's models and combos to Codex's model list and keeps conversation
summaries readable across providers, so switching models in Codex no longer
drops the conversation. Standard-library Python, run on the agent venv.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01HRDGW1LiKcDUZYUdKBoWYW
2026-10-09 15:46:10 +03:00

146 lines
7.8 KiB
Python

"""ozet_kopru testleri: 9Router icin sikistirma bagdastiricisi ve saglayicilar arasi ozet aktarimi."""
import json
import os
import tempfile
import unittest
import ozet_kopru as o
def msg(rol, metin):
return {"type": "message", "role": rol, "content": [{"type": "input_text", "text": metin}]}
def sse(*olaylar):
return b"".join(b"event: " + e["type"].encode() + b"\ndata: " + json.dumps(e, ensure_ascii=False).encode() + b"\n\n" for e in olaylar)
def coz(veri):
return [json.loads(s[5:]) for blok in veri.split(b"\n\n") for s in blok.split(b"\n") if s.startswith(b"data:")]
OPENAI_OZETI = {"type": "compaction", "id": "cmp_0123", "encrypted_content": "gAAAAABsahte", "internal_chat_message_metadata_passthrough": {"turn_id": "t"}}
class Istek(unittest.TestCase):
def test_tetik_algilanir_ve_ozet_istegine_cevrilir(self):
req = {"model": "m", "tools": [{"type": "function", "name": "x"}], "input": [msg("user", "KARAR-1: 14 Kasim"), {"type": "compaction_trigger"}]}
self.assertTrue(o.tetik_mi(req))
self.assertFalse(o.tetik_mi({"input": [msg("user", "normal")]}))
self.assertFalse(o.tetik_mi({"input": []}))
o.ozet_istegi_hazirla(req)
self.assertNotIn("compaction_trigger", json.dumps(req))
self.assertEqual(req["input"][0]["content"][0]["text"], "KARAR-1: 14 Kasim") # gecmis aynen kalir
self.assertEqual(req["input"][-1]["role"], "user")
istem = req["input"][-1]["content"][0]["text"]
for baslik in ("KARARLAR VE GEREKCELERI", "ACIK ISLER", "ONEMLI SONUCLAR", "KULLANICININ ISTEGI"):
self.assertIn(baslik, istem)
self.assertEqual(req["tool_choice"], "none")
def test_kopru_ozeti_her_hedefte_duz_metne_acilir(self):
oge = o.ozet_ogesi("KARAR-1: teslim 14 Kasim\nACIK-IS-1: fatura onayi")
self.assertTrue(oge["id"].startswith("cmp_astrabridge_"))
self.assertNotIn("KARAR-1", json.dumps(oge)) # ogede duz metin gorunmez (base64)
for yabanci in (False, True):
req = {"input": [msg("user", "once"), dict(oge), msg("user", "soru")]}
b = o.ozetleri_ac(req, yabanci_hedef=yabanci)
self.assertEqual(b["acilan"], 1)
metin = req["input"][1]["content"][0]["text"]
self.assertEqual(req["input"][1]["role"], "user")
self.assertIn("KARAR-1: teslim 14 Kasim", metin)
self.assertTrue(metin.startswith(o.OZET_BASLIK))
self.assertNotIn('"compaction"', json.dumps(req))
def test_openai_ozeti_openai_hedefinde_dokunulmaz(self):
req = {"input": [dict(OPENAI_OZETI), msg("user", "soru")]}
once = json.dumps(req, sort_keys=True)
b = o.ozetleri_ac(req, yabanci_hedef=False)
self.assertEqual(json.dumps(req, sort_keys=True), once)
self.assertEqual(b, {"acilan": 0, "cozulen": 0, "okunamayan": 0})
def test_openai_ozeti_yabanci_hedefte_duz_metne_cevrilir_ya_da_not_dusulur(self):
req = {"input": [dict(OPENAI_OZETI), msg("user", "soru")]}
b = o.ozetleri_ac(req, yabanci_hedef=True, coz=lambda oge: "KARAR-2: saglayici degismeyecek")
self.assertEqual(b["cozulen"], 1)
self.assertIn("KARAR-2", req["input"][0]["content"][0]["text"])
for bozuk in (None, lambda oge: None, lambda oge: 1 / 0):
req = {"input": [dict(OPENAI_OZETI), msg("user", "soru")]}
b = o.ozetleri_ac(req, yabanci_hedef=True, coz=bozuk)
self.assertEqual(b["okunamayan"], 1)
self.assertEqual(req["input"][0]["content"][0]["text"], o.OKUNAMADI_NOTU) # sessiz kayip yok
self.assertNotIn("gAAAAAB", json.dumps(req))
def test_coz_istegi_yalniz_gerekli_alanlari_tasir(self):
g = o.coz_istegi(OPENAI_OZETI, "gpt-6-luna")
self.assertEqual(g["input"][0], {"type": "compaction", "id": "cmp_0123", "encrypted_content": "gAAAAABsahte"})
self.assertEqual(g["model"], "gpt-6-luna")
self.assertFalse(g["store"])
class Akis(unittest.TestCase):
def _mesaj(self, metin, kimlik="msg_1"):
return {"id": kimlik, "type": "message", "role": "assistant", "content": [{"type": "output_text", "text": metin}]}
def test_metin_tek_compaction_ogesine_sarilir(self):
ozet = "1. KULLANICININ ISTEGI: x\n2. KARARLAR: KARAR-1 14 Kasim"
ham = sse({"type": "response.created", "response": {"id": "r1"}},
{"type": "response.output_item.added", "item": {"id": "msg_1", "type": "message"}},
{"type": "response.output_text.delta", "item_id": "msg_1", "delta": ozet[:10]},
{"type": "response.output_text.delta", "item_id": "msg_1", "delta": ozet[10:]},
{"type": "response.output_item.done", "item": self._mesaj(ozet)},
{"type": "response.completed", "response": {"id": "r1", "output": [self._mesaj(ozet)], "usage": {"input_tokens": 5}}})
a = o.OzetAkisi(tohum="7")
out = b"".join(a.feed(ham[i:i + 53]) for i in range(0, len(ham), 53)) + a.flush()
olaylar = coz(out)
self.assertEqual([e["type"] for e in olaylar], ["response.created", "response.output_item.done", "response.completed"])
oge = olaylar[1]["item"]
self.assertEqual(oge["type"], "compaction")
self.assertEqual(o.ozet_metni(oge), ozet)
self.assertEqual(olaylar[2]["response"]["output"], [oge]) # Codex tam bir compaction ogesi gorur
self.assertEqual(olaylar[2]["response"]["usage"], {"input_tokens": 5})
self.assertEqual(a.metin, ozet)
self.assertNotIn("KARAR-1", out.decode()) # ozet metni akista duz gorunmez
def test_yalniz_delta_gelirse_de_toplanir(self):
ham = sse({"type": "response.created", "response": {}},
{"type": "response.output_text.delta", "item_id": "a", "delta": "ozet "},
{"type": "response.output_text.delta", "item_id": "a", "delta": "metni"},
{"type": "response.completed", "response": {}})
a = o.OzetAkisi()
self.assertEqual(o.ozet_metni(coz(a.feed(ham))[1]["item"]), "ozet metni")
def test_metin_yoksa_compaction_uretilmez_hata_gorunur(self):
ham = sse({"type": "response.created", "response": {}},
{"type": "response.output_item.done", "item": {"type": "function_call", "name": "x"}},
{"type": "response.completed", "response": {"output": []}})
a = o.OzetAkisi()
olaylar = coz(a.feed(ham))
self.assertNotIn("compaction", json.dumps(olaylar))
self.assertFalse(a.metin)
hata = sse({"type": "response.created", "response": {}}, {"type": "response.failed", "response": {"error": {"message": "x"}}})
self.assertIn("response.failed", o.OzetAkisi().feed(hata).decode())
def test_onbellek(self):
klasor = os.path.join(tempfile.mkdtemp(), "c")
self.assertIsNone(o.onbellekten(klasor, OPENAI_OZETI))
o.onbellege(klasor, OPENAI_OZETI, "duz metin ozet")
self.assertEqual(o.onbellekten(klasor, OPENAI_OZETI), "duz metin ozet")
def test_sse_metni(self):
ham = sse({"type": "response.created", "response": {}}, {"type": "response.output_item.done", "item": self._mesaj("cozulmus ozet")},
{"type": "response.completed", "response": {}})
self.assertEqual(o.sse_metni(ham), "cozulmus ozet")
class Fren(unittest.TestCase):
def test_ayni_sohbette_ust_uste_ozet_durdurulur(self):
f = o.DonguFreni(sinir=3, pencere_sn=600)
self.assertEqual([f.izin("sohbet-a", t) for t in (0, 10, 20, 30, 40)], [True, True, True, False, False])
self.assertTrue(f.izin("sohbet-b", 40)) # baska sohbet etkilenmez
self.assertTrue(f.izin("sohbet-a", 700)) # pencere gecince yeniden izin
self.assertIn("Dongu durduruldu", o.DONGU_MESAJI % 3)
if __name__ == "__main__":
unittest.main()