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

Deliberately tiny (ponytail): it guards core app wiring and local Ollama
behavior without needing a running Ollama service or network. Not a full suite.
"""
from fastapi.testclient import TestClient
import pytest

from synapse.main import app
from synapse.memory.store import MemoryItem, PersistentMemoryStore
from synapse.ollama_manager import OllamaManager
from synapse import ollama_manager
from synapse.icons.compositor import _is_allowed_path


def test_app_builds_and_root_responds():
    # No `with` → startup event (which boots Ollama) does not run; `ollama`
    # stays the module-level None and root() handles it. Pure liveness check.
    resp = TestClient(app).get("/")
    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_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())