Homer
59739cc7fd
The second audit pass. Four things, and the first two were reported. The Prompts page put a screen of variables and a screen of preview above the editor, so the tabs began two screens down and switching one had to drag the whole page to be any use -- and on a short tab it could not drag far enough, leaving the panel stranded above a screenful of nothing. Editor first, reference after, bar sticky. Custom themes were three fixed slots: fifty-seven empty colour boxes on a fresh instance and no way to make a fourth theme. One block per theme plus a blank one, colours behind a disclosure. Both measured rather than argued about -- rendered through TestClient and driven under headless Chromium, where the tab bar moved 385->642px before and does not move now, and the themes page went from 5495px to 2820px. Asking where generated images go found the other two. Deleting a chat cascades to the attachment rows and leaves every file on disk; the helper written for exactly that was called from one place, and it was not the delete button, a schedule's chat, a helper's chat or deleting an account. Underneath it, `claim` bound message_id and never chat_id, so anything picked before a chat existed kept an empty chat_id forever -- which six readers filter on, so those files were also unnamed in the prompt, unopenable in the canvas, and invisible to the one caller the cleanup had. And folders nest now. The route has handled parent_id since folders existed, with a cycle guard and a depth cap the move path never applied; the sidebar has always drawn a tree. Nothing could ask for one. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
735 lines
29 KiB
Python
735 lines
29 KiB
Python
"""Attachments: upload validation, extraction, and how they reach the model."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import io
|
|
from datetime import timedelta
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
from PIL import Image
|
|
from sqlalchemy import select
|
|
|
|
from lembas.db.models import Attachment, Chat, Connection, Message, Model
|
|
from lembas.services import chat as chat_service
|
|
from lembas.services import files as files_service
|
|
from lembas.services import settings_store
|
|
from lembas.services.crypto import encrypt
|
|
|
|
|
|
def turns(body: dict) -> list[dict]:
|
|
"""The conversation turns, without the system preamble in front of them.
|
|
|
|
`build_request` always emits a system message now -- the harness carries the
|
|
date even for a model with no tools -- so a test about a *user* turn has to
|
|
say which turn it means rather than assume index 0.
|
|
"""
|
|
return [message for message in body["messages"] if message["role"] != "system"]
|
|
|
|
|
|
# --- Fixtures ----------------------------------------------------------------
|
|
def png_bytes(width: int = 40, height: int = 30, mode: str = "RGB") -> bytes:
|
|
buffer = io.BytesIO()
|
|
Image.new(mode, (width, height), "red").save(buffer, format="PNG")
|
|
return buffer.getvalue()
|
|
|
|
|
|
def jpeg_bytes(width: int = 40, height: int = 30) -> bytes:
|
|
buffer = io.BytesIO()
|
|
Image.new("RGB", (width, height), "blue").save(buffer, format="JPEG")
|
|
return buffer.getvalue()
|
|
|
|
|
|
def pdf_bytes(pages: list[str]) -> bytes:
|
|
"""A real PDF with a text layer, built with pypdf + reportlab-free drawing.
|
|
|
|
pypdf cannot author text, so the file is assembled by hand. It is minimal
|
|
but genuinely parseable, which is the point -- a fake would not exercise
|
|
extraction at all.
|
|
"""
|
|
objects: list[bytes] = []
|
|
|
|
def stream_for(text: str) -> bytes:
|
|
escaped = text.replace("\\", r"\\").replace("(", r"\(").replace(")", r"\)")
|
|
return f"BT /F1 12 Tf 72 720 Td ({escaped}) Tj ET".encode("latin-1")
|
|
|
|
page_ids = [4 + i * 2 for i in range(len(pages))]
|
|
kids = " ".join(f"{pid} 0 R" for pid in page_ids)
|
|
|
|
objects.append(b"<< /Type /Catalog /Pages 2 0 R >>")
|
|
objects.append(f"<< /Type /Pages /Kids [{kids}] /Count {len(pages)} >>".encode())
|
|
objects.append(b"<< /Type /Font /Subtype /Type1 /BaseFont /Helvetica >>")
|
|
|
|
for index, text in enumerate(pages):
|
|
content = stream_for(text)
|
|
objects.append(
|
|
f"<< /Type /Page /Parent 2 0 R /MediaBox [0 0 612 792] "
|
|
f"/Resources << /Font << /F1 3 0 R >> >> "
|
|
f"/Contents {page_ids[index] + 1} 0 R >>".encode()
|
|
)
|
|
objects.append(
|
|
b"<< /Length "
|
|
+ str(len(content)).encode()
|
|
+ b" >>\nstream\n"
|
|
+ content
|
|
+ b"\nendstream"
|
|
)
|
|
|
|
out = bytearray(b"%PDF-1.4\n")
|
|
offsets = [0]
|
|
for number, body in enumerate(objects, start=1):
|
|
offsets.append(len(out))
|
|
out += f"{number} 0 obj\n".encode() + body + b"\nendobj\n"
|
|
|
|
xref_at = len(out)
|
|
out += f"xref\n0 {len(objects) + 1}\n".encode()
|
|
out += b"0000000000 65535 f \n"
|
|
for offset in offsets[1:]:
|
|
out += f"{offset:010d} 00000 n \n".encode()
|
|
out += (
|
|
f"trailer\n<< /Size {len(objects) + 1} /Root 1 0 R >>\nstartxref\n{xref_at}\n%%EOF".encode()
|
|
)
|
|
return bytes(out)
|
|
|
|
|
|
@pytest.fixture
|
|
def chat_with_model(client: TestClient, db, registered, make_chat):
|
|
"""A chat whose model has vision turned on."""
|
|
connection = Connection(name="T", 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="seeing-model",
|
|
capabilities_json={"vision": True},
|
|
)
|
|
)
|
|
db.commit()
|
|
return make_chat()
|
|
|
|
|
|
# --- Type detection and processing -------------------------------------------
|
|
def test_png_is_recognised_as_an_image():
|
|
prepared = files_service.prepare(png_bytes(), "photo.png")
|
|
assert prepared.kind == "image"
|
|
assert prepared.width == 40 and prepared.height == 30
|
|
|
|
|
|
def test_large_images_are_downscaled():
|
|
"""A phone photo is megabytes of base64 and a large slice of the context."""
|
|
prepared = files_service.prepare(jpeg_bytes(4000, 3000), "big.jpg")
|
|
assert max(prepared.width, prepared.height) == files_service.MAX_IMAGE_EDGE
|
|
assert prepared.height == int(3000 * (files_service.MAX_IMAGE_EDGE / 4000))
|
|
|
|
|
|
def test_small_images_are_left_alone():
|
|
prepared = files_service.prepare(jpeg_bytes(100, 80), "small.jpg")
|
|
assert (prepared.width, prepared.height) == (100, 80)
|
|
|
|
|
|
def test_transparent_images_stay_png():
|
|
prepared = files_service.prepare(png_bytes(mode="RGBA"), "logo.png")
|
|
assert prepared.media_type in ("image/png", "image/jpeg")
|
|
|
|
|
|
def test_a_file_lying_about_its_type_is_judged_by_its_bytes():
|
|
"""The extension says PNG; the content is text, and content wins."""
|
|
prepared = files_service.prepare(b"just some words", "trick.png")
|
|
assert prepared.kind == "text"
|
|
|
|
|
|
def test_pdf_text_is_extracted():
|
|
prepared = files_service.prepare(pdf_bytes(["Lembas keeps a traveller going."]), "doc.pdf")
|
|
assert prepared.kind == "document"
|
|
assert prepared.pages == 1
|
|
assert "Lembas keeps a traveller going." in prepared.extracted_text
|
|
|
|
|
|
def test_multi_page_pdfs_are_labelled_by_page():
|
|
prepared = files_service.prepare(pdf_bytes(["First page here", "Second page here"]), "d.pdf")
|
|
assert prepared.pages == 2
|
|
assert "[page 1]" in prepared.extracted_text
|
|
assert "[page 2]" in prepared.extracted_text
|
|
|
|
|
|
def test_a_pdf_with_no_text_layer_says_so():
|
|
"""A scan otherwise looks like the model simply ignored the document."""
|
|
prepared = files_service.prepare(pdf_bytes([" "]), "scan.pdf")
|
|
assert prepared.extraction_error
|
|
assert "scanned" in prepared.extraction_error.lower()
|
|
|
|
|
|
def test_a_corrupt_pdf_is_stored_with_an_error_not_rejected():
|
|
prepared = files_service.prepare(b"%PDF-1.4\nthis is not really a pdf", "broken.pdf")
|
|
assert prepared.kind == "document"
|
|
assert prepared.extraction_error
|
|
|
|
|
|
def test_text_files_are_decoded():
|
|
prepared = files_service.prepare(b"line one\nline two", "notes.txt")
|
|
assert prepared.kind == "text"
|
|
assert prepared.extracted_text == "line one\nline two"
|
|
|
|
|
|
def test_source_files_keep_a_sensible_media_type():
|
|
assert files_service.prepare(b"print('hi')", "x.py").media_type == "text/x-python"
|
|
|
|
|
|
def test_binary_files_are_rejected():
|
|
with pytest.raises(files_service.FileError):
|
|
files_service.prepare(b"\x00\x01\x02\x03" * 100, "mystery.bin")
|
|
|
|
|
|
def test_empty_files_are_rejected():
|
|
with pytest.raises(files_service.FileError):
|
|
files_service.prepare(b"", "empty.txt")
|
|
|
|
|
|
def test_oversized_files_are_rejected():
|
|
with pytest.raises(files_service.FileError) as caught:
|
|
files_service.prepare(b"x" * (files_service.limits().max_upload_bytes + 1), "huge.txt")
|
|
assert "MB" in str(caught.value)
|
|
|
|
|
|
def test_extracted_text_is_capped(monkeypatch):
|
|
"""Through the setting rather than the constant. The constant is only the
|
|
default now; what `prepare` reads is the snapshot, which is the thing that
|
|
would have gone on returning 120,000 if the wiring were wrong."""
|
|
monkeypatch.setattr(files_service, "_LIMITS", files_service.Limits(max_extracted_chars=50))
|
|
prepared = files_service.prepare(b"x" * 500, "long.txt")
|
|
assert len(prepared.extracted_text) == 50
|
|
assert prepared.truncated is True
|
|
|
|
|
|
# --- Path safety -------------------------------------------------------------
|
|
@pytest.mark.parametrize(
|
|
"name", ["../../etc/passwd", "..\\windows", "/etc/passwd", ".hidden", ""]
|
|
)
|
|
def test_stored_path_refuses_traversal(name):
|
|
assert files_service.stored_path(name) is None
|
|
|
|
|
|
def test_stored_names_are_random_not_the_uploaders(client: TestClient, db, registered):
|
|
response = client.post(
|
|
"/api/files", files={"file": ("../../evil.txt", b"content", "text/plain")}
|
|
)
|
|
assert response.status_code == 200
|
|
attachment = db.scalar(select(Attachment))
|
|
assert "/" not in attachment.stored_name
|
|
assert ".." not in attachment.stored_name
|
|
# The display name is kept, but only as a label.
|
|
assert attachment.filename == "evil.txt"
|
|
|
|
|
|
# --- Upload through the API --------------------------------------------------
|
|
def test_upload_returns_a_chip(client: TestClient, db, registered):
|
|
response = client.post(
|
|
"/api/files", files={"file": ("notes.txt", b"hello there", "text/plain")}
|
|
)
|
|
assert response.status_code == 200
|
|
assert "notes.txt" in response.text
|
|
assert 'name="file_ids"' in response.text
|
|
|
|
|
|
def test_a_rejected_upload_returns_a_readable_error(client: TestClient, db, registered):
|
|
response = client.post(
|
|
"/api/files", files={"file": ("bad.bin", b"\x00\x01" * 500, "application/octet-stream")}
|
|
)
|
|
assert response.status_code == 200
|
|
assert "chip--error" in response.text
|
|
assert db.scalar(select(Attachment)) is None
|
|
|
|
|
|
def test_uploading_needs_permission(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,
|
|
)
|
|
settings_store.update(db, {"default_permissions": {"files.upload": False}})
|
|
response = client.post("/api/files", files={"file": ("a.txt", b"hi", "text/plain")})
|
|
assert response.status_code == 403
|
|
|
|
|
|
def test_you_cannot_fetch_someone_elses_file(client: TestClient, db, registered):
|
|
client.post("/api/files", files={"file": ("secret.txt", b"mine", "text/plain")})
|
|
attachment = db.scalar(select(Attachment))
|
|
|
|
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(f"/api/files/{attachment.id}/content").status_code == 404
|
|
|
|
|
|
def test_non_images_are_served_as_downloads(client: TestClient, db, registered):
|
|
"""An uploaded .html served inline would run in this origin."""
|
|
client.post("/api/files", files={"file": ("page.html", b"<b>hi</b>", "text/html")})
|
|
attachment = db.scalar(select(Attachment))
|
|
|
|
response = client.get(f"/api/files/{attachment.id}/content")
|
|
assert "attachment;" in response.headers["content-disposition"]
|
|
assert response.headers["content-type"] == "application/octet-stream"
|
|
assert response.headers["x-content-type-options"] == "nosniff"
|
|
|
|
|
|
def test_images_are_served_inline(client: TestClient, db, registered):
|
|
client.post("/api/files", files={"file": ("p.png", png_bytes(), "image/png")})
|
|
attachment = db.scalar(select(Attachment))
|
|
response = client.get(f"/api/files/{attachment.id}/content")
|
|
assert "inline;" in response.headers["content-disposition"]
|
|
|
|
|
|
def test_an_unsent_attachment_can_be_removed(client: TestClient, db, registered):
|
|
client.post("/api/files", files={"file": ("a.txt", b"hi", "text/plain")})
|
|
attachment = db.scalar(select(Attachment))
|
|
stored = files_service.stored_path(attachment.stored_name)
|
|
|
|
assert client.delete(f"/api/files/{attachment.id}").status_code == 200
|
|
assert db.scalar(select(Attachment)) is None
|
|
assert not stored.exists()
|
|
|
|
|
|
def test_a_sent_attachment_cannot_be_removed(client: TestClient, db, chat_with_model):
|
|
"""Deleting it would rewrite a conversation the user has already read."""
|
|
client.post("/api/files", files={"file": ("a.txt", b"hi", "text/plain")})
|
|
attachment = db.scalar(select(Attachment))
|
|
client.post(
|
|
f"/api/chats/{chat_with_model}/messages",
|
|
data={"content": "look", "file_ids": [attachment.id]},
|
|
)
|
|
assert client.delete(f"/api/files/{attachment.id}").status_code == 409
|
|
|
|
|
|
# --- Attaching to a message --------------------------------------------------
|
|
def test_sending_binds_the_attachment_to_the_message(client: TestClient, db, chat_with_model):
|
|
client.post("/api/files", files={"file": ("a.txt", b"hi", "text/plain")})
|
|
attachment = db.scalar(select(Attachment))
|
|
|
|
client.post(
|
|
f"/api/chats/{chat_with_model}/messages",
|
|
data={"content": "have a look", "file_ids": [attachment.id]},
|
|
)
|
|
db.refresh(attachment)
|
|
message = db.scalar(select(Message).where(Message.role == "user"))
|
|
assert attachment.message_id == message.id
|
|
|
|
|
|
def test_a_message_with_only_an_attachment_is_accepted(client: TestClient, db, chat_with_model):
|
|
""""Here, look at this" with no words is a legitimate turn."""
|
|
client.post("/api/files", files={"file": ("a.png", png_bytes(), "image/png")})
|
|
attachment = db.scalar(select(Attachment))
|
|
|
|
response = client.post(
|
|
f"/api/chats/{chat_with_model}/messages",
|
|
data={"content": "", "file_ids": [attachment.id]},
|
|
)
|
|
assert response.status_code == 200
|
|
assert db.scalar(select(Message).where(Message.role == "user")) is not None
|
|
|
|
|
|
def test_a_truly_empty_message_is_still_ignored(client: TestClient, db, chat_with_model):
|
|
response = client.post(f"/api/chats/{chat_with_model}/messages", data={"content": " "})
|
|
assert response.status_code == 204
|
|
assert db.scalar(select(Message)) is None
|
|
|
|
|
|
def test_you_cannot_attach_someone_elses_file(
|
|
client: TestClient, db, chat_with_model, make_chat
|
|
):
|
|
"""A forged id must not pull another user's file into a conversation."""
|
|
client.post("/api/files", files={"file": ("mine.txt", b"secret", "text/plain")})
|
|
stolen = db.scalar(select(Attachment))
|
|
|
|
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,
|
|
)
|
|
their_chat = make_chat(email="sam@shire.test")
|
|
client.post(
|
|
f"/api/chats/{their_chat}/messages",
|
|
data={"content": "gimme", "file_ids": [stolen.id]},
|
|
)
|
|
|
|
db.refresh(stolen)
|
|
assert stolen.message_id is None
|
|
|
|
|
|
# --- What reaches the model --------------------------------------------------
|
|
def _user_message(db, chat_id: str) -> Message:
|
|
return db.scalar(
|
|
select(Message).where(Message.chat_id == chat_id, Message.role == "user")
|
|
)
|
|
|
|
|
|
def test_images_become_multimodal_parts_for_a_vision_model(client: TestClient, db, chat_with_model):
|
|
client.post("/api/files", files={"file": ("p.png", png_bytes(), "image/png")})
|
|
attachment = db.scalar(select(Attachment))
|
|
client.post(
|
|
f"/api/chats/{chat_with_model}/messages",
|
|
data={"content": "what is this?", "file_ids": [attachment.id]},
|
|
)
|
|
|
|
chat = db.get(Chat, chat_with_model)
|
|
payload = chat_service.build_request(db, chat)
|
|
content = turns(payload)[0]["content"]
|
|
|
|
assert isinstance(content, list)
|
|
assert content[0] == {"type": "text", "text": "what is this?"}
|
|
assert content[1]["type"] == "image_url"
|
|
assert content[1]["image_url"]["url"].startswith("data:image/")
|
|
|
|
|
|
def test_images_are_withheld_from_a_model_without_vision(client: TestClient, db, chat_with_model):
|
|
"""Most endpoints reject the whole request rather than ignoring the image."""
|
|
model = db.scalar(select(Model))
|
|
model.capabilities_json = {"vision": False}
|
|
db.commit()
|
|
|
|
client.post("/api/files", files={"file": ("p.png", png_bytes(), "image/png")})
|
|
attachment = db.scalar(select(Attachment))
|
|
client.post(
|
|
f"/api/chats/{chat_with_model}/messages",
|
|
data={"content": "what is this?", "file_ids": [attachment.id]},
|
|
)
|
|
|
|
chat = db.get(Chat, chat_with_model)
|
|
content = turns(chat_service.build_request(db, chat))[0]["content"]
|
|
assert isinstance(content, str)
|
|
assert content == "what is this?"
|
|
|
|
|
|
def test_a_plain_turn_stays_a_plain_string(client: TestClient, db, chat_with_model):
|
|
"""The list form is a reliable 400 from endpoints that do not implement it."""
|
|
client.post(f"/api/chats/{chat_with_model}/messages", data={"content": "just words"})
|
|
chat = db.get(Chat, chat_with_model)
|
|
assert turns(chat_service.build_request(db, chat))[0]["content"] == "just words"
|
|
|
|
|
|
def test_document_text_is_prepended_in_tags(client: TestClient, db, chat_with_model):
|
|
client.post(
|
|
"/api/files", files={"file": ("report.txt", b"Quarterly results were good.", "text/plain")}
|
|
)
|
|
attachment = db.scalar(select(Attachment))
|
|
client.post(
|
|
f"/api/chats/{chat_with_model}/messages",
|
|
data={"content": "summarise this", "file_ids": [attachment.id]},
|
|
)
|
|
|
|
chat = db.get(Chat, chat_with_model)
|
|
content = turns(chat_service.build_request(db, chat))[0]["content"]
|
|
assert '<document name="report.txt">' in content
|
|
assert "Quarterly results were good." in content
|
|
# The question comes after the material it refers to.
|
|
assert content.index("</document>") < content.index("summarise this")
|
|
|
|
|
|
def test_documents_reach_a_model_without_vision(client: TestClient, db, chat_with_model):
|
|
model = db.scalar(select(Model))
|
|
model.capabilities_json = {}
|
|
db.commit()
|
|
|
|
client.post("/api/files", files={"file": ("notes.txt", b"important detail", "text/plain")})
|
|
attachment = db.scalar(select(Attachment))
|
|
client.post(
|
|
f"/api/chats/{chat_with_model}/messages",
|
|
data={"content": "read it", "file_ids": [attachment.id]},
|
|
)
|
|
|
|
chat = db.get(Chat, chat_with_model)
|
|
content = turns(chat_service.build_request(db, chat))[0]["content"]
|
|
assert "important detail" in content
|
|
|
|
|
|
def test_truncation_is_declared_to_the_model(client: TestClient, db, chat_with_model, monkeypatch):
|
|
"""A model asked about page 400 should be able to say it did not see it."""
|
|
monkeypatch.setattr(files_service, "_LIMITS", files_service.Limits(max_extracted_chars=20))
|
|
client.post("/api/files", files={"file": ("big.txt", b"y" * 200, "text/plain")})
|
|
attachment = db.scalar(select(Attachment))
|
|
client.post(
|
|
f"/api/chats/{chat_with_model}/messages",
|
|
data={"content": "read", "file_ids": [attachment.id]},
|
|
)
|
|
|
|
chat = db.get(Chat, chat_with_model)
|
|
assert "(truncated)" in turns(chat_service.build_request(db, chat))[0]["content"]
|
|
|
|
|
|
def test_an_attachment_only_turn_still_reaches_the_model(client: TestClient, db, chat_with_model):
|
|
client.post("/api/files", files={"file": ("p.png", png_bytes(), "image/png")})
|
|
attachment = db.scalar(select(Attachment))
|
|
client.post(
|
|
f"/api/chats/{chat_with_model}/messages", data={"content": "", "file_ids": [attachment.id]}
|
|
)
|
|
|
|
chat = db.get(Chat, chat_with_model)
|
|
messages = turns(chat_service.build_request(db, chat))
|
|
assert len(messages) == 1
|
|
assert messages[0]["content"][0]["type"] == "image_url"
|
|
|
|
|
|
# --- Housekeeping ------------------------------------------------------------
|
|
def test_orphans_are_swept(client: TestClient, db, registered):
|
|
client.post("/api/files", files={"file": ("a.txt", b"hi", "text/plain")})
|
|
attachment = db.scalar(select(Attachment))
|
|
path = files_service.stored_path(attachment.stored_name)
|
|
|
|
assert files_service.sweep_orphans(db, timedelta(hours=24)) == 0 # too new
|
|
assert files_service.sweep_orphans(db, timedelta(seconds=-1)) == 1
|
|
assert db.scalar(select(Attachment)) is None
|
|
assert not path.exists()
|
|
|
|
|
|
def test_sent_attachments_are_never_swept(client: TestClient, db, chat_with_model):
|
|
client.post("/api/files", files={"file": ("a.txt", b"hi", "text/plain")})
|
|
attachment = db.scalar(select(Attachment))
|
|
client.post(
|
|
f"/api/chats/{chat_with_model}/messages",
|
|
data={"content": "keep", "file_ids": [attachment.id]},
|
|
)
|
|
|
|
assert files_service.sweep_orphans(db, timedelta(seconds=-1)) == 0
|
|
assert db.scalar(select(Attachment)) is not None
|
|
|
|
|
|
def test_deleting_a_message_deletes_its_attachments(client: TestClient, db, chat_with_model):
|
|
client.post("/api/files", files={"file": ("a.txt", b"hi", "text/plain")})
|
|
attachment = db.scalar(select(Attachment))
|
|
client.post(
|
|
f"/api/chats/{chat_with_model}/messages",
|
|
data={"content": "x", "file_ids": [attachment.id]},
|
|
)
|
|
|
|
client.delete(f"/api/chats/{chat_with_model}")
|
|
assert db.scalar(select(Attachment)) is None
|
|
|
|
|
|
# --- Composer wiring ---------------------------------------------------------
|
|
# These assert on the rendered HTML rather than on behaviour, because the bug
|
|
# they guard against lives entirely in the template: every server-side test
|
|
# passed while the browser silently never sent file_ids at all.
|
|
def test_the_attachments_container_is_inside_the_composer_form(
|
|
client: TestClient, db, chat_with_model
|
|
):
|
|
"""The chips carry the hidden file_ids inputs. Outside the form they are
|
|
not serialised, and hx-include does not help -- it only has an effect on
|
|
the element issuing the request, not on a child of it."""
|
|
import re
|
|
|
|
page = client.get(f"/chat/{chat_with_model}").text
|
|
form = re.search(r'<form class="composer__form".*?</form>', page, re.S)
|
|
assert form, "composer form not found"
|
|
assert 'id="attachments"' in form.group(0)
|
|
|
|
|
|
def test_the_file_input_is_outside_the_composer_form(client: TestClient, db, chat_with_model):
|
|
"""Inside, it would be submitted as an empty file part on every message."""
|
|
import re
|
|
|
|
page = client.get(f"/chat/{chat_with_model}").text
|
|
form = re.search(r'<form class="composer__form".*?</form>', page, re.S)
|
|
assert 'id="file-input"' not in form.group(0)
|
|
assert 'id="file-input"' in page
|
|
|
|
|
|
def test_the_new_chat_composer_also_contains_the_attachments(
|
|
client: TestClient, db, chat_with_model
|
|
):
|
|
import re
|
|
|
|
page = client.get("/chat").text
|
|
form = re.search(r'<form class="composer__form".*?</form>', page, re.S)
|
|
assert form and 'id="attachments"' in form.group(0)
|
|
|
|
|
|
def test_the_chip_carries_a_file_ids_input(client: TestClient, db, registered):
|
|
"""That input is the entire mechanism by which an upload reaches a message."""
|
|
response = client.post(
|
|
"/api/files", files={"file": ("a.txt", b"hi", "text/plain")}
|
|
)
|
|
assert 'name="file_ids"' in response.text
|
|
assert 'type="hidden"' in response.text
|
|
|
|
|
|
def test_a_browser_serialising_the_form_actually_sends_the_attachment(
|
|
client: TestClient, db, chat_with_model
|
|
):
|
|
"""End-to-end wiring check.
|
|
|
|
Uploads a file, splices the returned chip into the page exactly as the
|
|
browser does, then serialises the composer form the way a browser would --
|
|
every named input inside <form> -- and posts that. This is the test that
|
|
fails when the chips drift back outside the form.
|
|
"""
|
|
import re
|
|
|
|
chip = client.post(
|
|
"/api/files", files={"file": ("proof.png", png_bytes(), "image/png")}
|
|
).text
|
|
attachment = db.scalar(select(Attachment))
|
|
assert attachment.message_id is None
|
|
|
|
page = client.get(f"/chat/{chat_with_model}").text
|
|
form_html = re.search(r'<form class="composer__form".*?</form>', page, re.S).group(0)
|
|
# The chips are inserted into #attachments, which lives inside the form.
|
|
form_html = form_html.replace(
|
|
'<div class="composer__attachments" id="attachments"></div>',
|
|
f'<div class="composer__attachments" id="attachments">{chip}</div>',
|
|
)
|
|
|
|
fields: list[tuple[str, str]] = []
|
|
for tag in re.findall(r"<(?:input|textarea)\b[^>]*>", form_html):
|
|
name = re.search(r'name="([^"]+)"', tag)
|
|
if not name:
|
|
continue
|
|
value = re.search(r'value="([^"]*)"', tag)
|
|
fields.append((name.group(1), value.group(1) if value else ""))
|
|
|
|
assert ("file_ids", attachment.id) in fields, f"file_ids not serialised: {fields}"
|
|
|
|
client.post(f"/api/chats/{chat_with_model}/messages", data=dict(fields) | {"content": "look"})
|
|
|
|
db.refresh(attachment)
|
|
message = db.scalar(select(Message).where(Message.role == "user"))
|
|
assert attachment.message_id == message.id
|
|
|
|
chat = db.get(Chat, chat_with_model)
|
|
content = turns(chat_service.build_request(db, chat))[0]["content"]
|
|
assert isinstance(content, list), "the image never reached the model"
|
|
assert any(p.get("type") == "image_url" for p in content)
|
|
|
|
|
|
# --- The extraction settings ---------------------------------------------------
|
|
# The constants above are defaults now, and what `prepare` actually reads is a
|
|
# process-level snapshot. Every one of these failures would be silent: a limit
|
|
# that looks configured and is not.
|
|
def test_a_saved_limit_reaches_the_snapshot(db, client, registered):
|
|
from lembas.services import settings_store
|
|
|
|
client.post(
|
|
"/admin/extraction",
|
|
data={
|
|
"max_upload_mb": "5",
|
|
"max_image_edge": "800",
|
|
"jpeg_quality": "70",
|
|
"max_pdf_pages": "10",
|
|
"max_extracted_chars": "2000",
|
|
"orphan_hours": "3",
|
|
"extra_text_extensions": "env\n.conf",
|
|
},
|
|
follow_redirects=False,
|
|
)
|
|
|
|
bounds = files_service.limits()
|
|
assert bounds.max_upload_bytes == 5 * 1024 * 1024
|
|
assert bounds.max_extracted_chars == 2000
|
|
assert bounds.max_image_edge == 800
|
|
assert settings_store.extraction(db)["extra_text_extensions"] == ["env", ".conf"]
|
|
|
|
|
|
def test_an_extra_extension_gets_a_leading_dot(db, client, registered):
|
|
"""Typed both ways by different people, and a mapping somebody has to get
|
|
right twice is one they get wrong once."""
|
|
client.post(
|
|
"/admin/extraction", data={"extra_text_extensions": "env"}, follow_redirects=False
|
|
)
|
|
|
|
assert files_service.limits().media_type_for(".env") == "text/plain"
|
|
|
|
|
|
def test_an_unknown_extension_is_still_stored_as_text(db):
|
|
"""Decodability is what decides. The list only picks a media type, which is
|
|
why an unlisted extension has always worked and must go on working."""
|
|
prepared = files_service.prepare(b"hello there", "notes.wat")
|
|
assert prepared.kind == "text"
|
|
assert prepared.extension == ".txt"
|
|
|
|
|
|
def test_a_number_out_of_range_is_clamped(db, client, registered):
|
|
client.post(
|
|
"/admin/extraction",
|
|
data={"max_upload_mb": "99999", "jpeg_quality": "1"},
|
|
follow_redirects=False,
|
|
)
|
|
|
|
bounds = files_service.limits()
|
|
assert bounds.max_upload_bytes == 512 * 1024 * 1024
|
|
assert bounds.jpeg_quality == 30
|
|
|
|
|
|
def test_the_snapshot_is_dropped_when_the_page_saves(db, client, registered):
|
|
"""Read once per process. A save that did not drop it would take effect at
|
|
the next restart, which is the failure this codebase keeps cataloguing."""
|
|
assert files_service.limits().max_upload_bytes == 20 * 1024 * 1024
|
|
|
|
client.post("/admin/extraction", data={"max_upload_mb": "1"}, follow_redirects=False)
|
|
|
|
assert files_service.limits().max_upload_bytes == 1024 * 1024
|
|
|
|
|
|
def test_only_an_administrator_may_change_extraction(client, 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.post("/admin/extraction", data={"max_upload_mb": "1"}).status_code == 403
|
|
assert client.post("/admin/extraction/search", data={}).status_code == 403
|
|
assert client.post("/admin/extraction/rebuild", data={}).status_code == 403
|
|
|
|
|
|
def test_deleting_a_chat_removes_the_files_it_held(
|
|
client: TestClient, db, chat_with_model
|
|
):
|
|
"""Deleting a Chat cascades to its message and attachment *rows* and leaves
|
|
every file on disk -- a generated image, an uploaded PDF -- with nothing that
|
|
will ever look at them again, since `sweep_orphans` only considers uploads
|
|
that were never attached.
|
|
|
|
`remove_files_for_chats` was written for this and was called from exactly one
|
|
place, the temporary-chat sweep. The delete button was not it. That is
|
|
`sharing.forget_principal` a third time: a correct helper, absent from the
|
|
path that needs it.
|
|
"""
|
|
client.post("/api/files", files={"file": ("a.png", png_bytes(), "image/png")})
|
|
attachment = db.scalar(select(Attachment))
|
|
path = files_service.stored_path(attachment.stored_name)
|
|
client.post(
|
|
f"/api/chats/{chat_with_model}/messages",
|
|
data={"content": "here", "file_ids": [attachment.id]},
|
|
)
|
|
assert path.exists()
|
|
|
|
client.delete(f"/api/chats/{chat_with_model}")
|
|
|
|
assert db.scalar(select(Attachment)) is None
|
|
assert not path.exists(), "the row went and the file stayed"
|
|
|
|
|
|
def test_delete_chats_is_what_every_path_uses(client: TestClient, db, chat_with_model):
|
|
"""One function, so the next thing that deletes a chat cannot forget. It has
|
|
to unlink *before* the rows go -- they are what says which files to remove."""
|
|
client.post("/api/files", files={"file": ("a.png", png_bytes(), "image/png")})
|
|
attachment = db.scalar(select(Attachment))
|
|
path = files_service.stored_path(attachment.stored_name)
|
|
client.post(
|
|
f"/api/chats/{chat_with_model}/messages",
|
|
data={"content": "here", "file_ids": [attachment.id]},
|
|
)
|
|
|
|
chat = db.get(Chat, chat_with_model)
|
|
assert chat_service.delete_chats(db, [chat]) == 1
|
|
db.commit()
|
|
|
|
assert not path.exists()
|
|
assert chat_service.delete_chats(db, []) == 0
|
|
assert chat_service.delete_chats(db, [None]) == 0
|