siftlode/backend/tests/test_upload.py

"""E7 S5a — local file upload ingest (the non-yt-dlp path into the media library).

Covers the ingest spine: media vs generic-file classification, the download-only serving of a
generic file (force-download so uploaded HTML/SVG can't execute inline), quota accounting with the
size known up front, the per-file cap enforced while streaming, filename sanitization, and the two
guards that keep an upload out of paths that assume a re-fetchable source (edit / redownload).

DB-backed (real Postgres) — skipped when no database is reachable. `settings.download_root` is
pointed at a per-test tmp dir so ingest's on-disk moves are isolated. The ingest/route tests
monkeypatch `_probe` so they don't shell out to ffprobe; the classifier's OWN parsing (image →
file, cover-art skip, dims, duration) is covered directly by `TestProbeClassifier` below, which
stubs one level down (`run_ffprobe_json`) and exercises the real `_probe`.
"""

from pathlib import Path

import pytest

from app.config import settings
from app.downloads import mediakind, quota, storage
from app.downloads import upload as uploadmod
from app.models import DownloadQuota, MediaAsset


# --- pure: the kind invariant ---------------------------------------------------------------


def test_derive_media_kind():
    assert uploadmod.mediakind.derive("h264", "aac") == mediakind.VIDEO
    assert uploadmod.mediakind.derive("vp9", None) == mediakind.VIDEO
    assert uploadmod.mediakind.derive(None, "aac") == mediakind.AUDIO
    assert uploadmod.mediakind.derive(None, None) == mediakind.FILE


# --- ingest (direct) ------------------------------------------------------------------------


@pytest.fixture
def _root(tmp_path, monkeypatch):
    monkeypatch.setattr(settings, "download_root", str(tmp_path))
    return tmp_path


def _probe(monkeypatch, result):
    monkeypatch.setattr(uploadmod, "_probe", lambda _p: result)


def test_ingest_generic_file_is_download_only(db, make_user, _root, monkeypatch):
    _probe(monkeypatch, (None, None, None, None, None))
    u = make_user()
    src = _root / "src.bin"
    src.write_bytes(b"%PDF-1.4 hello")
    job = uploadmod.ingest_upload(
        db, u.id, filename="My Notes.pdf", src_path=src, size=src.stat().st_size
    )
    asset = db.get(MediaAsset, job.asset_id)
    assert asset.media_kind == mediakind.FILE
    assert asset.status == "ready" and job.status == "done" and job.progress == 100
    assert asset.container == "pdf"
    assert asset.poster_path is None  # a generic file has no player and no poster
    assert asset.title == "My Notes"  # display title from the original name
    assert job.job_kind == "upload" and job.source_kind == "upload"
    # The file was MOVED into the per-user uploads tree, inside the root, and the temp is gone.
    assert not src.exists()
    assert asset.rel_path.startswith(".uploads/")
    assert storage.safe_abs_path(str(_root), asset.rel_path) is not None


def test_ingest_audio_is_media_no_poster(db, make_user, _root, monkeypatch):
    _probe(monkeypatch, (None, "aac", None, None, 200))
    u = make_user()
    src = _root / "song.m4a"
    src.write_bytes(b"fake-audio")
    job = uploadmod.ingest_upload(db, u.id, filename="song.m4a", src_path=src, size=10)
    asset = db.get(MediaAsset, job.asset_id)
    assert asset.media_kind == mediakind.AUDIO
    assert asset.duration_s == 200
    assert asset.poster_path is None  # audio has no frame to poster


def test_ingest_video_records_dims_and_poster(db, make_user, _root, monkeypatch):
    _probe(monkeypatch, ("h264", "aac", 640, 480, 60))
    # Stub the frame-cut so the test doesn't shell out to ffmpeg.
    monkeypatch.setattr(uploadmod.editmod, "ensure_poster", lambda _a, _root: "p/x.jpg")
    u = make_user()
    src = _root / "clip.mp4"
    src.write_bytes(b"fake-video")
    job = uploadmod.ingest_upload(db, u.id, filename="clip.mp4", src_path=src, size=10)
    asset = db.get(MediaAsset, job.asset_id)
    assert asset.media_kind == mediakind.VIDEO
    assert (asset.width, asset.height) == (640, 480)
    assert asset.poster_path == "p/x.jpg"


def test_ingest_filename_is_sanitized_no_traversal(db, make_user, _root, monkeypatch):
    _probe(monkeypatch, (None, None, None, None, None))
    u = make_user()
    src = _root / "src"
    src.write_bytes(b"x")
    job = uploadmod.ingest_upload(db, u.id, filename="../../etc/passwd", src_path=src, size=1)
    asset = db.get(MediaAsset, job.asset_id)
    assert ".." not in asset.rel_path
    assert asset.rel_path.startswith(f".uploads/{u.id}/")
    assert storage.safe_abs_path(str(_root), asset.rel_path) is not None


