NexusOS/tests/test_smoke.py

"""Smoke test for the Synapse backend's core wiring.

Run from nexus-core/ with the Promethean venv active:  pytest -q
Or the whole gate (tests + frontend lint):             bin/check.sh

Deliberately tiny (ponytail): it guards the things v1 promises - app wiring,
model defaults, playbook ordering, persistence - without needing a running
Ollama service or network. Not a full suite.
"""
from pathlib import Path

from fastapi.testclient import TestClient
import pytest

from synapse.main import app
from synapse.memory.store import MemoryItem, PersistentMemoryStore
from synapse.nexus_config import DEFAULT_CHAT_MODEL, DEFAULT_MEMORY_MODEL
from synapse.ollama_manager import OllamaManager
from synapse import ollama_manager
from synapse.icons.compositor import _is_allowed_path
from synapse.playbook_manager import PlaybookManager
from synapse.playbooks.store import PlaybookFileStore, PlaybookItem

REPO_ROOT = Path(__file__).resolve().parent.parent


def test_app_builds_and_status_responds():
    # No `with` → startup event does not run; `ollama` stays the module-level
    # None and /status handles it. Pure liveness check. Not `/` — since the
    # single-process change that path serves the built web UI, not JSON.
    resp = TestClient(app).get("/status")
    assert resp.status_code == 200
    assert resp.json()["status"] == "online"


def test_keep_alive_pins_the_model():
    # Default keeps the model resident so chats skip cold reloads...
    assert PersistentMemoryStore._SETTINGS_DEFAULTS["keep_alive"] == "30m"
    mgr = OllamaManager()
    assert mgr._apply_keep_alive({"model": "x"}) == {"model": "x", "keep_alive": "30m"}
    # ...and an empty value omits the field (falls back to Ollama's default).
    mgr.keep_alive = ""
    assert "keep_alive" not in mgr._apply_keep_alive({"model": "x"})


def test_default_models_have_one_source_of_truth():
    # The curator default is the config constant, not a copy of it.
    assert PersistentMemoryStore._SETTINGS_DEFAULTS["memory_model"] == DEFAULT_MEMORY_MODEL
    # And the Windows installer reads the constants instead of hardcoding a tag,
    # which is what used to let installer and backend drift onto different models.
    ps1 = (REPO_ROOT / "install-windows.ps1").read_text(encoding="utf-8")
    assert "DEFAULT_CHAT_MODEL" in ps1
    for hardcoded in (DEFAULT_CHAT_MODEL, DEFAULT_MEMORY_MODEL, "qwen", "llama3.1"):
        assert hardcoded not in ps1, f"install-windows.ps1 hardcodes {hardcoded!r}"


def test_install_windows_stays_ascii():
    # PowerShell 5.1 decodes BOM-less files as ANSI: one stray Unicode dash
    # eats a quote and the whole script dies at parse time.
    for name in ("install-windows.ps1", "launch_nexus.ps1"):
        raw = (REPO_ROOT / name).read_bytes()
        bad = [(i, b) for i, b in enumerate(raw) if b > 0x7F]
        assert not bad, f"{name} has non-ASCII bytes at {bad[:3]}"


def test_first_playbook_is_the_system_prompt(tmp_path, monkeypatch):
    store = PlaybookFileStore(tmp_path)
    store.add_playbook(PlaybookItem(id="ctx", title="Reference", goal="ref goal",
                                    instructions="ref instructions", order=1))
    store.add_playbook(PlaybookItem(id="main", title="Main", goal="Be useful.",
                                    instructions="Answer briefly.", order=0))
    monkeypatch.setattr("synapse.playbook_manager.playbook_store", store)

    assert PlaybookManager.get_main_playbook().id == "main"
    assert [p.id for p in PlaybookManager.get_context_playbooks()] == ["ctx"]
    assert PlaybookManager.get_system_prompt() == "Be useful.\n\nAnswer briefly."


def test_settings_round_trip_over_defaults(tmp_path):
    store = PersistentMemoryStore(tmp_path / "memory.db")
    store.update_settings({"model": "some-model:8b"})
    settings = store.get_settings()
    assert settings["model"] == "some-model:8b"          # written value wins
    assert settings["memory_model"] == DEFAULT_MEMORY_MODEL  # untouched keys still default


def test_past_conversations_are_searchable(tmp_path):
    # The chat system prompt is built from these snippets, so a broken search
    # silently drops the assistant's recall of past chats.
    store = PersistentMemoryStore(tmp_path / "memory.db")
    store.create_conversation("c1")
    store.add_message("c1", "user", "how do I mount the Wingdrive?")
    store.add_message("c1", "assistant", "use rsync over ssh")

    assert store.search_conversations("wingdrive")   # case-insensitive substring
    assert store.search_conversations("nothing here") == []
    assert store.search_conversations("  ") == []


def test_memory_store_reads_changes_from_another_instance(tmp_path):
    db_path = tmp_path / "memory.db"
    writer = PersistentMemoryStore(db_path)
    reader = PersistentMemoryStore(db_path)

    writer.add(MemoryItem(id="fresh", text="written by another process"))
    assert reader.get("fresh").text == "written by another process"


