File size: 2,514 Bytes
46e3de3 3f30086 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 | from pathlib import Path
import json
import urllib.error
import naldo_opencode_bridge as bridge
ROOT = Path(__file__).parents[1]
def test_naldo_embedding_proxy_retries_transient_503(monkeypatch):
calls = {"n": 0}
class Response:
def __enter__(self): return self
def __exit__(self, *args): return False
def read(self):
return json.dumps({"data": [{"embedding": [0.1]}]}).encode()
def fake_urlopen(*args, **kwargs):
calls["n"] += 1
if calls["n"] < 3:
raise urllib.error.HTTPError("http://embed", 503, "busy", {}, None)
return Response()
monkeypatch.setattr(bridge, "EMBED_RETRIES", 3)
monkeypatch.setattr(bridge, "EMBED_BACKOFF", 0.1)
monkeypatch.setattr(bridge.time, "sleep", lambda _: None)
monkeypatch.setattr(bridge.urllib.request, "urlopen", fake_urlopen)
result = bridge._forward_embeddings(b"{}")
assert calls["n"] == 3
assert result["data"][0]["embedding"] == [0.1]
def test_gemini_embedding_proxy_has_bounded_retry_and_threadpool():
source = (ROOT / "gemini_voyager_proxy.py").read_text(encoding="utf-8")
assert "GEMINI_EMBED_RETRIES" in source
assert "EMBED_RETRYABLE" in source
assert "run_in_threadpool(_forward_embeddings" in source
assert "time.sleep(min(EMBED_BACKOFF * (2 ** attempt), 8.0))" in source
def test_both_supervisors_gate_voyager_on_embedding_readiness():
naldo = (ROOT / "voyager_stack.py").read_text(encoding="utf-8")
gemini = (ROOT / "gemini_bot_stack.py").read_text(encoding="utf-8")
assert "Naldo embedding backend did not become ready" in naldo
assert "Gemini embedding backend did not become ready" in gemini
def test_altoclef_opencode_install_is_runtime_only_not_persisted():
source = (ROOT / "altoclef_stack.py").read_text(encoding="utf-8")
assert '"/tmp/altoclef-runtime"' in source
assert 'install = ROOT / "opencode-cli"' not in source
assert '"opencode.cmd" if os.name == "nt" else "opencode"' in source
def test_server_only_routes_embeddings_to_model_space_not_local_7860():
naldo = (ROOT / "voyager_stack.py").read_text(encoding="utf-8")
gemini = (ROOT / "gemini_bot_stack.py").read_text(encoding="utf-8")
assert "https://erinaldordso-mine.hf.space/v1" in naldo
assert 'env["NALDO_EMBED_BASE"] = EMBED_BASE_URL' in naldo
assert '"GEMINI_VOYAGER_EMBED_BASE": VOYAGER_EMBED_BASE_URL' in gemini
assert '"GEMINI_VOYAGER_EMBED_BASE": "http://127.0.0.1:7860/v1"' not in gemini
|