File size: 4,723 Bytes
72cca28 7e83939 ef1d4f0 7e83939 ef1d4f0 7e83939 ef1d4f0 7e83939 051a00d 7e83939 72cca28 d8603a7 72cca28 7e83939 72cca28 7e83939 72cca28 7e83939 ef1d4f0 5ddf32a ef1d4f0 6ff2ff0 5ddf32a ef1d4f0 5ddf32a 6ff2ff0 5ddf32a 6ff2ff0 ef1d4f0 6ff2ff0 ef1d4f0 6ff2ff0 ef1d4f0 7e83939 ef1d4f0 7e83939 | 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 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 | """Gemini_Bot runs the same Voyager architecture with isolated state and credentials."""
import ast
import json
import os
from pathlib import Path
import subprocess
import sys
from tempfile import TemporaryDirectory
import threading
import time
from types import SimpleNamespace
from unittest.mock import patch
ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT))
import gemini_bot_stack as module
def test_isolation():
assert module.CLI_VERSION == "0.61.0"
assert module.MODEL == "gemini-3.5-flash-lite"
assert len(module.KEY_NAMES) == 6
with patch.dict(os.environ, {name: f"private-value-{i}" for i, name in enumerate(module.KEY_NAMES)}):
status = module.GeminiBotStack().status()
assert status["bot_username"] == "Gemini_Bot"
assert status["architecture"] == "Voyager"
assert status["same_voyager_runtime_as_naldo"] is False # no live runner attestation yet
assert status["engine_runtime"] is None
assert "gemini-voyager" in status["checkpoint_dir"].replace("\\", "/")
assert status["configured_key_count"] == 6
for name in module.KEY_NAMES:
assert name not in str(status)
assert "private-value" not in str(status)
def test_no_stored_credentials():
for name in ["gemini_bot.js", "gemini_bot_stack.py", "gemini_bot_logic.js", "gemini_voyager_proxy.py"]:
source = (ROOT / name).read_text()
assert "AIzaSy" not in source
assert "AQ.Ab8" not in source
stack_source = (ROOT / "gemini_bot_stack.py").read_text()
assert "voyager_runner.py" in stack_source
assert "gemini_voyager_proxy.py" in stack_source
assert '"gemini_bot.js"' not in stack_source
tree = ast.parse(stack_source)
assert tree is not None
def test_proxy_preserves_safe_failure_diagnostics():
source = ast.parse((ROOT / "gemini_voyager_proxy.py").read_text())
wanted = {"_call_cli", "_failure_category", "_select_available_key", "KeyStatus", "KeyState"}
nodes = [node for node in source.body
if (isinstance(node, ast.FunctionDef) and node.name in wanted)
or (isinstance(node, ast.ClassDef) and node.name in wanted)]
assert {n.name for n in nodes} == wanted, "proxy surface changed"
state = {"calls": 0, "failures": 0, "selected_key": 0, "last_elapsed_s": None,
"prompt_tokens": 0, "completion_tokens": 0, "last_error": None,
"last_failure": None, "last_failure_at": None, "last_failed_key_slot": None,
"failures_by_key_slot": [0, 0], "last_failure_category": None,
"failure_category_by_key_slot": [None, None]}
ns = {"RAW_KEYS": ["private-1", "private-2"], "LOCK": threading.Lock(),
"STATE_LOCK": threading.Lock(), "state": state, "os": os, "subprocess": subprocess,
"json": json, "time": time, "CLI": "gemini", "MODEL": "gemini-3.5-flash-lite",
"WORKSPACE": "/tmp", "HOME": "/tmp", "SYSTEM_MD": "/tmp/system.md",
"KEY_STATUS": [], "_usage": lambda stats: (12, 3)}
ns["Any"] = object
ns["dataclass"] = __import__("dataclasses").dataclass
ns["field"] = __import__("dataclasses").field
ns["Enum"] = __import__("enum").Enum
exec(compile(ast.Module(body=nodes, type_ignores=[]), "proxy.py", "exec"), ns)
ns["KEY_STATUS"] = [ns["KeyStatus"](key) for key in ns["RAW_KEYS"]]
responses = [SimpleNamespace(returncode=1, stdout='{"error":{"code":147,"type":"QuotaError"}}', stderr=""),
SimpleNamespace(returncode=0, stdout='{"response":"ok","stats":{}}', stderr="")]
with patch.object(subprocess, "run", side_effect=responses) as run:
assert ns["_call_cli"]("hello") == ("ok", 12, 3)
assert run.call_args_list[0].kwargs["env"]["GEMINI_API_KEY"] == "private-1"
assert run.call_args_list[1].kwargs["env"]["GEMINI_API_KEY"] == "private-2"
assert state["failures"] == 1 and state["last_error"] is None
assert state["last_failure"] == "Gemini CLI exit 1 API 147"
assert state["last_failed_key_slot"] == 1
assert state["failures_by_key_slot"] == [1, 0]
assert state["last_failure_category"] == "quota_or_rate_limit"
assert state["failure_category_by_key_slot"] == ["quota_or_rate_limit", None]
assert ns["_failure_category"]({}, "API key is invalid") == "authentication_or_permission"
assert ns["_failure_category"]({"code": 147}) == "unclassified"
assert ns["_failure_category"]({}, "connection reset by peer") == "transient_transport"
assert "private-" not in str(state)
if __name__ == "__main__":
test_isolation()
test_no_stored_credentials()
test_proxy_preserves_safe_failure_diagnostics()
print("Gemini bot stack tests passed")
|