def test_ingest_rejects_over_footprint_cap_before_writing(db, make_user, _root, monkeypatch):
    _probe(monkeypatch, (None, None, None, None, None))
    u = make_user()
    db.add(
        DownloadQuota(user_id=u.id, max_bytes=1000, max_concurrent=2, max_jobs=50, unlimited=False)
    )
    db.commit()
    src = _root / "big.bin"
    src.write_bytes(b"x" * 10)
    with pytest.raises(quota.QuotaExceeded) as e:
        uploadmod.ingest_upload(db, u.id, filename="big.bin", src_path=src, size=2000)
    assert e.value.reason == "max_bytes"
    assert src.exists()  # rejected before any disk move — nothing landed
    assert db.query(MediaAsset).count() == 0  # and no orphan asset row


# --- route ----------------------------------------------------------------------------------


def test_upload_route_generic_forces_download(as_user, make_user, _root, monkeypatch):
    _probe(monkeypatch, (None, None, None, None, None))
    c = as_user(make_user())
    r = c.post(
        "/api/downloads/upload",
        files={"file": ("report.html", b"<script>alert(1)</script>", "text/html")},
    )
    assert r.status_code == 200, r.text
    job = r.json()
    assert job["asset"]["media_kind"] == "file"
    assert job["can_download"] is True
    # The served file must be a neutral force-download attachment — never inline text/html.
    fr = c.get(f"/api/downloads/{job['id']}/file")
    assert fr.status_code == 200
    assert fr.headers["content-type"].startswith("application/octet-stream")
    assert "attachment" in fr.headers["content-disposition"].lower()


def test_upload_route_media_html_extension_not_served_as_html(
    as_user, make_user, _root, monkeypatch
):
    """The XSS guard: a MEDIA-classified upload (real codecs) named with an executable extension must
    NOT be served with a guessed text/html type. The served Content-Type comes from the verified
    media_kind + a container allow-list ('html' isn't one), so it force-downloads as octet-stream —
    on both the authenticated file endpoint AND the login-free public watch endpoint. Without the fix
    the public inline branch let a media/HTML polyglot named evil.html run as stored XSS."""
    _probe(monkeypatch, ("h264", "aac", 320, 240, 5))  # ffprobe says it's video…
    c = as_user(make_user())
    r = c.post(
        "/api/downloads/upload",
        files={"file": ("evil.html", b"<script>alert(1)</script>", "text/html")},
    )
    assert r.status_code == 200, r.text
    job = r.json()
    assert job["asset"]["media_kind"] == "video" and job["asset"]["container"] == "html"

    # Authenticated endpoint: never text/html.
    fr = c.get(f"/api/downloads/{job['id']}/file")
    assert fr.status_code == 200
    assert fr.headers["content-type"].startswith("application/octet-stream")
    assert "attachment" in fr.headers["content-disposition"].lower()

    # Public login-free endpoint via a STREAM-ONLY link (the inline branch) — still not text/html.
    link = c.post(f"/api/downloads/{job['id']}/links", json={"allow_download": False}).json()
    pr = c.get(f"/api/public/watch/{link['token']}/file")
    assert pr.status_code == 200
    assert not pr.headers["content-type"].startswith("text/html")
    assert pr.headers["content-type"].startswith("application/octet-stream")
    assert "attachment" in pr.headers["content-disposition"].lower()


def test_public_watch_file_forces_download_for_generic_file(as_user, make_user, _root, monkeypatch):
    """The login-free public serve of a generic upload (media_kind='file') must be a neutral
    force-download attachment — the branch that stops uploaded HTML/SVG executing on the public
    origin. This is a SEPARATE code path from the authenticated /file route, so it needs its own
    route-level assertion (previously only the authenticated branch was tested)."""
    _probe(monkeypatch, (None, None, None, None, None))
    c = as_user(make_user())
    job = c.post(
        "/api/downloads/upload",
        files={"file": ("notes.html", b"<script>alert(1)</script>", "text/html")},
    ).json()
    assert job["asset"]["media_kind"] == "file"
    link = c.post(f"/api/downloads/{job['id']}/links", json={"allow_download": True}).json()
    pr = c.get(f"/api/public/watch/{link['token']}/file")
    assert pr.status_code == 200
    assert pr.headers["content-type"].startswith("application/octet-stream")
    assert "attachment" in pr.headers["content-disposition"].lower()


