NexusOS/management/test_nexus_api.py

"""Pin the SSE parser in nexus_api. Run: python management/test_nexus_api.py"""
import sys, os
sys.path.insert(0, os.path.dirname(__file__))
from nexus_api import iter_chunks

# A realistic /chat/stream frame: two token chunks, a meta block, then done.
lines = [
    'data: "Hello"', "",
    'data: " world"', "",
    "event: meta", 'data: {"model":"mistral"}', "",
    "event: done", "data: {}", "",
]
out = list(iter_chunks(lines))
assert out == [
    ("chunk", "Hello"),
    ("chunk", " world"),
    ("meta", '{"model":"mistral"}'),
    ("done", "{}"),
], out
# error frame surfaces as its own kind, not a chunk
assert list(iter_chunks(["event: error", 'data: {"detail":"boom"}'])) == [
    ("error", '{"detail":"boom"}')
]
print("ok")