forked from enderofwings/NexusOS
jonandClaude Opus 4.8
031e522704
Settings "Auto model routing" picks which installed model fires for chat vs coding intent when no model is pinned (auto_chat_model / auto_code_model). _auto_select_model honors the remap; _MODEL_PREFERENCE["code"] prefers real coder models first. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
331 lines
15 KiB
Python
331 lines
15 KiB
Python
"""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_auto_model_remap(monkeypatch):
|
|
import asyncio
|
|
from synapse import main
|
|
cfg = {"model": "", "auto_chat_model": "chatX", "auto_code_model": "coderY"}
|
|
monkeypatch.setattr(main.store, "get_settings", lambda: cfg)
|
|
# code intent ("function") routes to the code remap; chat intent to the chat remap
|
|
assert asyncio.run(main._auto_select_model("write a function to sort a list")) == "coderY"
|
|
assert asyncio.run(main._auto_select_model("how are you today")) == "chatX"
|
|
# an explicit pin beats the remap
|
|
cfg["model"] = "pinnedZ"
|
|
assert asyncio.run(main._auto_select_model("debug this code")) == "pinnedZ"
|
|
|
|
|
|
def test_hardware_fit_logic():
|
|
from synapse import hardware
|
|
assert hardware._fit(2.5, 4.0, 16.0) == "gpu" # 2.5+1 <= 4 -> fits GPU
|
|
assert hardware._fit(4.7, 4.0, 16.0) == "ram" # too big for 4GB GPU, fits RAM
|
|
assert hardware._fit(4.7, None, 16.0) == "ram" # VRAM unknown -> RAM
|
|
assert hardware._fit(40.0, 4.0, 16.0) == "no" # too big everywhere
|
|
rec = hardware.recommend()
|
|
assert "hardware" in rec and all("fit" in m for m in rec["models"])
|
|
|
|
|
|
def test_stt_status_endpoint():
|
|
# Reports whether local Whisper is installed; wiring must respond either way.
|
|
resp = TestClient(app).get("/stt/status")
|
|
assert resp.status_code == 200
|
|
assert isinstance(resp.json()["available"], bool)
|
|
|
|
|
|
def test_num_ctx_option_only_when_positive():
|
|
# 0 / None -> omit num_ctx so Ollama uses the model default; positive -> set it.
|
|
from synapse.ollama_manager import _chat_options
|
|
assert "num_ctx" not in _chat_options(None, None, 0)
|
|
assert "num_ctx" not in _chat_options(None, None, None)
|
|
assert _chat_options(None, None, 8192)["num_ctx"] == 8192
|
|
|
|
|
|
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_windows_scripts_stay_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. cmd.exe is worse
|
|
# still - it decodes by the console codepage. Globbed rather than listed by
|
|
# name so a newly added script is covered without editing this test.
|
|
skip = {"Promethean", "node_modules", ".git", "dist"}
|
|
ps1s = [p for pat in ("*.ps1", "*.cmd") for p in REPO_ROOT.rglob(pat)
|
|
if not skip & set(p.parts)]
|
|
assert ps1s, "no .ps1/.cmd files found - did the Windows path move?"
|
|
for path in ps1s:
|
|
raw = path.read_bytes()
|
|
bad = [(i, b) for i, b in enumerate(raw) if b > 0x7F]
|
|
assert not bad, f"{path.relative_to(REPO_ROOT)} has non-ASCII bytes at {bad[:3]}"
|
|
|
|
|
|
def test_ncp_is_registered_on_path_not_in_a_shell_profile():
|
|
"""A `function ncp` in a shell profile is invisible to cron, .desktop Exec
|
|
lines, cmd.exe and Task Scheduler - and on Windows the default Restricted
|
|
execution policy blocks the profile outright. Both installers must put ncp
|
|
on PATH; the profile wiring only survives as a no-sudo fallback."""
|
|
linux = (REPO_ROOT / "bin" / "restore-linux.sh").read_text(encoding="utf-8")
|
|
runtime = linux.split('if [ "$stage" = "runtime" ]; then')[1].split("\n exit 0\nfi")[0]
|
|
assert "/usr/local/bin/ncp" in runtime
|
|
|
|
ps1 = (REPO_ROOT / "install-windows.ps1").read_text(encoding="utf-8")
|
|
assert "ncp.cmd" in ps1, "installer must register the .cmd shim, not a profile function"
|
|
# setx truncates PATH at 1024 characters and has permanently broken machines.
|
|
# Comments stripped so the code that explains the ban does not trip it.
|
|
code = "\n".join(ln for ln in ps1.splitlines() if not ln.strip().startswith("#"))
|
|
assert "setx" not in code.lower()
|
|
assert 'SetEnvironmentVariable("Path"' in code
|
|
|
|
# The model pull is the only multi-GB step and the only one the script
|
|
# invites a Ctrl+C on -- which in PS 5.1 kills the whole script. Anything
|
|
# after it is lost, so it has to come last. It used to sit in the middle,
|
|
# and skipping the download silently skipped the ncp registration too.
|
|
assert ps1.index("ollama pull") > ps1.index("Registering the ncp command"), \
|
|
"model pull must come after ncp registration - a Ctrl+C there aborts the installer"
|
|
assert ps1.index("ollama pull") > ps1.index("Creating desktop shortcut"), \
|
|
"model pull must come after the desktop shortcut"
|
|
|
|
# And it must not TELL anyone to press Ctrl+C: in PS 5.1 that kills the
|
|
# script, so the advertised way to skip the download was also the way to
|
|
# abort the install. Skipping is a prompt now. Comments stripped so the
|
|
# comment explaining this does not trip the check.
|
|
assert "Ctrl+C" not in code, "installer must not offer Ctrl+C as a skip"
|
|
|
|
|
|
def test_no_ps1_shadows_the_ncp_path_shim():
|
|
"""PowerShell resolves ExternalScript (.ps1) ahead of Application (.cmd), so
|
|
an ncp.ps1 sitting next to ncp.cmd wins in PowerShell and drags the execution
|
|
policy back in - the exact thing the .cmd exists to avoid. Observed on the
|
|
Windows VM: `Get-Command ncp -All` listed ncp.ps1 first, from the same
|
|
directory the installer had just put on PATH."""
|
|
shim = REPO_ROOT / "management" / "ncp.cmd"
|
|
assert shim.exists(), "the Windows PATH shim is missing"
|
|
twin = shim.with_suffix(".ps1")
|
|
assert not twin.exists(), f"{twin.name} shadows {shim.name} in PowerShell"
|
|
|
|
|
|
|
|
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]
|
|
# Shell wiring is the one deliberate exception: registering ncp in ~/.bashrc
|
|
# and in the PowerShell profile is how you launch NexusOS at all, and each is
|
|
# a grep-guarded no-op on re-run. Desktop config (xfconf, plank, themes,
|
|
# os-release) must stay in the desktop stage so --no-desktop really is safe.
|
|
shell_wiring = (".bashrc",)
|
|
home_writes = [ln for ln in runtime.splitlines()
|
|
if "$HOME" in ln and not any(w in ln for w in shell_wiring)]
|
|
assert not home_writes, f"runtime stage writes to $HOME: {home_writes}"
|
GitNexus · part of the Nexus stack