"""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 # noqa: E402 import naldo_opencode_bridge as bridge_module # noqa: E402 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") # Gemini_Bot's own lock stays; Naldo's lock must not accept granite. 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 # Rejected models: granite/gemini must not pass the predicate. 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 # type: ignore[assignment] stack.bridge_proc = bridge # type: ignore[assignment] stack.opencode_proc = opencode # type: ignore[assignment] 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")