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