def test_public_watch_file_serves_media_inline_with_real_type(
    as_user, make_user, _root, monkeypatch
):
    """A recognised media container IS still served inline with its real audio/video type (so it
    plays in the watch page) — the fix must not break legitimate playback. A stream-only link →
    inline; the type is video/mp4 (from the container allow-list), never text/* and never a guess."""
    _probe(monkeypatch, ("h264", "aac", 640, 360, 12))
    c = as_user(make_user())
    job = c.post(
        "/api/downloads/upload",
        files={"file": ("clip.mp4", b"fake-mp4-bytes", "video/mp4")},
    ).json()
    link = c.post(f"/api/downloads/{job['id']}/links", json={"allow_download": False}).json()
    pr = c.get(f"/api/public/watch/{link['token']}/file")
    assert pr.status_code == 200
    assert pr.headers["content-type"].startswith("video/mp4")
    assert "inline" in pr.headers["content-disposition"].lower()


def test_upload_route_enforces_per_file_cap(as_user, make_user, db, _root, monkeypatch):
    from app import sysconfig

    sysconfig.set_value(db, "download_upload_max_bytes", 5)
    db.commit()
    c = as_user(make_user())
    r = c.post(
        "/api/downloads/upload",
        files={"file": ("big.bin", b"way over the cap", "application/octet-stream")},
    )
    assert r.status_code == 422
    assert r.json()["detail"]["code"] == "upload_too_large"


def test_upload_cap_cannot_be_disabled_with_zero(db):
    """download_upload_max_bytes floors at 1: a stored 0 would read falsy in the route's
    `if cap and size > cap` guard and SILENTLY disable the per-file cap, so set_value must reject it.
    Pins the min=1 invariant so a revert to min=0 can't pass the gate."""
    from app import sysconfig

    with pytest.raises(ValueError):
        sysconfig.set_value(db, "download_upload_max_bytes", 0)


def test_upload_route_rejects_empty_file(as_user, make_user, _root, monkeypatch):
    c = as_user(make_user())
    r = c.post(
        "/api/downloads/upload", files={"file": ("empty.bin", b"", "application/octet-stream")}
    )
    assert r.status_code == 400
    assert r.json()["detail"]["code"] == "upload_empty"


def test_redownload_rejects_an_upload(as_user, make_user, _root, monkeypatch):
    _probe(monkeypatch, (None, None, None, None, None))
    c = as_user(make_user())
    job = c.post(
        "/api/downloads/upload", files={"file": ("a.bin", b"data", "application/octet-stream")}
    ).json()
    r = c.post(f"/api/downloads/{job['id']}/redownload")
    assert r.status_code == 409


def test_edit_rejects_a_generic_upload_source(as_user, make_user, _root, monkeypatch):
    _probe(monkeypatch, (None, None, None, None, None))
    c = as_user(make_user())
    job = c.post(
        "/api/downloads/upload", files={"file": ("a.bin", b"data", "application/octet-stream")}
    ).json()
    r = c.post(
        "/api/downloads/edit",
        json={"source_job_id": job["id"], "edit_spec": {"trim": {"start_s": 0, "end_s": 1}}},
    )
    assert r.status_code == 400
    assert r.json()["detail"]["reason"] == "not_media"


def test_ingest_rejects_over_max_jobs(db, make_user, _root, monkeypatch):
    """The other check_upload rejection branch (holdings count, not bytes): once the user is at
    max_jobs, a further upload is refused before any disk write."""
    _probe(monkeypatch, (None, None, None, None, None))
    u = make_user()
    db.add(
        DownloadQuota(user_id=u.id, max_bytes=10**12, max_concurrent=2, max_jobs=1, unlimited=False)
    )
    db.commit()
    src1 = _root / "a.bin"
    src1.write_bytes(b"a")
    uploadmod.ingest_upload(db, u.id, filename="a.bin", src_path=src1, size=1)  # now 1 holding job
    src2 = _root / "b.bin"
    src2.write_bytes(b"b")
    with pytest.raises(quota.QuotaExceeded) as e:
        uploadmod.ingest_upload(db, u.id, filename="b.bin", src_path=src2, size=1)
    assert e.value.reason == "max_jobs"
    assert src2.exists()  # rejected before the move — nothing landed


