| 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 |
|
|
