| """Naldo_Bot on OpenCode only: offline unit tests. |
| |
| No network, no model, no Minecraft. Pure stdlib plus text-level assertions |
| on the deployment files. |
| """ |
|
|
| from __future__ import annotations |
|
|
| import ast |
| import sys |
| from pathlib import Path |
| from types import SimpleNamespace |
| from unittest import mock |
|
|
| ROOT = Path(__file__).resolve().parents[1] |
| sys.path.insert(0, str(ROOT)) |
|
|
| import naldo_opencode as brain |
| import naldo_opencode_bridge as bridge_module |
|
|
|
|
| def test_primary_and_fallbacks_are_all_opencode_free(): |
| assert brain.OPENCODE_PROVIDER == "opencode" |
| assert brain.PRIMARY_MODEL == "muse-spark-1.3-contributor-free" |
| assert brain.FALLBACK_MODELS[0] == brain.PRIMARY_MODEL |
| assert set(brain.FALLBACK_MODELS) == { |
| "muse-spark-1.3-contributor-free", |
| "fledge-alpha-free", |
| "nemotron-3-ultra-free", |
| } |
| for model in brain.FALLBACK_MODELS: |
| assert brain.is_allowed_model(model) |
| assert brain.is_allowed_model(f"opencode/{model}") |
| lowered = model.lower() |
| for forbidden in ("granite", "gemini", "gpt", "claude", "anthropic", "qwen"): |
| assert forbidden not in lowered |
|
|
|
|
| def test_no_granite_gemini_fallback_anywhere(): |
| for name in ("granite", "gemini", "gpt-", "anthropic", "claude"): |
| assert name not in {m.lower() for m in brain.FALLBACK_MODELS} |
| stack = (ROOT / "voyager_stack.py").read_text(encoding="utf-8") |
| assert "granite-4.0-h-tiny" not in stack, "Naldo stack must not reference Granite anymore" |
| bridge = (ROOT / "naldo_opencode_bridge.py").read_text(encoding="utf-8") |
| lowered = bridge.lower() |
| for forbidden in ("granite", "gemini", "anthropic"): |
| assert forbidden not in lowered, f"bridge leaked {forbidden}" |
| runner = (ROOT / "voyager_runner.py").read_text(encoding="utf-8") |
| |
| assert "naldo-granite" not in runner |
|
|
|
|
| def test_brain_lock_requires_opencode_profile_base_and_models(): |
| runner = (ROOT / "voyager_runner.py").read_text(encoding="utf-8") |
| assert "NALDO_BRAIN_PROFILE" in runner or "naldo-opencode" in runner |
| assert "http://127.0.0.1:7864/v1" in runner |
| assert "muse-spark-1.3-contributor-free" in runner |
| assert "_naldo_model_allowed" in runner |
| |
| assert not brain.is_allowed_model("granite-4.0-h-tiny") |
| assert not brain.is_allowed_model("gemini-3.5-flash-lite") |
| assert not brain.is_allowed_model("gpt-4o") |
| assert brain.is_allowed_model("fledge-alpha-free") |
| assert brain.is_allowed_model("nemotron-3-ultra-free") |
|
|
|
|
| def test_bridge_formats_messages_without_confusing_roles(): |
| messages = [ |
| {"role": "system", "content": "You are a planner."}, |
| {"role": "user", "content": "Do X"}, |
| {"role": "assistant", "content": "Plan Y"}, |
| {"role": "tool", "content": "tool output Z"}, |
| {"role": "user", "content": "Now?", "tool_calls": [{"id": "1", "function": {"name": "f", "arguments": "{}"}}]}, |
| ] |
| rendered = brain.build_opencode_prompt(messages) |
| assert "[SYSTEM]" in rendered["prompt"] |
| assert "[USER]" in rendered["prompt"] |
| assert "[ASSISTANT]" in rendered["prompt"] |
| assert "[TOOL]" in rendered["prompt"] |
| assert "You are a planner." in rendered["system"] |
| assert "TOOL_CALLS" in rendered["prompt"] |
| assert "text generator, not a file/shell executor" in rendered["prompt"] |
|
|
|
|
| def test_bridge_extracts_only_text_parts(): |
| payload = { |
| "info": {"role": "assistant"}, |
| "parts": [ |
| {"type": "reasoning", "thinking": "secret plan"}, |
| {"type": "text", "text": "hello "}, |
| {"type": "tool", "name": "bash", "input": "rm -rf /"}, |
| {"type": "text", "text": "world"}, |
| ], |
| } |
| assert brain.extract_assistant_text(payload) == "hello world" |
| assert brain.extract_assistant_text({"parts": []}) == "" |
| assert brain.extract_assistant_text({}) == "" |
|
|
|
|
| def test_sessions_use_free_tier_compatible_tool_sandbox(tmp_path): |
| captured = {} |
|
|
| def fake_http(url, method="GET", body=None, timeout=0, retries=0): |
| captured.update({"url": url, "method": method, "body": body}) |
| return {"id": "ses_test"} |
|
|
| with mock.patch.object(bridge_module, "OPENCODE_SESSION_DIR", tmp_path), \ |
| mock.patch.object(bridge_module, "_http_json", side_effect=fake_http): |
| session_id = bridge_module._create_session("muse-spark-1.3-contributor-free") |
|
|
| assert session_id == "ses_test" |
| assert captured["method"] == "POST" |
| assert captured["url"].endswith("/session") |
| assert captured["body"]["model"] == { |
| "id": "muse-spark-1.3-contributor-free", |
| "providerID": "opencode", |
| } |
| assert captured["body"]["location"] == {"directory": str(tmp_path)} |
| assert "permission" not in captured["body"], "custom permissions trigger Zen free-tier 403" |
|
|
| message = brain.opencode_message_body("muse-spark-1.3-contributor-free", "sys", "do X") |
| assert message["tools"] == {}, "empty tools map is the verified free-tier tool sandbox" |
| assert "temperature" not in message |
|
|
|
|
| def test_loopback_only(): |
| assert brain.is_loopback_url(brain.OPENCODE_BASE) |
| assert brain.is_loopback_url(brain.BRIDGE_BASE) |
| assert "0.0.0.0" not in (ROOT / "naldo_opencode_bridge.py").read_text(encoding="utf-8") |
| stack = (ROOT / "voyager_stack.py").read_text(encoding="utf-8") |
| assert "opencode_serve_command(opencode_binary" in stack |
| assert 'host="127.0.0.1"' in (ROOT / "naldo_opencode_bridge.py").read_text(encoding="utf-8") |
| assert "shell=True" not in (ROOT / "naldo_opencode_bridge.py").read_text(encoding="utf-8") |
| assert "shell=True" not in (ROOT / "naldo_opencode.py").read_text(encoding="utf-8") |
| serve = brain.opencode_serve_command("/tmp/opencode-local", 7865) |
| assert serve[:3] == ["/tmp/opencode-local", "serve", "--hostname"] |
| assert "127.0.0.1" in serve |
| assert "0.0.0.0" not in serve |
|
|
|
|
| def test_no_credential_dependency_or_leak(tmp_path): |
| bridge_src = (ROOT / "naldo_opencode_bridge.py").read_text(encoding="utf-8") |
| assert "auth.json" not in bridge_src |
| assert 'os.environ.get("OPENCODE_API_KEY"' not in bridge_src |
| assert 'os.environ["OPENCODE_API_KEY"]' not in bridge_src |
|
|
| base = {"OPENCODE_API_KEY": "must-not-survive", "PATH": "x"} |
| env = brain.opencode_subprocess_env(base, tmp_path / "home") |
| assert "OPENCODE_API_KEY" not in env |
| assert env["PATH"] == "x" |
| assert env["HOME"] == str(tmp_path / "home") |
| assert env["XDG_CONFIG_HOME"] == str(tmp_path / "home" / "config") |
| assert env["XDG_DATA_HOME"] == str(tmp_path / "home" / "data") |
| assert base["OPENCODE_API_KEY"] == "must-not-survive", "parent env must not be mutated" |
| assert "TEMPERATURE_NOTE" in (ROOT / "naldo_opencode.py").read_text(encoding="utf-8") |
|
|
|
|
| def test_local_opencode_install_is_pinned_and_non_root(tmp_path): |
| calls = [] |
|
|
| def fake_run(args, **kwargs): |
| calls.append(list(args)) |
| return SimpleNamespace(returncode=0, stdout="installed") |
|
|
| with mock.patch.object(brain, "_probe_opencode", side_effect=["", brain.OPENCODE_VERSION]), \ |
| mock.patch.object(brain.subprocess, "run", side_effect=fake_run): |
| binary, version = brain.ensure_opencode_installed(tmp_path / "cli") |
|
|
| assert version == brain.OPENCODE_VERSION |
| assert Path(binary) == brain.opencode_binary_path(tmp_path / "cli") |
| install = calls[0] |
| assert "-g" not in install and "--global" not in install |
| assert "--prefix" in install |
| assert install[install.index("--prefix") + 1] == str(tmp_path / "cli") |
| assert brain.OPENCODE_NPM_SPEC in install |
|
|
|
|
| def test_max_tokens_is_telemetry_not_output_truncation(): |
| long_answer = "function build() { return 1; }\n" * 300 |
| with mock.patch.object(bridge_module, "_create_session", return_value="ses_x"), \ |
| mock.patch.object(bridge_module, "_send_message", return_value=long_answer): |
| result = bridge_module._chat_payload({ |
| "model": brain.PRIMARY_MODEL, |
| "messages": [{"role": "user", "content": "write program"}], |
| "temperature": 0, |
| "max_tokens": 1, |
| }) |
| assert result["choices"][0]["message"]["content"] == long_answer |
| assert bridge_module.state["last_requested_max_tokens"] == 1 |
|
|
|
|
| def test_supervisor_start_stop_contract(): |
| stack_src = (ROOT / "voyager_stack.py").read_text(encoding="utf-8") |
| for marker in ( |
| "_ensure_naldo_brain", |
| "_brain_healthy", |
| "_stop_brain", |
| "brain_status", |
| "_terminate_naldo_child", |
| "ensure_opencode_installed", |
| "NALDO_BRIDGE_BASE", |
| "NALDO_BRAIN_PROFILE", |
| "NALDO_PRIMARY_MODEL", |
| "opencode_serve_command", |
| "opencode_subprocess_env", |
| ): |
| assert marker in stack_src, f"supervisor marker missing: {marker}" |
| assert "without Granite fallback" in stack_src or "without granite" in stack_src.lower() |
| tree = ast.parse(stack_src) |
| methods = set() |
| for node in ast.walk(tree): |
| if isinstance(node, ast.ClassDef) and node.name == "VoyagerStack": |
| for child in node.body: |
| if isinstance(child, (ast.FunctionDef, ast.AsyncFunctionDef)): |
| methods.add(child.name) |
| for required in ("_ensure_naldo_brain", "_brain_healthy", "_stop_brain", "brain_status", "stop", "start_async"): |
| assert required in methods |
| assert "self.stop_event.wait(7)" in stack_src |
| assert "OpenCode/bridge died mid-run; stopping Naldo immediately" in stack_src |
| assert "opencode_serve_command(opencode_binary" in stack_src |
|
|
| runner_src = (ROOT / "voyager_runner.py").read_text(encoding="utf-8") |
| assert "Naldo OpenCode brain module unavailable" in runner_src |
| assert "except Exception as exc" in runner_src |
|
|
|
|
| def test_supervisor_stop_terminates_all_children(): |
| import voyager_stack as stack_module |
|
|
| stack = stack_module.VoyagerStack() |
| naldo = SimpleNamespace(poll=lambda: None, terminate=lambda: setattr(naldo, "t", True), kill=lambda: None, wait=lambda timeout=None: 0) |
| bridge = SimpleNamespace(poll=lambda: None, terminate=lambda: setattr(bridge, "t", True), kill=lambda: None, wait=lambda timeout=None: 0) |
| opencode = SimpleNamespace(poll=lambda: None, terminate=lambda: setattr(opencode, "t", True), kill=lambda: None, wait=lambda timeout=None: 0) |
| stack.proc = naldo |
| stack.bridge_proc = bridge |
| stack.opencode_proc = opencode |
| naldo.pid = 111 |
| bridge.pid = 222 |
| opencode.pid = 333 |
| stack.stop_event.set() |
| with mock.patch.object(stack_module, "_wait_tcp", return_value=True): |
| with mock.patch.object(stack, "brain_status", return_value={"ok": True}): |
| with mock.patch.object(stack, "_mineflayer_status", return_value={"ready": False}): |
| status = stack.stop() |
| assert getattr(naldo, "t", False) is True |
| assert getattr(bridge, "t", False) is True |
| assert getattr(opencode, "t", False) is True |
| assert status["state"] == "stopped" |
|
|
|
|
| def test_gemini_bot_untouched(): |
| stack_src = (ROOT / "gemini_bot_stack.py").read_text(encoding="utf-8") |
| assert "7863" in stack_src |
| assert "gemini-3.5-flash-lite" in stack_src |
| assert "opencode" not in stack_src.lower() |
| assert "7864" not in stack_src |
| assert "7865" not in stack_src |
|
|
|
|
| if __name__ == "__main__": |
| test_primary_and_fallbacks_are_all_opencode_free() |
| test_no_granite_gemini_fallback_anywhere() |
| test_brain_lock_requires_opencode_profile_base_and_models() |
| test_bridge_formats_messages_without_confusing_roles() |
| test_bridge_extracts_only_text_parts() |
| test_sessions_created_with_denied_permissions() |
| test_loopback_only() |
| test_no_credential_dependency_or_leak() |
| test_supervisor_start_stop_contract() |
| test_gemini_bot_untouched() |
| print("PASS naldo opencode") |
|
|
