feat(files): a file that is not media opens in a viewer — an image, a PDF, text as text
- GET /api/files/{id}/preview: an allow-list on the stored extension (downloads/preview.py) picks
image / PDF / text; served inline under `CSP: sandbox; frame-ancestors 'self'`, the type from the
table; HTML and SVG arrive as plain text; a text is its first 1 MiB (X-Preview-Truncated), sent
as UTF-8 whatever it was written in (BOM, UTF-8, else cp1250); a NUL means binary: 415
- the list says which preview each file gets (asset.preview); FE/BE contract on kinds and header
- the viewer (shared dialog shell): ← / → through the list's such files in its order, Save, open in
a tab, the file's menu; one stage height for every kind; a no-preview card offers the file
- the name, the card's picture and the menu's Preview open it; an image up to 8 MB is its own
thumbnail
This commit is contained in:
@@ -0,0 +1,174 @@
|
||||
"""A library file's preview (`GET /api/files/{id}/preview`, `downloads/preview.py`), DB-lane.
|
||||
|
||||
What it may show is an allow-list on the stored extension: an image or a PDF inline, a text format
|
||||
(HTML and SVG among them) as plain text, its head only. Every answer is sandboxed and may be framed
|
||||
by the app alone; anything else, a binary "text" included, has no preview.
|
||||
"""
|
||||
|
||||
import pytest
|
||||
from sqlalchemy import select
|
||||
|
||||
from app.config import settings
|
||||
from app.downloads import folders, preview, service
|
||||
from app.downloads.status import JobStatus
|
||||
from app.models import DownloadJob, DownloadShare, MediaAsset
|
||||
from app.routes.files import TRUNCATED_HEADER
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def _root(tmp_path, monkeypatch):
|
||||
monkeypatch.setattr(settings, "files_root", str(tmp_path))
|
||||
return tmp_path
|
||||
|
||||
|
||||
def _upload(db, root, user_id, ext, body, *, media_kind="file", title="notes"):
|
||||
"""A done uploaded file named `title.ext` holding `body`."""
|
||||
n = len(db.execute(select(MediaAsset.id)).all())
|
||||
rel = f"up{n}.{ext}" if ext else f"up{n}"
|
||||
(root / rel).write_bytes(body)
|
||||
asset = MediaAsset(
|
||||
source_kind="upload",
|
||||
source_ref=f"up{n}",
|
||||
format_sig=f"sig-up{n}",
|
||||
status="ready",
|
||||
rel_path=rel,
|
||||
size_bytes=len(body),
|
||||
ref_count=0,
|
||||
title=title,
|
||||
media_kind=media_kind,
|
||||
container=ext,
|
||||
)
|
||||
db.add(asset)
|
||||
db.flush()
|
||||
job = DownloadJob(
|
||||
user_id=user_id,
|
||||
asset_id=asset.id,
|
||||
source_kind="upload",
|
||||
source_ref=f"up{n}",
|
||||
profile_snapshot={},
|
||||
status=JobStatus.DONE,
|
||||
)
|
||||
db.add(job)
|
||||
db.flush()
|
||||
folders.place(db, job)
|
||||
service.incr_ref_count(db, asset.id)
|
||||
db.commit()
|
||||
return job
|
||||
|
||||
|
||||
def _assert_sandboxed(r):
|
||||
assert r.headers["content-security-policy"] == "sandbox; frame-ancestors 'self'"
|
||||
assert r.headers["x-frame-options"] == "SAMEORIGIN"
|
||||
assert r.headers["x-content-type-options"] == "nosniff"
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"ext,ctype",
|
||||
[
|
||||
("png", "image/png"),
|
||||
("JPG", "image/jpeg"),
|
||||
("webp", "image/webp"),
|
||||
("pdf", "application/pdf"),
|
||||
],
|
||||
)
|
||||
def test_an_image_or_a_pdf_is_shown_inline_with_its_listed_type_in_a_sandbox(
|
||||
db, make_user, as_user, _root, ext, ctype
|
||||
):
|
||||
alice = make_user()
|
||||
job = _upload(db, _root, alice.id, ext, b"bytes of it")
|
||||
r = as_user(alice).get(f"/api/files/{job.id}/preview")
|
||||
assert r.status_code == 200 and r.content == b"bytes of it"
|
||||
assert r.headers["content-type"] == ctype
|
||||
assert r.headers["content-disposition"].startswith("inline;")
|
||||
assert f'filename="notes.{ext}"' in r.headers["content-disposition"]
|
||||
_assert_sandboxed(r)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("ext", ["html", "svg", "txt", "py"])
|
||||
def test_html_svg_and_code_arrive_as_plain_text_never_as_a_page(db, make_user, as_user, _root, ext):
|
||||
alice = make_user()
|
||||
body = b"<svg onload=x><script>y</script></svg>"
|
||||
job = _upload(db, _root, alice.id, ext, body)
|
||||
r = as_user(alice).get(f"/api/files/{job.id}/preview")
|
||||
assert r.status_code == 200 and r.content == body
|
||||
assert r.headers["content-type"] == "text/plain; charset=utf-8"
|
||||
assert TRUNCATED_HEADER.lower() not in r.headers
|
||||
_assert_sandboxed(r)
|
||||
|
||||
|
||||
def test_a_long_text_is_cut_to_its_head_and_says_so(db, make_user, as_user, _root):
|
||||
alice = make_user()
|
||||
body = b"a" * preview.TEXT_PREVIEW_BYTES + b"tail"
|
||||
job = _upload(db, _root, alice.id, "log", body)
|
||||
r = as_user(alice).get(f"/api/files/{job.id}/preview")
|
||||
assert r.status_code == 200 and len(r.content) == preview.TEXT_PREVIEW_BYTES
|
||||
assert r.headers[TRUNCATED_HEADER] == "1"
|
||||
exact = _upload(db, _root, alice.id, "log", b"a" * preview.TEXT_PREVIEW_BYTES)
|
||||
whole = as_user(alice).get(f"/api/files/{exact.id}/preview")
|
||||
assert len(whole.content) == preview.TEXT_PREVIEW_BYTES
|
||||
assert TRUNCATED_HEADER.lower() not in whole.headers
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"ext,body,media_kind",
|
||||
[
|
||||
("zip", b"PK\x03\x04", "file"), # not on the list
|
||||
("docx", b"PK\x03\x04", "file"),
|
||||
(None, b"plain", "file"), # no extension
|
||||
("txt", b"text\x00with a nul", "file"), # a "text" that is binary
|
||||
("mp4", b"video", "video"), # media plays in the player instead
|
||||
("png", b"png", "video"), # an extension alone never makes media a preview
|
||||
],
|
||||
)
|
||||
def test_anything_else_has_no_preview(db, make_user, as_user, _root, ext, body, media_kind):
|
||||
alice = make_user()
|
||||
job = _upload(db, _root, alice.id, ext, body, media_kind=media_kind)
|
||||
r = as_user(alice).get(f"/api/files/{job.id}/preview")
|
||||
assert r.status_code == 415
|
||||
assert r.json()["detail"] == {"code": "refused", "reason": "file_no_preview"}
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"body",
|
||||
[
|
||||
"Árvíztűrő tükörfúrógép".encode("cp1250"), # an old Hungarian subtitle
|
||||
"Árvíztűrő tükörfúrógép".encode("utf-8-sig"),
|
||||
"Árvíztűrő tükörfúrógép".encode("utf-16"), # a BOM, then NULs: text all the same
|
||||
"Árvíztűrő tükörfúrógép".encode("utf-8"),
|
||||
],
|
||||
)
|
||||
def test_a_text_arrives_as_utf8_whatever_it_was_written_in(db, make_user, as_user, _root, body):
|
||||
alice = make_user()
|
||||
job = _upload(db, _root, alice.id, "srt", body)
|
||||
r = as_user(alice).get(f"/api/files/{job.id}/preview")
|
||||
assert r.status_code == 200
|
||||
assert r.content.decode("utf-8") == "Árvíztűrő tükörfúrógép"
|
||||
|
||||
|
||||
def test_a_character_cut_at_the_cap_keeps_the_text_utf8(db, make_user, as_user, _root):
|
||||
alice = make_user()
|
||||
body = b"a" * (preview.TEXT_PREVIEW_BYTES - 1) + "ő".encode() # its 2nd byte past the cap
|
||||
job = _upload(db, _root, alice.id, "txt", body)
|
||||
r = as_user(alice).get(f"/api/files/{job.id}/preview")
|
||||
assert r.content == body[: preview.TEXT_PREVIEW_BYTES] # not read as cp1250
|
||||
|
||||
|
||||
def test_the_list_says_which_preview_each_file_gets(db, make_user, as_user, _root):
|
||||
alice = make_user()
|
||||
for ext in ["png", "pdf", "html", "zip"]:
|
||||
_upload(db, _root, alice.id, ext, b"x", title=ext)
|
||||
_upload(db, _root, alice.id, "mp4", b"x", media_kind="video", title="mp4")
|
||||
got = {
|
||||
f["asset"]["container"]: f["asset"]["preview"]
|
||||
for f in as_user(alice).get("/api/files").json()
|
||||
}
|
||||
assert got == {"png": "image", "pdf": "pdf", "html": "text", "zip": None, "mp4": None}
|
||||
|
||||
|
||||
def test_a_preview_is_the_holders_or_a_friends_it_was_shared_with(db, make_user, as_user, _root):
|
||||
alice, friend = make_user(), make_user()
|
||||
job = _upload(db, _root, alice.id, "png", b"img")
|
||||
assert as_user(make_user()).get(f"/api/files/{job.id}/preview").status_code == 404
|
||||
db.add(DownloadShare(job_id=job.id, from_user_id=alice.id, to_user_id=friend.id))
|
||||
db.commit()
|
||||
assert as_user(friend).get(f"/api/files/{job.id}/preview").status_code == 200
|
||||
Reference in New Issue
Block a user