from __future__ import annotations import json import os import subprocess import sys from pathlib import Path ROOT = Path(__file__).resolve().parents[1] def run_cli(tmp_path: Path, *args: str) -> subprocess.CompletedProcess[str]: env = os.environ.copy() for key in tuple(env): if key.startswith("NEXUS_"): env.pop(key) env["NEXUS_HOME"] = str(tmp_path / "state") env["NEXUS_CONFIG_DIR"] = str(tmp_path / "config") return subprocess.run( [sys.executable, "-m", "nexusos_cli.cli", *args], cwd=ROOT, env=env, text=True, capture_output=True, timeout=30, check=False, ) def test_help_exposes_portable_command_tree(tmp_path): result = run_cli(tmp_path, "--help") assert result.returncode == 0, result.stderr for command in ("init", "config", "provider", "doctor", "serve", "models", "chat", "monitor"): assert command in result.stdout def test_legacy_cli_spellings_remain_compatible(): from nexusos_cli.cli import _normalize_legacy_argv assert _normalize_legacy_argv(["start", "-b"]) == ["start", "backend"] assert _normalize_legacy_argv(["stop", "--ai"]) == ["stop", "ai"] assert _normalize_legacy_argv(["backup", "full"]) == ["backup", "--full"] # -f is --follow for `logs`, but --frontend for start/stop. Translating it # for logs turned `logs -f` into a one-shot tail of the frontend log. assert _normalize_legacy_argv(["logs", "-f"]) == ["logs", "-f"] assert _normalize_legacy_argv(["start", "-f"]) == ["start", "frontend"] assert _normalize_legacy_argv(["restore", "-f"]) == ["restore"] assert _normalize_legacy_argv(["help"]) == ["--help"] def test_init_uses_external_state_and_seeds_playbooks(tmp_path): result = run_cli(tmp_path, "init", "--json") assert result.returncode == 0, result.stderr payload = json.loads(result.stdout) assert Path(payload["state_dir"]) == (tmp_path / "state").resolve() assert payload["seeded_playbooks"] == len(list((ROOT / "data" / "playbooks").glob("*.yaml"))) assert len(list((tmp_path / "state" / "data" / "playbooks").glob("*.yaml"))) > 0 def test_config_persists_validated_values(tmp_path): set_result = run_cli(tmp_path, "config", "set", "backend_port", "8123", "--json") assert set_result.returncode == 0, set_result.stderr get_result = run_cli(tmp_path, "config", "get", "backend_port", "--json") assert get_result.returncode == 0, get_result.stderr assert json.loads(get_result.stdout)["backend_port"] == 8123 invalid = run_cli(tmp_path, "config", "set", "backend_port", "70000") assert invalid.returncode == 2 assert "between 1 and 65535" in invalid.stderr def test_remote_provider_configuration_is_explicit(tmp_path): result = run_cli( tmp_path, "provider", "use", "remote", "--url", "http://phone-lan:11434", "--json" ) assert result.returncode == 0, result.stderr payload = json.loads(result.stdout) assert payload["provider"] == "ollama-remote" assert payload["url"] == "http://phone-lan:11434" values = json.loads((tmp_path / "config" / "config.json").read_text(encoding="utf-8")) assert values["provider"] == "ollama-remote" show = run_cli(tmp_path, "provider", "show", "--json") assert show.returncode == 0, show.stderr assert json.loads(show.stdout)["url"] == "http://phone-lan:11434" def test_serve_rejects_invalid_ports_before_startup(tmp_path): result = run_cli(tmp_path, "serve", "--port", "0") assert result.returncode == 2 assert "between 1 and 65535" in result.stderr def test_remote_provider_is_never_stopped_or_force_killed(monkeypatch): from nexusos_cli import ncp killed_ports = [] killed_patterns = [] monkeypatch.setattr(ncp.settings, "manage_ollama", False) monkeypatch.setattr(ncp.settings, "ollama_host", "http://remote.test:11434") monkeypatch.setattr(ncp, "kill_port", lambda port: killed_ports.append(port) or False) monkeypatch.setattr( ncp, "kill_matching", lambda patterns, force=False: killed_patterns.extend(patterns) or 0 ) ncp.stop_ollama() ncp.cmd_kill() assert 11434 not in killed_ports assert "ollama serve" not in killed_patterns def test_allow_lan_names_addresses_instead_of_disabling_the_host_check(): """ALLOWED_HOSTS=* would switch TrustedHostMiddleware off entirely, and that middleware is the DNS-rebinding defense for an unauthenticated API.""" from nexusos_cli.cli import _lan_hostnames assert _lan_hostnames("192.168.1.20") == ["192.168.1.20"] wildcard = _lan_hostnames("0.0.0.0") assert wildcard, "a wildcard bind must resolve to concrete host names" assert "*" not in wildcard # IPv6 literals need to match a Host header written either way. for name in wildcard: if ":" in name and not name.startswith("["): assert f"[{name}]" in wildcard def test_empty_platform_dir_variables_do_not_put_state_in_the_cwd(tmp_path, monkeypatch): """os.getenv's default only fires when a variable is *unset*; an exported but empty XDG_DATA_HOME / LOCALAPPDATA made Path("") == "." the base, so state landed in whatever directory the command happened to run from. Patching os.name to exercise the other platform's branch is not an option - pathlib dispatches on it - so this checks the branch this host actually takes.""" from synapse import nexus_config if os.name == "nt": monkeypatch.setenv("LOCALAPPDATA", "") expected = Path.home() / "AppData" / "Local" / "NexusOS" else: monkeypatch.setenv("XDG_DATA_HOME", "") expected = Path.home() / ".local/share" / "nexusos" monkeypatch.delenv("NEXUS_HOME", raising=False) monkeypatch.chdir(tmp_path) resolved = nexus_config._user_dir("NEXUS_HOME", "NexusOS", "XDG_DATA_HOME", ".local/share") assert resolved.is_absolute() assert tmp_path.resolve() != resolved.parent assert resolved == expected.resolve() def test_config_set_warns_before_orphaning_the_database(tmp_path): """Repointing the DB does not move it - say so, or history looks deleted.""" assert run_cli(tmp_path, "init").returncode == 0 db = tmp_path / "state" / "data" / "memory.db" db.parent.mkdir(parents=True, exist_ok=True) db.write_bytes(b"SQLite format 3" + bytes(1)) result = run_cli(tmp_path, "config", "set", "memory_db", str(tmp_path / "elsewhere.db"), "--json") assert result.returncode == 0, result.stderr assert "warning" in json.loads(result.stdout) same = run_cli(tmp_path, "config", "set", "memory_db", str(db), "--json") assert "warning" not in json.loads(same.stdout)