Working chat: auth, connections, streaming, folders
LLeMbas now runs end to end. Register, add an OpenAI-compatible connection, and hold a real streaming conversation organised into folders. Verified against the local llama-swap instance. Streaming is the one genuinely tricky part. Sending a message returns two HTML fragments -- the user bubble and an empty assistant bubble carrying an sse-connect -- and that attribute is the ONLY thing that starts a generation. Rendering an incomplete assistant message as a streaming shell falls out of the same template, which means loading a page whose last reply never finished simply picks it up again. Details worth knowing about, each commented where it matters: - SSE payloads are split across several data: lines. A raw newline in one data: line truncates the event, which shows up the first time a model emits a code block. - Markdown is rendered server-side by the same helper for both the page and the final streamed frame, so the two cannot disagree. The fence renderer is replaced outright rather than using markdown-it's highlight option, which re-wraps output in a second <pre>. - escape_text is html.escape, not nh3.clean_text: it escapes character by character, so escaping stream chunks separately equals escaping the whole string. - The stream opens its own session via session_scope(); it outlives the request handler and the dependency-scoped session may be closed. - Deleting a folder keeps the chats inside it (FK is SET NULL). Losing a conversation to a mis-clicked folder delete is unforgivable. - Login failures use one message for "no such account" and "wrong password" so the form cannot enumerate registered addresses. Also adds deploy/ for the gamebox install at https://chat.lan: system unit, nginx vhost with buffering off (buffering on turns streaming into one lump at the end), and install/update scripts following the same service-user and /srv bind-mount conventions as llama-swap and comfyui. 70 tests, ruff clean. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Jaroslav Beneš
@@ -0,0 +1,98 @@
|
||||
"""Test fixtures.
|
||||
|
||||
Every test runs against a throwaway SQLite file in a tmp_path, never the real
|
||||
data directory. The environment has to be set before lembas.config is imported,
|
||||
because Settings is a cached singleton read at import time.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import tempfile
|
||||
from collections.abc import Iterator
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
# Must happen before any lembas import.
|
||||
_TMP = tempfile.mkdtemp(prefix="lembas-tests-")
|
||||
os.environ.update(
|
||||
{
|
||||
"LEMBAS_SECRET_KEY": "test-secret-key-not-for-real-use",
|
||||
"LEMBAS_DATA_DIR": _TMP,
|
||||
"LEMBAS_ALLOW_SIGNUP": "true",
|
||||
"LEMBAS_LOG_LEVEL": "warning",
|
||||
}
|
||||
)
|
||||
|
||||
from fastapi.testclient import TestClient # noqa: E402
|
||||
from sqlalchemy.orm import Session # noqa: E402
|
||||
|
||||
from lembas.config import settings # noqa: E402
|
||||
from lembas.db.base import Base # noqa: E402
|
||||
from lembas.db.session import get_engine, get_session_factory, reset_engine # noqa: E402
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def fresh_database(tmp_path: Path) -> Iterator[None]:
|
||||
"""Point the engine at a per-test database and build the schema.
|
||||
|
||||
reset_engine() is essential: the engine is a module-level singleton, so
|
||||
without it every test after the first would share the first one's file.
|
||||
"""
|
||||
settings.data_dir = tmp_path
|
||||
reset_engine()
|
||||
settings.ensure_dirs()
|
||||
|
||||
import lembas.db.models # noqa: F401 (registers the tables)
|
||||
|
||||
Base.metadata.create_all(bind=get_engine())
|
||||
yield
|
||||
reset_engine()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def db() -> Iterator[Session]:
|
||||
session = get_session_factory()()
|
||||
try:
|
||||
yield session
|
||||
finally:
|
||||
session.close()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def client() -> Iterator[TestClient]:
|
||||
from lembas.main import app
|
||||
|
||||
# raise_server_exceptions=False so error-handler behaviour is exercised
|
||||
# rather than the exception propagating into the test.
|
||||
with TestClient(app, raise_server_exceptions=False) as test_client:
|
||||
yield test_client
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def registered(client: TestClient) -> dict[str, str]:
|
||||
"""Register the first account. It becomes the administrator."""
|
||||
credentials = {
|
||||
"name": "Frodo",
|
||||
"email": "frodo@shire.test",
|
||||
"password": "speak-friend-and-enter",
|
||||
}
|
||||
response = client.post("/auth/register", data=credentials, follow_redirects=False)
|
||||
assert response.status_code == 303, response.text
|
||||
return credentials
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def user_id(db: Session, registered: dict[str, str]) -> str:
|
||||
"""The registered user's id.
|
||||
|
||||
Chats have a real foreign key to users and SQLite enforces it (the
|
||||
connect-time PRAGMA in db/session.py turns that on), so tests that build a
|
||||
Chat directly need a user that actually exists.
|
||||
"""
|
||||
from sqlalchemy import select
|
||||
|
||||
from lembas.db.models import User
|
||||
|
||||
return db.scalar(select(User).where(User.email == registered["email"])).id
|
||||
@@ -0,0 +1,132 @@
|
||||
"""Registration, sign-in, and the route guards."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from fastapi.testclient import TestClient
|
||||
from sqlalchemy import select
|
||||
|
||||
from lembas.db.models import ROLE_ADMIN, ROLE_USER, User
|
||||
|
||||
|
||||
def test_empty_instance_sends_you_to_registration(client: TestClient):
|
||||
response = client.get("/auth/login", follow_redirects=False)
|
||||
assert response.status_code == 303
|
||||
assert response.headers["location"] == "/auth/register"
|
||||
|
||||
|
||||
def test_first_account_becomes_admin(client: TestClient, db, registered):
|
||||
user = db.scalar(select(User).where(User.email == registered["email"]))
|
||||
assert user.role == ROLE_ADMIN
|
||||
|
||||
|
||||
def test_second_account_is_an_ordinary_user(client: TestClient, db, registered):
|
||||
client.post("/auth/logout", follow_redirects=False)
|
||||
client.post(
|
||||
"/auth/register",
|
||||
data={"name": "Sam", "email": "sam@shire.test", "password": "potatoes-po-ta-toes"},
|
||||
follow_redirects=False,
|
||||
)
|
||||
user = db.scalar(select(User).where(User.email == "sam@shire.test"))
|
||||
assert user.role == ROLE_USER
|
||||
|
||||
|
||||
def test_registration_signs_you_in(client: TestClient, registered):
|
||||
assert client.get("/chat").status_code == 200
|
||||
|
||||
|
||||
def test_duplicate_email_is_rejected(client: TestClient, registered):
|
||||
client.post("/auth/logout", follow_redirects=False)
|
||||
response = client.post("/auth/register", data=registered, follow_redirects=False)
|
||||
assert response.status_code == 400
|
||||
assert "already exists" in response.text
|
||||
|
||||
|
||||
def test_short_password_is_rejected(client: TestClient):
|
||||
response = client.post(
|
||||
"/auth/register",
|
||||
data={"name": "A", "email": "a@b.test", "password": "short"},
|
||||
follow_redirects=False,
|
||||
)
|
||||
assert response.status_code == 400
|
||||
assert "8 characters" in response.text
|
||||
|
||||
|
||||
def test_sign_in_and_out(client: TestClient, registered):
|
||||
client.post("/auth/logout", follow_redirects=False)
|
||||
assert client.get("/chat", follow_redirects=False).status_code == 303
|
||||
|
||||
response = client.post(
|
||||
"/auth/login",
|
||||
data={"email": registered["email"], "password": registered["password"]},
|
||||
follow_redirects=False,
|
||||
)
|
||||
assert response.status_code == 303
|
||||
assert client.get("/chat").status_code == 200
|
||||
|
||||
|
||||
def test_wrong_password_does_not_reveal_whether_the_account_exists(
|
||||
client: TestClient, registered
|
||||
):
|
||||
client.post("/auth/logout", follow_redirects=False)
|
||||
wrong = client.post(
|
||||
"/auth/login",
|
||||
data={"email": registered["email"], "password": "wrong-password-here"},
|
||||
follow_redirects=False,
|
||||
)
|
||||
missing = client.post(
|
||||
"/auth/login",
|
||||
data={"email": "nobody@nowhere.test", "password": "wrong-password-here"},
|
||||
follow_redirects=False,
|
||||
)
|
||||
assert wrong.status_code == missing.status_code == 401
|
||||
assert "do not match" in wrong.text
|
||||
# Identical wording is the whole point: the form must not be usable to
|
||||
# enumerate which addresses are registered.
|
||||
assert ("do not match" in missing.text) == ("do not match" in wrong.text)
|
||||
|
||||
|
||||
def test_logout_revokes_the_session_immediately(client: TestClient, registered):
|
||||
client.post("/auth/logout", follow_redirects=False)
|
||||
assert client.get("/chat", follow_redirects=False).status_code == 303
|
||||
|
||||
|
||||
def test_signed_out_pages_redirect_to_login(client: TestClient, registered):
|
||||
client.post("/auth/logout", follow_redirects=False)
|
||||
for path in ("/", "/chat", "/settings", "/admin/connections"):
|
||||
assert client.get(path, follow_redirects=False).status_code == 303, path
|
||||
|
||||
|
||||
def test_htmx_requests_get_a_redirect_header_not_a_login_page(
|
||||
client: TestClient, registered
|
||||
):
|
||||
"""An htmx request must never swap a login form into a fragment of the UI."""
|
||||
client.post("/auth/logout", follow_redirects=False)
|
||||
response = client.post("/api/chats", headers={"HX-Request": "true"})
|
||||
assert response.status_code == 204
|
||||
assert response.headers["HX-Redirect"] == "/auth/login"
|
||||
|
||||
|
||||
def test_admin_area_is_closed_to_ordinary_users(client: TestClient, registered):
|
||||
client.post("/auth/logout", follow_redirects=False)
|
||||
client.post(
|
||||
"/auth/register",
|
||||
data={"name": "Sam", "email": "sam@shire.test", "password": "potatoes-po-ta-toes"},
|
||||
follow_redirects=False,
|
||||
)
|
||||
assert client.get("/admin/connections", follow_redirects=False).status_code == 403
|
||||
|
||||
|
||||
def test_login_next_parameter_cannot_be_used_for_an_open_redirect(
|
||||
client: TestClient, registered
|
||||
):
|
||||
client.post("/auth/logout", follow_redirects=False)
|
||||
response = client.post(
|
||||
"/auth/login",
|
||||
data={
|
||||
"email": registered["email"],
|
||||
"password": registered["password"],
|
||||
"next": "https://evil.example.com/steal",
|
||||
},
|
||||
follow_redirects=False,
|
||||
)
|
||||
assert response.headers["location"] == "/"
|
||||
@@ -0,0 +1,313 @@
|
||||
"""Chat, folders, and the streaming reply path against a mocked endpoint."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import httpx
|
||||
import pytest
|
||||
from fastapi.testclient import TestClient
|
||||
from sqlalchemy import select
|
||||
|
||||
from lembas.db.models import Chat, Connection, Folder, Message, Model
|
||||
from lembas.services import chat as chat_service
|
||||
from lembas.services.crypto import encrypt
|
||||
from lembas.services.llm.openai_client import Endpoint, LLMError, delta_text, list_models
|
||||
from lembas.services.sse import event
|
||||
|
||||
|
||||
# --- SSE framing -------------------------------------------------------------
|
||||
def test_sse_event_framing():
|
||||
assert event("token", "hello") == "event: token\ndata: hello\n\n"
|
||||
|
||||
|
||||
def test_sse_splits_newlines_across_data_lines():
|
||||
"""A payload with a newline must become several data: lines. Sending a raw
|
||||
newline truncates the event, which is what breaks the first code block a
|
||||
model emits."""
|
||||
assert event("token", "a\nb") == "event: token\ndata: a\ndata: b\n\n"
|
||||
|
||||
|
||||
def test_sse_round_trips_through_the_browser_rejoin_rule():
|
||||
payload = "line one\nline two\n\nline four"
|
||||
framed = event("token", payload)
|
||||
body = framed.split("\n", 1)[1]
|
||||
rejoined = "\n".join(
|
||||
line.removeprefix("data: ") for line in body.split("\n") if line.startswith("data:")
|
||||
)
|
||||
assert rejoined == payload
|
||||
|
||||
|
||||
# --- Delta parsing -----------------------------------------------------------
|
||||
def test_delta_text_reads_the_normal_shape():
|
||||
assert delta_text({"choices": [{"delta": {"content": "hi"}}]}) == "hi"
|
||||
|
||||
|
||||
def test_delta_text_handles_typed_content_parts():
|
||||
chunk = {"choices": [{"delta": {"content": [{"type": "text", "text": "hi"}]}}]}
|
||||
assert delta_text(chunk) == "hi"
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"chunk", [{}, {"choices": []}, {"choices": [{}]}, {"choices": [{"delta": {}}]}]
|
||||
)
|
||||
def test_delta_text_tolerates_junk(chunk):
|
||||
"""Providers vary; an unexpected chunk shape must not kill a reply."""
|
||||
assert delta_text(chunk) == ""
|
||||
|
||||
|
||||
# --- Endpoint URL handling ---------------------------------------------------
|
||||
@pytest.mark.parametrize(
|
||||
("base", "expected"),
|
||||
[
|
||||
("http://host:1234", "http://host:1234/v1/models"),
|
||||
("http://host:1234/v1", "http://host:1234/v1/models"),
|
||||
("http://host:1234/", "http://host:1234/v1/models"),
|
||||
],
|
||||
)
|
||||
def test_base_url_with_or_without_v1(base, expected):
|
||||
"""Users should not have to guess which form is expected."""
|
||||
assert Endpoint(base_url=base.rstrip("/"), api_key="", extra_headers={}).url(
|
||||
"models"
|
||||
) == expected
|
||||
|
||||
|
||||
def test_no_authorization_header_without_a_key():
|
||||
"""Local runners often reject an empty bearer token outright."""
|
||||
assert "Authorization" not in Endpoint("http://h", "", {}).headers()
|
||||
assert Endpoint("http://h", "k", {}).headers()["Authorization"] == "Bearer k"
|
||||
|
||||
|
||||
# --- Model discovery ---------------------------------------------------------
|
||||
async def test_list_models_accepts_the_bare_list_shape():
|
||||
"""The spec says {"data": [...]}, but some servers return a bare list."""
|
||||
|
||||
def handler(_request):
|
||||
return httpx.Response(200, json=[{"id": "a"}, {"id": "b"}])
|
||||
|
||||
original = httpx.AsyncClient
|
||||
|
||||
class Patched(original):
|
||||
def __init__(self, **kwargs):
|
||||
super().__init__(transport=httpx.MockTransport(handler), **kwargs)
|
||||
|
||||
httpx.AsyncClient = Patched
|
||||
try:
|
||||
assert [m["id"] for m in await list_models(Endpoint("http://h", "", {}))] == ["a", "b"]
|
||||
finally:
|
||||
httpx.AsyncClient = original
|
||||
|
||||
|
||||
async def test_list_models_reports_a_rejected_key_readably():
|
||||
def handler(_request):
|
||||
return httpx.Response(401, json={"error": {"message": "Incorrect API key."}})
|
||||
|
||||
original = httpx.AsyncClient
|
||||
|
||||
class Patched(original):
|
||||
def __init__(self, **kwargs):
|
||||
super().__init__(transport=httpx.MockTransport(handler), **kwargs)
|
||||
|
||||
httpx.AsyncClient = Patched
|
||||
try:
|
||||
with pytest.raises(LLMError) as caught:
|
||||
await list_models(Endpoint("http://h", "bad", {}))
|
||||
assert "rejected" in caught.value.message
|
||||
assert "Incorrect API key." in caught.value.message
|
||||
finally:
|
||||
httpx.AsyncClient = original
|
||||
|
||||
|
||||
# --- Titles ------------------------------------------------------------------
|
||||
def test_fallback_title_keeps_a_short_message_intact():
|
||||
assert chat_service.fallback_title("What is lembas?") == "What is lembas?"
|
||||
|
||||
|
||||
def test_fallback_title_trims_on_a_word_boundary():
|
||||
title = chat_service.fallback_title("word " * 60)
|
||||
assert len(title) <= chat_service.MAX_TITLE_LENGTH + 1
|
||||
assert title.endswith("…")
|
||||
|
||||
|
||||
def test_fallback_title_of_nothing():
|
||||
assert chat_service.fallback_title(" ") == "New chat"
|
||||
|
||||
|
||||
# --- Chats and folders (through the API) -------------------------------------
|
||||
def _add_connection(db) -> Connection:
|
||||
# Port 1 refuses connections, which is what the error-path test relies on.
|
||||
connection = Connection(
|
||||
name="Test", base_url="http://127.0.0.1:1", api_key_encrypted=encrypt("")
|
||||
)
|
||||
db.add(connection)
|
||||
db.commit()
|
||||
db.add(Model(connection_id=connection.id, model_id="test-model"))
|
||||
db.commit()
|
||||
return connection
|
||||
|
||||
|
||||
def test_new_chat_redirects_to_its_own_url(client: TestClient, db, registered):
|
||||
_add_connection(db)
|
||||
response = client.post("/api/chats", headers={"HX-Request": "true"})
|
||||
assert response.status_code == 204
|
||||
assert response.headers["HX-Redirect"].startswith("/chat/")
|
||||
|
||||
|
||||
def test_new_chat_picks_up_the_default_model(client: TestClient, db, registered):
|
||||
_add_connection(db)
|
||||
client.post("/api/chats", headers={"HX-Request": "true"})
|
||||
assert db.scalar(select(Chat)).model_id == "test-model"
|
||||
|
||||
|
||||
def test_posting_a_message_stores_both_turns(client: TestClient, db, registered):
|
||||
_add_connection(db)
|
||||
chat_id = client.post("/api/chats").headers["HX-Redirect"].rsplit("/", 1)[1]
|
||||
|
||||
response = client.post(f"/api/chats/{chat_id}/messages", data={"content": "Hello there"})
|
||||
assert response.status_code == 200
|
||||
|
||||
messages = db.scalars(select(Message).order_by(Message.created_at)).all()
|
||||
assert [m.role for m in messages] == ["user", "assistant"]
|
||||
assert messages[0].content == "Hello there"
|
||||
# The assistant row is created empty and incomplete; that is what carries
|
||||
# the sse-connect the browser uses to start the stream.
|
||||
assert messages[1].content == ""
|
||||
assert messages[1].complete is False
|
||||
assert "sse-connect" in response.text
|
||||
|
||||
|
||||
def test_empty_message_is_ignored(client: TestClient, db, registered):
|
||||
_add_connection(db)
|
||||
chat_id = client.post("/api/chats").headers["HX-Redirect"].rsplit("/", 1)[1]
|
||||
assert client.post(f"/api/chats/{chat_id}/messages", data={"content": " "}).status_code == 204
|
||||
assert db.scalar(select(Message)) is None
|
||||
|
||||
|
||||
def test_a_chat_belonging_to_someone_else_is_not_found(client: TestClient, db, registered):
|
||||
_add_connection(db)
|
||||
chat_id = client.post("/api/chats").headers["HX-Redirect"].rsplit("/", 1)[1]
|
||||
|
||||
client.post("/auth/logout", follow_redirects=False)
|
||||
client.post(
|
||||
"/auth/register",
|
||||
data={"name": "Sam", "email": "sam@shire.test", "password": "potatoes-po-ta-toes"},
|
||||
follow_redirects=False,
|
||||
)
|
||||
# 404 and not 403: whether that id exists is not this endpoint's to reveal.
|
||||
assert client.get(f"/chat/{chat_id}").status_code == 404
|
||||
|
||||
|
||||
def test_renaming_a_chat_stops_it_being_auto_titled(client: TestClient, db, registered):
|
||||
_add_connection(db)
|
||||
chat_id = client.post("/api/chats").headers["HX-Redirect"].rsplit("/", 1)[1]
|
||||
client.patch(f"/api/chats/{chat_id}", data={"title": "My own title"})
|
||||
|
||||
chat = db.get(Chat, chat_id)
|
||||
db.refresh(chat)
|
||||
assert chat.title == "My own title"
|
||||
assert chat.title_generated is True
|
||||
|
||||
|
||||
def test_deleting_a_chat_removes_its_messages(client: TestClient, db, registered):
|
||||
_add_connection(db)
|
||||
chat_id = client.post("/api/chats").headers["HX-Redirect"].rsplit("/", 1)[1]
|
||||
client.post(f"/api/chats/{chat_id}/messages", data={"content": "Hello"})
|
||||
|
||||
client.delete(f"/api/chats/{chat_id}")
|
||||
assert db.scalar(select(Chat)) is None
|
||||
assert db.scalar(select(Message)) is None
|
||||
|
||||
|
||||
def test_deleting_a_folder_keeps_the_chats_inside_it(client: TestClient, db, registered):
|
||||
"""Losing a conversation to a mis-clicked folder delete is unforgivable."""
|
||||
_add_connection(db)
|
||||
client.post("/api/folders", data={"name": "Quests"})
|
||||
folder = db.scalar(select(Folder))
|
||||
|
||||
chat_id = client.post("/api/chats").headers["HX-Redirect"].rsplit("/", 1)[1]
|
||||
client.patch(f"/api/chats/{chat_id}", data={"folder_id": folder.id})
|
||||
|
||||
client.delete(f"/api/folders/{folder.id}")
|
||||
|
||||
chat = db.get(Chat, chat_id)
|
||||
db.refresh(chat)
|
||||
assert chat is not None
|
||||
assert chat.folder_id is None
|
||||
|
||||
|
||||
def test_a_folder_cannot_be_moved_inside_itself(client: TestClient, db, registered):
|
||||
client.post("/api/folders", data={"name": "Outer"})
|
||||
folder = db.scalar(select(Folder))
|
||||
response = client.patch(f"/api/folders/{folder.id}", data={"parent_id": folder.id})
|
||||
assert response.status_code == 400
|
||||
|
||||
|
||||
# --- Request building --------------------------------------------------------
|
||||
def test_request_forwards_only_known_sampling_parameters(db, user_id):
|
||||
"""A stray key in params_json must not become a 400 from the provider that
|
||||
looks like a LLeMbas bug."""
|
||||
connection = _add_connection(db)
|
||||
chat = Chat(
|
||||
user_id=user_id,
|
||||
model_id="test-model",
|
||||
connection_id=connection.id,
|
||||
params_json={"temperature": 0.4, "nonsense": "drop me"},
|
||||
)
|
||||
db.add(chat)
|
||||
db.commit()
|
||||
|
||||
payload = chat_service.build_request(db, chat)
|
||||
assert payload["temperature"] == 0.4
|
||||
assert "nonsense" not in payload
|
||||
|
||||
|
||||
def test_history_skips_failed_and_empty_turns(db, user_id):
|
||||
"""Sending an empty assistant message upsets several providers."""
|
||||
connection = _add_connection(db)
|
||||
chat = Chat(user_id=user_id, model_id="test-model", connection_id=connection.id)
|
||||
db.add(chat)
|
||||
db.commit()
|
||||
|
||||
db.add_all(
|
||||
[
|
||||
Message(chat_id=chat.id, role="user", content="one"),
|
||||
Message(chat_id=chat.id, role="assistant", content="", error="boom"),
|
||||
Message(chat_id=chat.id, role="user", content="two"),
|
||||
]
|
||||
)
|
||||
db.commit()
|
||||
|
||||
contents = [m["content"] for m in chat_service.build_request(db, chat)["messages"]]
|
||||
assert contents == ["one", "two"]
|
||||
|
||||
|
||||
def test_system_prompt_leads_the_message_list(db, user_id):
|
||||
connection = _add_connection(db)
|
||||
chat = Chat(
|
||||
user_id=user_id,
|
||||
model_id="test-model",
|
||||
connection_id=connection.id,
|
||||
system_prompt="You are terse.",
|
||||
)
|
||||
db.add(chat)
|
||||
db.commit()
|
||||
|
||||
messages = chat_service.build_request(db, chat)["messages"]
|
||||
assert messages[0] == {"role": "system", "content": "You are terse."}
|
||||
|
||||
|
||||
def test_streaming_reports_an_unreachable_endpoint_in_the_thread(
|
||||
client: TestClient, db, registered
|
||||
):
|
||||
"""A failed turn must never be an unexplained blank bubble."""
|
||||
_add_connection(db) # points at 127.0.0.1:1, which refuses connections
|
||||
chat_id = client.post("/api/chats").headers["HX-Redirect"].rsplit("/", 1)[1]
|
||||
client.post(f"/api/chats/{chat_id}/messages", data={"content": "Hello"})
|
||||
message = db.scalar(select(Message).where(Message.role == "assistant"))
|
||||
|
||||
response = client.get(f"/api/chats/{chat_id}/messages/{message.id}/stream")
|
||||
assert response.status_code == 200
|
||||
assert "Could not reach" in response.text
|
||||
assert "alert--error" in response.text
|
||||
|
||||
db.refresh(message)
|
||||
assert message.complete is True
|
||||
assert message.error
|
||||
@@ -0,0 +1,67 @@
|
||||
"""Secret encryption and password hashing."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from lembas.security.passwords import (
|
||||
hash_password,
|
||||
validate_password,
|
||||
verify_password,
|
||||
)
|
||||
from lembas.services.crypto import decrypt, encrypt, mask
|
||||
|
||||
|
||||
def test_encrypt_round_trip():
|
||||
secret = "sk-proj-abcdef1234567890"
|
||||
assert decrypt(encrypt(secret)) == secret
|
||||
|
||||
|
||||
def test_encrypt_is_not_reversible_by_eye():
|
||||
secret = "sk-proj-abcdef1234567890"
|
||||
assert secret not in encrypt(secret)
|
||||
|
||||
|
||||
def test_encrypt_is_randomised():
|
||||
"""Fernet includes a random IV, so the same input must not repeat."""
|
||||
assert encrypt("same") != encrypt("same")
|
||||
|
||||
|
||||
def test_empty_secret_stays_empty():
|
||||
"""Endpoints that need no key store nothing, not an encrypted blank."""
|
||||
assert encrypt("") == ""
|
||||
assert decrypt("") == ""
|
||||
|
||||
|
||||
def test_decrypt_fails_soft_on_garbage():
|
||||
"""An unreadable value means the secret key changed. The admin UI has to
|
||||
stay usable so the key can simply be re-entered."""
|
||||
assert decrypt("not-a-valid-token") == ""
|
||||
|
||||
|
||||
def test_mask_keeps_enough_to_identify_but_not_to_use():
|
||||
masked = mask("sk-proj-abcdef1234567890")
|
||||
assert masked.startswith("sk-")
|
||||
assert masked.endswith("7890")
|
||||
assert "abcdef" not in masked
|
||||
|
||||
|
||||
def test_mask_hides_short_secrets_entirely():
|
||||
assert set(mask("short")) == {"*"}
|
||||
|
||||
|
||||
def test_password_round_trip():
|
||||
stored = hash_password("speak-friend")
|
||||
assert verify_password("speak-friend", stored)
|
||||
assert not verify_password("Speak-Friend", stored)
|
||||
|
||||
|
||||
def test_password_hash_is_salted():
|
||||
assert hash_password("same") != hash_password("same")
|
||||
|
||||
|
||||
def test_verify_rejects_a_corrupt_hash_instead_of_raising():
|
||||
assert not verify_password("anything", "not-a-hash")
|
||||
|
||||
|
||||
def test_short_passwords_are_rejected():
|
||||
assert validate_password("short") is not None
|
||||
assert validate_password("long-enough-password") is None
|
||||
@@ -0,0 +1,97 @@
|
||||
"""Markdown rendering and sanitisation.
|
||||
|
||||
Model output is untrusted input: it routinely contains HTML and a model can be
|
||||
talked into emitting a script tag. These are the tests that keep that boundary.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from lembas.services.markdown import escape_text, render_markdown
|
||||
|
||||
|
||||
def test_basic_formatting():
|
||||
html = render_markdown("Some **bold** and *italic* text.")
|
||||
assert "<strong>bold</strong>" in html
|
||||
assert "<em>italic</em>" in html
|
||||
|
||||
|
||||
def test_script_tags_are_stripped():
|
||||
html = render_markdown("Hello <script>alert('xss')</script> world")
|
||||
assert "<script" not in html
|
||||
assert "alert" not in html or "<script" in html
|
||||
|
||||
|
||||
def test_javascript_urls_never_become_links():
|
||||
"""markdown-it refuses the scheme and leaves the source as literal text, so
|
||||
the guarantee to assert is that no anchor is produced -- not that the
|
||||
substring is absent, which it legitimately is not."""
|
||||
html = render_markdown("[click me](javascript:alert(1))")
|
||||
assert "<a " not in html
|
||||
assert 'href="javascript:' not in html
|
||||
|
||||
|
||||
def test_javascript_urls_in_raw_html_anchors_are_stripped():
|
||||
html = render_markdown('<a href="javascript:alert(1)">click</a>')
|
||||
assert 'href="javascript:' not in html
|
||||
|
||||
|
||||
def test_event_handlers_are_stripped():
|
||||
html = render_markdown('<img src="x" onerror="alert(1)">')
|
||||
assert "onerror" not in html
|
||||
|
||||
|
||||
def test_external_links_get_protective_rel():
|
||||
html = render_markdown("[example](https://example.com)")
|
||||
assert "noopener" in html
|
||||
assert "noreferrer" in html
|
||||
|
||||
|
||||
def test_code_block_is_highlighted_and_not_double_wrapped():
|
||||
html = render_markdown("```python\ndef f():\n return 1\n```")
|
||||
assert 'class="code-block"' in html
|
||||
assert "pg-k" in html # a Pygments keyword span
|
||||
# markdown-it wraps highlight output in its own <pre><code> unless the
|
||||
# fence rule is replaced outright. This is the regression guard.
|
||||
assert "<pre><code" not in html
|
||||
|
||||
|
||||
def test_code_block_language_label():
|
||||
assert ">python<" in render_markdown("```python\nx = 1\n```")
|
||||
|
||||
|
||||
def test_unlabelled_code_block_still_renders():
|
||||
html = render_markdown("```\njust text\n```")
|
||||
assert 'class="code-block"' in html
|
||||
assert "just text" in html
|
||||
|
||||
|
||||
def test_code_content_is_escaped():
|
||||
html = render_markdown("```\n<script>alert(1)</script>\n```")
|
||||
assert "<script>" not in html
|
||||
|
||||
|
||||
def test_tables_render():
|
||||
html = render_markdown("| a | b |\n|---|---|\n| 1 | 2 |")
|
||||
assert "<table>" in html and "<td>1</td>" in html
|
||||
|
||||
|
||||
def test_bare_urls_are_linkified():
|
||||
assert "<a href=" in render_markdown("see https://example.com for more")
|
||||
|
||||
|
||||
def test_empty_input():
|
||||
assert render_markdown("") == ""
|
||||
|
||||
|
||||
def test_escape_text_handles_structural_characters():
|
||||
assert escape_text("<b>hi</b>") == "<b>hi</b>"
|
||||
assert escape_text("a & b") == "a & b"
|
||||
|
||||
|
||||
def test_escape_text_is_chunk_safe():
|
||||
"""Streaming escapes each token as it arrives, so escaping the pieces must
|
||||
equal escaping the whole -- otherwise a stream would diverge from the final
|
||||
rendering."""
|
||||
whole = "<a>&</a> text"
|
||||
chunks = ["<a", ">&<", "/a> ", "text"]
|
||||
assert "".join(escape_text(c) for c in chunks) == escape_text(whole)
|
||||
Reference in New Issue
Block a user