siftlode/backend/tests/test_files_preview.py

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