yilsemandClaude Opus 5.5
be14cf2923
Codex Baglama rejected every Responses WebSocket with 426, so Codex's own subscription models fell back to HTTP and re-sent the whole conversation on every step. The splitter now tunnels the WebSocket of Codex's own models to chatgpt.com byte for byte (model known from x-codex-routing-hint). Added models, remembered transfers and transfer history in a message still get 426 / a close frame and stay on the existing HTTP path, so added-model and transition handling is unchanged. bridge.json "abonelik_websocket": false restores the old refusal. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01XeNgWNzPjzUKQuS32RUYXx
389 lines
19 KiB
Python
389 lines
19 KiB
Python
"""The splitter (bridge.py): Codex's own models go to chatgpt.com byte for byte, added models to the bridge process.
|
|
|
|
Real sockets, no network: a fake chatgpt.com and a fake bridge process record what reaches them."""
|
|
import gzip
|
|
import hashlib
|
|
import http.client
|
|
import http.server
|
|
import json
|
|
import socket
|
|
import struct
|
|
import tempfile
|
|
import threading
|
|
import time
|
|
import unittest
|
|
import zlib
|
|
|
|
import bridge
|
|
|
|
|
|
class Kayit(http.server.BaseHTTPRequestHandler):
|
|
"""Records every request; answers from the server's `cevap` function."""
|
|
protocol_version = "HTTP/1.1"
|
|
|
|
def log_message(self, *a):
|
|
pass
|
|
|
|
def _her(self):
|
|
n = int(self.headers.get("Content-Length") or 0)
|
|
govde = self.rfile.read(n) if n else b""
|
|
self.server.gelen.append({"yontem": self.command, "yol": self.path, "baslik": dict(self.headers), "govde": govde})
|
|
self.server.cevap(self, govde)
|
|
|
|
do_GET = do_POST = _her
|
|
|
|
|
|
def sunucu(cevap):
|
|
s = http.server.ThreadingHTTPServer(("127.0.0.1", 0), Kayit)
|
|
s.daemon_threads = True
|
|
s.gelen, s.cevap = [], cevap
|
|
threading.Thread(target=s.serve_forever, daemon=True).start()
|
|
return s
|
|
|
|
|
|
def duz(durum, govde, tur="application/json"):
|
|
def f(h, _):
|
|
h.send_response(durum)
|
|
h.send_header("Content-Type", tur)
|
|
h.send_header("Content-Length", str(len(govde)))
|
|
h.end_headers()
|
|
h.wfile.write(govde)
|
|
return f
|
|
|
|
|
|
def zstd(veri):
|
|
try:
|
|
from compression import zstd as z
|
|
return z.compress(veri)
|
|
except ImportError:
|
|
import zstandard
|
|
return zstandard.ZstdCompressor().compress(veri)
|
|
|
|
|
|
def ws_cerceve(veri, op=1, rsv1=False, maske=b"\x01\x02\x03\x04"):
|
|
"""Istemci (Codex) cercevesi: maskeli."""
|
|
b0 = 0x80 | (0x40 if rsv1 else 0) | op
|
|
n = len(veri)
|
|
bas = bytes([b0, 0x80 | n]) if n < 126 else bytes([b0, 0x80 | 126]) + struct.pack(">H", n) if n < 65536 \
|
|
else bytes([b0, 0x80 | 127]) + struct.pack(">Q", n)
|
|
return bas + maske + bytes(c ^ maske[i % 4] for i, c in enumerate(veri))
|
|
|
|
|
|
def sikistir(sik, veri):
|
|
return (sik.compress(veri) + sik.flush(zlib.Z_SYNC_FLUSH))[:-4]
|
|
|
|
|
|
class SahteWs:
|
|
"""Sahte chatgpt.com WebSocket ucu: el sikismayi ve gelen ham cerceveleri kaydeder, her mesaja ozetle cevap verir."""
|
|
|
|
def __init__(self, durum=101, uzanti=""):
|
|
self.durum, self.uzanti, self.ham, self.baslik, self.baglanti = durum, uzanti, [], {}, 0
|
|
self.dinle = socket.socket()
|
|
self.dinle.bind(("127.0.0.1", 0))
|
|
self.dinle.listen(8)
|
|
self.port = self.dinle.getsockname()[1]
|
|
threading.Thread(target=self._kabul, daemon=True).start()
|
|
|
|
def _kabul(self):
|
|
while True:
|
|
try:
|
|
s, _ = self.dinle.accept()
|
|
except OSError:
|
|
return
|
|
self.baglanti += 1
|
|
threading.Thread(target=self._bag, args=(s,), daemon=True).start()
|
|
|
|
def _bag(self, s):
|
|
f = s.makefile("rb")
|
|
satirlar = []
|
|
while True:
|
|
satir = f.readline().decode("latin-1").rstrip("\r\n")
|
|
if not satir:
|
|
break
|
|
satirlar.append(satir)
|
|
for satir in satirlar[1:]:
|
|
k, _, v = satir.partition(":")
|
|
self.baslik[k.strip().lower()] = v.strip()
|
|
if self.durum != 101:
|
|
s.sendall(b"HTTP/1.1 %d Hayir\r\nContent-Length: 0\r\n\r\n" % self.durum)
|
|
s.close()
|
|
return
|
|
cevap = "HTTP/1.1 101 Switching Protocols\r\nUpgrade: websocket\r\nConnection: Upgrade\r\n"
|
|
if self.uzanti:
|
|
cevap += f"Sec-WebSocket-Extensions: {self.uzanti}\r\n"
|
|
s.sendall((cevap + "\r\n").encode())
|
|
ac = zlib.decompressobj(-zlib.MAX_WBITS)
|
|
while True:
|
|
c = bridge.ws_cerceve_oku(f.read)
|
|
if c is None or c[3] == 8:
|
|
break
|
|
self.ham.append(c[0])
|
|
veri = ac.decompress(c[4] + b"\x00\x00\xff\xff") if c[2] else c[4]
|
|
yuk = b"yanit:" + hashlib.sha256(veri).hexdigest().encode()
|
|
s.sendall(bytes([0x81, len(yuk)]) + yuk)
|
|
s.close()
|
|
|
|
|
|
class Ayirici(unittest.TestCase):
|
|
def setUp(self):
|
|
self.ozgun = sunucu(duz(200, b'event: response.completed\ndata: {"ok":1}\n\n', "text/event-stream"))
|
|
def worker(h, body):
|
|
if h.path.endswith("/prepare-transfer"):
|
|
return duz(200, json.dumps({"request": json.loads(body)["request"]}).encode())(h, body)
|
|
return duz(200, b'event: response.completed\ndata: {"kopru":1}\n\n', "text/event-stream")(h, body)
|
|
self.kopru = sunucu(worker)
|
|
self.on = bridge.Sunucu(("127.0.0.1", 0), bridge.H)
|
|
threading.Thread(target=self.on.serve_forever, daemon=True).start()
|
|
self.eski = (bridge.CHATGPT, bridge.KOPRU, bridge.LOGDIR, dict(bridge.CFG))
|
|
self.gecici = tempfile.TemporaryDirectory()
|
|
self.transfer_old = bridge.TRANSFER_PATH, dict(bridge.TRANSFERS)
|
|
bridge.TRANSFER_PATH = self.gecici.name + "/transfers.json"
|
|
bridge.TRANSFERS.clear()
|
|
bridge.WS_HTTP.clear()
|
|
bridge.CHATGPT = f"http://127.0.0.1:{self.ozgun.server_address[1]}"
|
|
bridge.KOPRU = f"http://127.0.0.1:{self.kopru.server_address[1]}"
|
|
bridge.LOGDIR = self.gecici.name
|
|
bridge.CFG.clear()
|
|
|
|
def tearDown(self):
|
|
for s in (self.on, self.ozgun, self.kopru):
|
|
s.shutdown()
|
|
s.server_close()
|
|
bridge.CHATGPT, bridge.KOPRU, bridge.LOGDIR, cfg = self.eski
|
|
bridge.CFG.clear()
|
|
bridge.CFG.update(cfg)
|
|
bridge.TRANSFER_PATH, transfers = self.transfer_old
|
|
bridge.TRANSFERS.clear()
|
|
bridge.TRANSFERS.update(transfers)
|
|
self.gecici.cleanup()
|
|
|
|
def iste(self, yontem, yol, govde=None, baslik=None):
|
|
b = http.client.HTTPConnection("127.0.0.1", self.on.server_address[1], timeout=10)
|
|
b.request(yontem, yol, body=govde, headers=baslik or {})
|
|
y = b.getresponse()
|
|
veri = y.read()
|
|
b.close()
|
|
return y.status, veri, dict(y.getheaders())
|
|
|
|
def test_codex_own_model_reaches_chatgpt_byte_for_byte(self):
|
|
"""Compressed or not, with a command, a lone dot or a compaction trigger in it: nothing is changed."""
|
|
istekler = [{"model": "gpt-6-astra", "input": [{"role": "user", "content": "modeli opus 5 5 yap"}]},
|
|
{"model": "gpt-6-luna", "input": [{"role": "user", "content": "."}]},
|
|
{"model": "gpt-6-astra", "prompt_cache_key": "s", "input": [{"type": "compaction_trigger"}]},
|
|
{"model": "gpt-6-astra", "input": [{"type": "compaction", "encrypted_content": "gAAAA_native"},
|
|
{"type": "reasoning", "id": "rs_native"}]}]
|
|
for veri in istekler:
|
|
ham = json.dumps(veri, ensure_ascii=False).encode("utf-8")
|
|
for kodlama, govde in (("zstd", zstd(ham)), ("gzip", gzip.compress(ham)), ("", ham)):
|
|
with self.subTest(model=veri["model"], kodlama=kodlama):
|
|
baslik = {"Authorization": "Bearer sahte", "originator": "Codex Desktop", "chatgpt-account-id": "h1"}
|
|
if kodlama:
|
|
baslik["Content-Encoding"] = kodlama
|
|
durum, cevap, _ = self.iste("POST", "/backend-api/codex/responses", govde, baslik)
|
|
giden = self.ozgun.gelen[-1]
|
|
self.assertEqual(hashlib.sha256(giden["govde"]).hexdigest(), hashlib.sha256(govde).hexdigest())
|
|
basliklar = {k.lower(): v for k, v in giden["baslik"].items()}
|
|
self.assertEqual(basliklar.get("content-encoding", ""), kodlama)
|
|
self.assertEqual((basliklar["authorization"], basliklar["originator"], basliklar["chatgpt-account-id"]),
|
|
("Bearer sahte", "Codex Desktop", "h1"))
|
|
self.assertEqual(giden["yol"], "/backend-api/codex/responses")
|
|
self.assertEqual((durum, cevap), (200, b'event: response.completed\ndata: {"ok":1}\n\n'))
|
|
self.assertEqual(self.kopru.gelen, []) # the bridge process never saw any of them
|
|
|
|
def test_added_models_go_to_the_bridge_process_and_only_there(self):
|
|
for model in ("ide/Opus-5", "9r/cc/claude-opus-5-5"):
|
|
govde = zstd(json.dumps({"model": model, "input": []}).encode())
|
|
durum, cevap, _ = self.iste("POST", "/backend-api/codex/responses", govde,
|
|
{"Content-Encoding": "zstd", "originator": "Codex Desktop"})
|
|
self.assertEqual((durum, cevap), (200, b'event: response.completed\ndata: {"kopru":1}\n\n'))
|
|
self.assertEqual(self.kopru.gelen[-1]["govde"], govde)
|
|
basliklar = {k.lower(): v for k, v in self.kopru.gelen[-1]["baslik"].items()}
|
|
self.assertEqual((basliklar["originator"], basliklar["content-encoding"]), ("Codex Desktop", "zstd"))
|
|
bridge.CFG["model_map"] = {"ozel-model": ["rota", "hedef"]}
|
|
self.iste("POST", "/backend-api/codex/responses", json.dumps({"model": "ozel-model"}).encode())
|
|
self.assertEqual(len([x for x in self.kopru.gelen if x["yol"].endswith("/responses")]), 3)
|
|
self.assertEqual(self.ozgun.gelen, [])
|
|
|
|
def test_a_body_that_cannot_be_read_takes_codex_own_path_untouched(self):
|
|
for govde, baslik in ((b"\x00\x01bozuk", {"Content-Encoding": "zstd"}), (b"json degil", {}), (b"[1,2]", {})):
|
|
self.iste("POST", "/backend-api/codex/responses", govde, baslik)
|
|
self.assertEqual(self.ozgun.gelen[-1]["govde"], govde)
|
|
self.assertEqual(self.kopru.gelen, [])
|
|
|
|
def test_codex_own_model_keeps_working_while_the_bridge_process_is_down(self):
|
|
self.kopru.shutdown()
|
|
self.kopru.server_close()
|
|
govde = json.dumps({"model": "gpt-6-astra", "input": []}).encode()
|
|
self.assertEqual(self.iste("POST", "/backend-api/codex/responses", govde)[0], 200)
|
|
# The model list falls back to chatgpt.com's own list, unchanged.
|
|
self.ozgun.cevap = duz(200, b'{"models":[{"slug":"gpt-6-astra"}]}')
|
|
durum, liste, _ = self.iste("GET", "/backend-api/codex/models?client_version=1", None, {"If-None-Match": "x"})
|
|
self.assertEqual((durum, liste), (200, b'{"models":[{"slug":"gpt-6-astra"}]}'))
|
|
# An added model says so instead of hanging or going to chatgpt.com under another name.
|
|
durum, cevap, _ = self.iste("POST", "/backend-api/codex/responses", json.dumps({"model": "ide/Opus-5"}).encode())
|
|
self.assertEqual(durum, 200)
|
|
self.assertIn(b"response.failed", cevap)
|
|
self.assertEqual([g["yol"].split("?")[0] for g in self.ozgun.gelen],
|
|
["/backend-api/codex/responses", "/backend-api/codex/models"])
|
|
# Status still answers in the shape AITURK IDE expects.
|
|
durum, cevap, _ = self.iste("GET", "/aiturk/codex-baglama/status")
|
|
d = json.loads(cevap)
|
|
self.assertEqual((durum, d["surum"], d["kopru"], d["ayirici"]), (200, bridge.SURUM, "kapali", True))
|
|
|
|
def test_model_list_comes_from_the_bridge_and_falls_back_when_it_fails(self):
|
|
self.kopru.cevap = duz(200, b'{"models":[{"slug":"gpt-6-astra"},{"slug":"ide/Opus-5"}]}')
|
|
self.assertIn(b"ide/Opus-5", self.iste("GET", "/backend-api/codex/models")[1])
|
|
self.assertEqual(self.ozgun.gelen, [])
|
|
self.kopru.cevap = duz(500, b'{"error":"x"}')
|
|
self.ozgun.cevap = duz(200, b'{"models":[{"slug":"gpt-6-astra"}]}')
|
|
self.assertEqual(self.iste("GET", "/backend-api/codex/models")[1], b'{"models":[{"slug":"gpt-6-astra"}]}')
|
|
|
|
def test_everything_else_and_errors_pass_through_as_they_are(self):
|
|
self.ozgun.cevap = duz(200, b'{"sonuc":1}')
|
|
self.assertEqual(self.iste("POST", "/backend-api/codex/alpha/search", b'{"q":"x"}')[1], b'{"sonuc":1}')
|
|
self.assertEqual(self.ozgun.gelen[-1]["govde"], b'{"q":"x"}')
|
|
self.ozgun.cevap = duz(400, b'{"error":{"message":"gecersiz"}}')
|
|
durum, cevap, _ = self.iste("POST", "/backend-api/codex/responses", json.dumps({"model": "gpt-6-astra"}).encode())
|
|
self.assertEqual((durum, cevap), (400, b'{"error":{"message":"gecersiz"}}'))
|
|
self.assertEqual(self.kopru.gelen, [])
|
|
|
|
def test_a_long_answer_streams_piece_by_piece_without_waiting_for_the_end(self):
|
|
"""This morning's failure was an idle stream: Codex got no bytes for five minutes. Bytes must flow as they come."""
|
|
def yavas(h, _):
|
|
h.send_response(200)
|
|
h.send_header("Content-Type", "text/event-stream")
|
|
h.send_header("Transfer-Encoding", "chunked")
|
|
h.end_headers()
|
|
for parca in (b"data: bir\n\n", b"data: iki\n\n"):
|
|
h.wfile.write(b"%x\r\n" % len(parca) + parca + b"\r\n")
|
|
h.wfile.flush()
|
|
time.sleep(1.0)
|
|
h.wfile.write(b"0\r\n\r\n")
|
|
self.ozgun.cevap = yavas
|
|
b = http.client.HTTPConnection("127.0.0.1", self.on.server_address[1], timeout=10)
|
|
bas = time.time()
|
|
b.request("POST", "/backend-api/codex/responses", body=json.dumps({"model": "gpt-6-astra"}).encode())
|
|
y = b.getresponse()
|
|
ilk = y.read1(64)
|
|
ilk_sure = time.time() - bas
|
|
kalan = y.read()
|
|
b.close()
|
|
self.assertEqual(ilk, b"data: bir\n\n")
|
|
self.assertLess(ilk_sure, 0.8) # the first piece arrived before the answer ended
|
|
self.assertEqual(kalan, b"data: iki\n\n")
|
|
|
|
def test_status_is_the_bridge_status_plus_the_splitter_fields(self):
|
|
self.kopru.cevap = duz(200, b'{"ok": true, "surum": "8", "kod": "abc", "ide_durum": "tamam", "ide_model": 147}')
|
|
d = json.loads(self.iste("GET", "/aiturk/codex-baglama/status")[1])
|
|
self.assertEqual((d["surum"], d["ide_model"], d["kopru"], d["ayirici"]), (bridge.SURUM, 147, "calisiyor", True))
|
|
|
|
def test_a_persistent_connection_without_a_model_hint_is_refused_the_way_it_is_today(self):
|
|
durum, _, _ = self.iste("GET", "/backend-api/codex/responses", None, {"Upgrade": "websocket", "Connection": "Upgrade"})
|
|
self.assertEqual(durum, 426)
|
|
self.assertEqual(self.ozgun.gelen + self.kopru.gelen, [])
|
|
|
|
# --- Surekli baglanti: abonelik modeli chatgpt.com'a tunellenir, ek model/gecis HTTP'de kalir ---------------
|
|
def ws_ust(self, durum=101, uzanti=""):
|
|
ust = SahteWs(durum, uzanti)
|
|
bridge.CHATGPT = f"http://127.0.0.1:{ust.port}"
|
|
return ust
|
|
|
|
def ws_ac(self, model="gpt-6.1-sol", sid="sohbet-1", uzanti=True):
|
|
k = socket.create_connection(("127.0.0.1", self.on.server_address[1]), timeout=10)
|
|
baslik = ["GET /backend-api/codex/responses HTTP/1.1", "Host: 127.0.0.1", "Connection: Upgrade",
|
|
"Upgrade: websocket", "Sec-WebSocket-Version: 13", "Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==",
|
|
"authorization: Bearer sahte", "chatgpt-account-id: hesap", f"session-id: {sid}", f"thread-id: {sid}",
|
|
"openai-beta: responses_websockets=2026-02-06"]
|
|
if model:
|
|
baslik.append(f"x-codex-routing-hint: model={model}")
|
|
if uzanti:
|
|
baslik.append("sec-websocket-extensions: permessage-deflate; client_max_window_bits")
|
|
k.sendall(("\r\n".join(baslik) + "\r\n\r\n").encode())
|
|
cevap = b""
|
|
while b"\r\n\r\n" not in cevap:
|
|
parca = k.recv(4096)
|
|
if not parca:
|
|
break
|
|
cevap += parca
|
|
return k, int(cevap.split(b" ", 2)[1]), cevap.partition(b"\r\n\r\n")[2]
|
|
|
|
def test_codex_own_model_websocket_is_tunnelled_to_chatgpt_byte_for_byte(self):
|
|
ust = self.ws_ust()
|
|
k, durum, _ = self.ws_ac()
|
|
self.assertEqual(durum, 101)
|
|
mesaj = json.dumps({"type": "response.create", "model": "gpt-6.1-sol", "input": [{"type": "message", "role": "user",
|
|
"content": [{"type": "input_text", "text": "selam " * 5000}]}]}).encode()
|
|
cerceve = ws_cerceve(mesaj)
|
|
k.sendall(cerceve)
|
|
cevap = bridge.ws_cerceve_oku(k.makefile("rb").read)
|
|
self.assertEqual(cevap[4], b"yanit:" + hashlib.sha256(mesaj).hexdigest().encode())
|
|
self.assertEqual(ust.ham[0], cerceve) # Codex'in cercevesi aynen gitti
|
|
self.assertEqual(ust.baslik["authorization"], "Bearer sahte")
|
|
self.assertEqual(ust.baslik["openai-beta"], "responses_websockets=2026-02-06")
|
|
self.assertEqual(self.kopru.gelen, [])
|
|
k.close()
|
|
|
|
def test_added_model_websocket_falls_back_to_http_at_once(self):
|
|
ust = self.ws_ust()
|
|
k, durum, _ = self.ws_ac(model="ide/claude-claude-haiku-5-5")
|
|
self.assertEqual(durum, 426)
|
|
self.assertEqual(ust.baglanti, 0)
|
|
k.close()
|
|
|
|
def test_switching_to_an_added_model_on_an_open_tunnel_moves_the_chat_to_http(self):
|
|
ust = self.ws_ust()
|
|
k, durum, _ = self.ws_ac(sid="sohbet-gecis")
|
|
self.assertEqual(durum, 101)
|
|
k.sendall(ws_cerceve(json.dumps({"type": "response.create", "model": "ide/agy-gemini-3.8-flash-low", "input": []}).encode()))
|
|
kapanis = bridge.ws_cerceve_oku(k.makefile("rb").read)
|
|
self.assertEqual(kapanis[3], 8) # kapanis cercevesi
|
|
self.assertEqual(ust.ham, []) # ek model istegi chatgpt.com'a gitmedi
|
|
k.close()
|
|
k2, durum2, _ = self.ws_ac(sid="sohbet-gecis")
|
|
self.assertEqual(durum2, 426) # ayni sohbet artik HTTP yolunda
|
|
k2.close()
|
|
|
|
def test_compressed_message_with_transfer_history_is_detected(self):
|
|
ust = self.ws_ust(uzanti="permessage-deflate")
|
|
k, durum, _ = self.ws_ac(sid="sohbet-sikisik")
|
|
self.assertEqual(durum, 101)
|
|
sik = zlib.compressobj(wbits=-zlib.MAX_WBITS)
|
|
ilk = json.dumps({"type": "response.create", "model": "gpt-6.1-sol", "input": [{"type": "message", "role": "user",
|
|
"content": [{"type": "input_text", "text": "ilk"}]}]}).encode()
|
|
k.sendall(ws_cerceve(sikistir(sik, ilk), rsv1=True))
|
|
okur = k.makefile("rb").read
|
|
self.assertEqual(bridge.ws_cerceve_oku(okur)[3], 1)
|
|
ikinci = json.dumps({"type": "response.create", "model": "gpt-6.1-sol",
|
|
"input": [{"type": "message", "role": "assistant", "id": "chatcmpl-abc", "content": []}]}).encode()
|
|
k.sendall(ws_cerceve(sikistir(sik, ikinci), rsv1=True))
|
|
self.assertEqual(bridge.ws_cerceve_oku(okur)[3], 8)
|
|
self.assertEqual(len(ust.ham), 1)
|
|
k.close()
|
|
|
|
def test_chat_with_a_remembered_transfer_stays_on_http(self):
|
|
self.ws_ust()
|
|
scope = hashlib.sha256(b"hesap|sohbet-eski").hexdigest()
|
|
bridge.TRANSFERS[scope] = "ide/Jail-Omni"
|
|
k, durum, _ = self.ws_ac(sid="sohbet-eski")
|
|
self.assertEqual(durum, 426)
|
|
k.close()
|
|
|
|
def test_chatgpt_refusing_the_tunnel_or_setting_off_falls_back_to_http(self):
|
|
ust = self.ws_ust(durum=401)
|
|
k, durum, _ = self.ws_ac(sid="sohbet-red")
|
|
self.assertEqual((durum, ust.baglanti), (426, 1))
|
|
k.close()
|
|
bridge.CFG["abonelik_websocket"] = False
|
|
k, durum, _ = self.ws_ac(sid="sohbet-kapali")
|
|
self.assertEqual((durum, ust.baglanti), (426, 1))
|
|
k.close()
|
|
|
|
def test_which_side_a_model_is_on(self):
|
|
for slug, ek in (("ide/Opus-5", True), ("9r/ag/gemini", True), ("gpt-6-astra", False), ("gpt-6.1-sol", False),
|
|
("", False), (None, False), ("idea", False)):
|
|
self.assertEqual(bridge.ek_model_mu(slug), ek, slug)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|