def test_ingest_cleans_up_placed_files_on_commit_failure(db, make_user, _root, monkeypatch):
    """A failure AFTER storage.place_file has moved the file onto disk must roll back AND unlink the
    placed file + any (even partial) poster — otherwise the orphan is permanent (the per-user delete
    keys off the rolled-back asset.rel_path, and the GC orphan sweep skips the dot-prefixed .uploads/
    tree). Exercises the rollback/cleanup branch that IS the leak remediation, incl. the poster case
    where ensure_poster left a partial file but returned None."""
    _probe(monkeypatch, ("h264", "aac", 320, 240, 5))  # video → the poster path runs
    u = make_user()

    # Simulate ffmpeg leaving a partial poster on disk, then failing (ensure_poster returns None).
    def _partial_poster(asset, root):
        storage.abs_path(root, str(Path(asset.rel_path).with_suffix(".jpg"))).write_bytes(
            b"partial"
        )
        return None

    monkeypatch.setattr(uploadmod.editmod, "ensure_poster", _partial_poster)

    # Force the commit to fail after the file (and partial poster) are on disk.
    def _boom():
        raise RuntimeError("db down")

    monkeypatch.setattr(db, "commit", _boom)

    src = _root / "clip.mp4"
    src.write_bytes(b"fake-video-bytes")
    with pytest.raises(RuntimeError):
        uploadmod.ingest_upload(db, u.id, filename="clip.mp4", src_path=src, size=16)

    db.rollback()  # drop the monkeypatched-commit session state before querying
    uploads = _root / ".uploads"
    leftover = [p for p in uploads.rglob("*") if p.is_file()] if uploads.exists() else []
    assert leftover == [], f"orphaned files left on disk: {leftover}"
    assert db.query(MediaAsset).count() == 0  # the flushed asset rolled back — no dangling row


# --- classifier: the REAL _probe parsing (stubs ffprobe one level down) ----------------------
# The ingest/route tests above stub _probe wholesale; these exercise _probe's OWN logic — image
# detection, the cover-art skip, dimension/duration parsing — by faking only run_ffprobe_json, so a
# regression in the classifier the force-download guarantee leans on is actually caught.


class TestProbeClassifier:
    def _probe_with(self, monkeypatch, data):
        monkeypatch.setattr(uploadmod.editmod, "run_ffprobe_json", lambda *a, **k: data)
        return uploadmod._probe(Path("x"))

    def test_probe_failure_is_a_generic_file(self, monkeypatch):
        # run_ffprobe_json returns None (ffprobe failed / not media) → all None → derive() == FILE.
        v, a, w, h, d = self._probe_with(monkeypatch, None)
        assert (v, a, w, h, d) == (None, None, None, None, None)
        assert mediakind.derive(v, a) == mediakind.FILE

    @pytest.mark.parametrize("fmt", ["image2", "png_pipe", "jpeg_pipe", "webp_pipe", "gif"])
    def test_still_image_classifies_as_file_not_video(self, monkeypatch, fmt):
        # A standalone image exposes a real video stream, but its container format is an image demuxer
        # — it must NOT become media_kind='video' (broken player + inline serving). Stays a file.
        data = {
            "format": {"format_name": fmt},
            "streams": [{"codec_type": "video", "codec_name": "png", "width": 8, "height": 8}],
        }
        v, a, w, h, d = self._probe_with(monkeypatch, data)
        assert v is None and mediakind.derive(v, a) == mediakind.FILE

    def test_cover_art_video_stream_is_skipped_audio_stays_audio(self, monkeypatch):
        # Album art embedded in an audio file is a video stream flagged attached_pic — skip it so the
        # asset classifies as audio, not video.
        data = {
            "format": {"format_name": "mp3", "duration": "212.5"},
            "streams": [
                {
                    "codec_type": "video",
                    "codec_name": "mjpeg",
                    "disposition": {"attached_pic": 1},
                    "width": 600,
                    "height": 600,
                },
                {"codec_type": "audio", "codec_name": "mp3"},
            ],
        }
        v, a, w, h, d = self._probe_with(monkeypatch, data)
        assert (v, a, d) == (None, "mp3", 212)
        assert mediakind.derive(v, a) == mediakind.AUDIO

    def test_real_video_records_codecs_dims_duration(self, monkeypatch):
        data = {
            "format": {"format_name": "mov,mp4,m4a", "duration": "60.0"},
            "streams": [
                {"codec_type": "video", "codec_name": "h264", "width": 1920, "height": 1080},
                {"codec_type": "audio", "codec_name": "aac"},
            ],
        }
        v, a, w, h, d = self._probe_with(monkeypatch, data)
        assert (v, a, w, h, d) == ("h264", "aac", 1920, 1080, 60)
        assert mediakind.derive(v, a) == mediakind.VIDEO

    def test_unparseable_duration_is_none(self, monkeypatch):
        data = {
            "format": {"format_name": "wav", "duration": "N/A"},
            "streams": [{"codec_type": "audio", "codec_name": "pcm_s16le"}],
        }
        v, a, w, h, d = self._probe_with(monkeypatch, data)
        assert d is None and mediakind.derive(v, a) == mediakind.AUDIO