def test_icon_source_requires_real_allowed_file_boundary(tmp_path):
    allowed = tmp_path / "icons"
    allowed.mkdir()
    source = allowed / "app.svg"
    source.write_text("<svg />")
    sibling = allowed.parent / "icons-other"
    sibling.mkdir()
    (sibling / "app.svg").write_text("<svg />")
    old_roots = list(__import__("synapse.icons.compositor", fromlist=["_ALLOWED_ROOTS"])._ALLOWED_ROOTS)
    module = __import__("synapse.icons.compositor", fromlist=["_ALLOWED_ROOTS"])
    module._ALLOWED_ROOTS[:] = [str(allowed)]
    try:
        assert _is_allowed_path(str(source))
        assert not _is_allowed_path(str(sibling / "app.svg"))
        assert not _is_allowed_path(str(allowed / "missing.svg"))
    finally:
        module._ALLOWED_ROOTS[:] = old_roots


def test_ollama_stream_propagates_transport_errors(monkeypatch):
    class FailingResponse:
        async def __aenter__(self):
            return self

        async def __aexit__(self, *args):
            return False

        def raise_for_status(self):
            raise RuntimeError("Ollama unavailable")

    class FailingClient:
        def __init__(self, **kwargs):
            pass

        async def __aenter__(self):
            return self

        async def __aexit__(self, *args):
            return False

        def stream(self, *args, **kwargs):
            return FailingResponse()

    monkeypatch.setattr(ollama_manager.httpx, "AsyncClient", FailingClient)

    async def consume():
        async for _ in OllamaManager()._chat_stream([], "model", 0):
            pass

    with pytest.raises(RuntimeError, match="Ollama unavailable"):
        import asyncio
        asyncio.run(consume())


def _load_sync():
    """bin/ isn't a package - load the cross-platform sync core by path."""
    import importlib.util
    spec = importlib.util.spec_from_file_location("sync", REPO_ROOT / "bin" / "sync.py")
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    return module


def test_sync_compare_detects_direction(tmp_path):
    """The guard that stops a stale box from overwriting the other's chats.
    Both backup and restore refuse to run when this says the wrong thing, so a
    silent break here loses conversation history."""
    import sqlite3 as sq
    sync = _load_sync()
    db, dump = tmp_path / "memory.db", tmp_path / "memory.db.sql"

    def write(rows):
        db.unlink(missing_ok=True)
        with sq.connect(db) as conn:
            conn.executescript(
                "create table conversations (id text primary key, updated_at text);"
                "create table memory (id text primary key);"
            )
            conn.executemany("insert into conversations values (?, ?)", rows)

    assert sync.compare(db, dump) == "no-dump"
    write([("a", "1")])
    assert sync.compare(db, dump) == "no-dump"

    # Dump matches the live DB exactly.
    sync.DB, sync.DB_SQL = db, dump
    assert sync.dump_db()
    assert sync.compare(db, dump) == "same"

    # A newer message bumps updated_at -> this box is ahead of the backup.
    write([("a", "2")])
    assert sync.compare(db, dump) == "local-ahead"

    # The backup holds a conversation this box never saw.
    write([])
    assert sync.compare(db, dump) == "local-behind"

    # Each side has something the other lacks.
    write([("b", "1")])
    assert sync.compare(db, dump) == "diverged"

    db.unlink()
    assert sync.compare(db, dump) == "no-live"


def test_restore_stages_match_the_script():
    """sync.py invokes restore-linux.sh stages by name with check=False, so a
    rename on one side alone fails silently - and the runtime stage is what
    installs Ollama and the ncp alias. Keep the two in agreement."""
    import re
    called = set(re.findall(
        r'linux_stage\(\s*"restore-linux\.sh"\s*,\s*"(\w+)"',
        (REPO_ROOT / "bin" / "sync.py").read_text()))
    script = (REPO_ROOT / "bin" / "restore-linux.sh").read_text()
    handled = set(re.search(r"case \"\$stage\" in\s*\n\s*([\w|]+)\)", script).group(1).split("|"))
    assert called, "no restore-linux.sh stages found in sync.py"
    assert called <= handled, f"sync.py calls unhandled stage(s): {called - handled}"


def test_desktop_stage_is_the_only_one_touching_home():
    """--no-desktop is only a real safety valve if the $HOME writes all live in
    the desktop stage. A test clone runs prep + runtime unconditionally."""
    script = (REPO_ROOT / "bin" / "restore-linux.sh").read_text()
    runtime = script.split('if [ "$stage" = "runtime" ]; then')[1].split("\n    exit 0\nfi")[0]
    # The ncp/promethean shell aliases are the one deliberate exception - they're
    # how you launch NexusOS at all, and both are grep-guarded no-ops on re-run.
    home_writes = [ln for ln in runtime.splitlines()
                   if "$HOME" in ln and ".bashrc" not in ln]
    assert not home_writes, f"runtime stage writes to $HOME: {home_writes}"