"""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.downloads.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"" 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