NexusOS/tests/test_cli_packaging.py

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(["logs", "-m", "--follow"]) == ["logs", "memory", "--follow"]
    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(["logs", "-m", "-f"]) == ["logs", "memory", "-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)