chore: gitignore hardening, VERSION, CI workflow, tests, docs
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,96 @@
|
||||
"""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())
|
||||
Reference in New Issue
Block a user