"""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; `_probe` is monkeypatched so the tests don't depend on ffprobe (its real behaviour is exercised by the download/edit suites). """ 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"", "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_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_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"