peter
1102ad6a28
Deep-review round 1 on bug/download-mediakind-probe found a real HIGH the prior review missed: - resolve_media_facts used a per-field `probe or info` fallback, so a probe that RAN and deliberately found no video track (cover-art-only audio: attached_pic skipped; a still image: format-guarded) had its None video overridden by yt-dlp's stale vcodec. An audio-mode download from a progressive source (info carries the source's video codec, embed-thumbnail on) was then classified VIDEO -> a Play button + a broken in-app <video> with no video track. Only the download path regressed (upload calls probe_media directly). Fix: gate precedence on whether ffprobe RAN, not per-field. If the probe read the file, its codec verdict wins outright (a None means 'no such track'); only a full ffprobe failure falls back to info. Dimensions still gap-fill from info. Refactored the parsing into _facts_from_probe so resolve_media_facts can see the raw probe-ran signal; probe_media keeps its 5-tuple (upload + TestProbeClassifier unchanged). - tests: the cover-art-audio and still-image resurrection cases, a differing populated-info precedence case (the prior-flagged gap), and the ffprobe-failed info fallback — now stubbing run_ffprobe_json so the real parsing + precedence run together. Backend gate green (ruff + full pytest).
504 lines
23 KiB
Python
504 lines
23 KiB
Python
"""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 the shared `probe.probe_media` 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_media`.
|
|
"""
|
|
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from app.config import settings
|
|
from app.downloads import mediakind, quota, storage
|
|
from app.downloads import probe as probemod
|
|
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):
|
|
# upload ingest now calls the shared `probe.probe_media` (same ground-truth ffprobe the download
|
|
# worker uses); patch it at that shared home.
|
|
monkeypatch.setattr(probemod, "probe_media", 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_media parsing (stubs ffprobe one level down) ------------------
|
|
# The ingest/route tests above stub probe_media wholesale; these exercise its 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(probemod.editmod, "run_ffprobe_json", lambda *a, **k: data)
|
|
return probemod.probe_media(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
|
|
|
|
|
|
# --- resolve_media_facts: probe-of-the-file wins over yt-dlp's info (the FB regression fix) ---------
|
|
# The download worker's finalize must classify from the ACTUAL file, not the extractor's `info` dict.
|
|
# yt-dlp omits vcodec/acodec/dims for some sources (Facebook), and deriving media_kind from those
|
|
# nulls turned a real H.264 video into a download-only 'file' (no in-app play). These pin that the
|
|
# probe is authoritative and `info` is only a gap-filler.
|
|
|
|
|
|
class TestResolveMediaFacts:
|
|
# These stub run_ffprobe_json (one level below probe_media) so the REAL probe parsing + the
|
|
# probe-vs-info precedence are exercised together — the precedence keys off whether ffprobe RAN.
|
|
@staticmethod
|
|
def _ffprobe(monkeypatch, data):
|
|
monkeypatch.setattr(probemod.editmod, "run_ffprobe_json", lambda *a, **k: data)
|
|
|
|
def test_probe_wins_when_info_lacks_codecs(self, monkeypatch):
|
|
# THE original regression: info has no vcodec/acodec/dims (Facebook), but the file probes as
|
|
# h264/aac.
|
|
self._ffprobe(
|
|
monkeypatch,
|
|
{
|
|
"format": {"format_name": "mov,mp4,m4a"},
|
|
"streams": [
|
|
{"codec_type": "video", "codec_name": "h264", "width": 720, "height": 1280},
|
|
{"codec_type": "audio", "codec_name": "aac"},
|
|
],
|
|
},
|
|
)
|
|
v, a, w, h = probemod.resolve_media_facts(Path("x"), {"width": None, "height": None})
|
|
assert (v, a, w, h) == ("h264", "aac", 720, 1280)
|
|
assert mediakind.derive(v, a) == mediakind.VIDEO # NOT 'file'
|
|
|
|
def test_probe_wins_over_a_DIFFERING_populated_info(self, monkeypatch):
|
|
# Both the probe AND info are populated but DISAGREE (info carries a pre-remux/pre-merge codec).
|
|
# The finished file on disk is ground truth, so the probe wins outright — pins the precedence a
|
|
# `info or probe` inversion would silently break.
|
|
self._ffprobe(
|
|
monkeypatch,
|
|
{
|
|
"format": {"format_name": "mov,mp4,m4a"},
|
|
"streams": [
|
|
{"codec_type": "video", "codec_name": "h264", "width": 1920, "height": 1080},
|
|
{"codec_type": "audio", "codec_name": "aac"},
|
|
],
|
|
},
|
|
)
|
|
v, a, w, h = probemod.resolve_media_facts(
|
|
Path("x"), {"vcodec": "vp9", "acodec": "opus", "width": 640, "height": 480}
|
|
)
|
|
assert (v, a, w, h) == ("h264", "aac", 1920, 1080) # probe, not info
|
|
|
|
def test_cover_art_audio_download_is_not_resurrected_to_video_by_info(self, monkeypatch):
|
|
# THE bug this fix targets: an audio-mode download from a progressive source embeds cover art,
|
|
# and yt-dlp's info still carries the source's video codec. The probe correctly finds only audio
|
|
# (the attached_pic is skipped); info's vcodec must NOT resurrect a video track → media_kind
|
|
# stays audio, not a broken in-app 'video'.
|
|
self._ffprobe(
|
|
monkeypatch,
|
|
{
|
|
"format": {"format_name": "mp3", "duration": "180.0"},
|
|
"streams": [
|
|
{
|
|
"codec_type": "video",
|
|
"codec_name": "mjpeg",
|
|
"disposition": {"attached_pic": 1},
|
|
},
|
|
{"codec_type": "audio", "codec_name": "mp3"},
|
|
],
|
|
},
|
|
)
|
|
v, a, w, h = probemod.resolve_media_facts(
|
|
Path("x"), {"vcodec": "avc1", "acodec": "mp4a", "width": 1280, "height": 720}
|
|
)
|
|
assert v is None and a == "mp3" # info's avc1 did NOT win
|
|
assert mediakind.derive(v, a) == mediakind.AUDIO # NOT VIDEO
|
|
|
|
def test_still_image_download_is_not_resurrected_to_video_by_info(self, monkeypatch):
|
|
# A downloaded still image: the probe returns no vcodec (image-format guard). info's image
|
|
# "vcodec" must not turn it into a broken 'video' → stays a generic file.
|
|
self._ffprobe(
|
|
monkeypatch,
|
|
{
|
|
"format": {"format_name": "png_pipe"},
|
|
"streams": [{"codec_type": "video", "codec_name": "png", "width": 8, "height": 8}],
|
|
},
|
|
)
|
|
v, a, w, h = probemod.resolve_media_facts(
|
|
Path("x"), {"vcodec": "png", "width": 8, "height": 8}
|
|
)
|
|
assert v is None and mediakind.derive(v, a) == mediakind.FILE
|
|
|
|
def test_info_fills_in_only_when_ffprobe_could_not_read_the_file(self, monkeypatch):
|
|
# ffprobe failed entirely (None) → fall back to info so we never regress the working path.
|
|
self._ffprobe(monkeypatch, None)
|
|
v, a, w, h = probemod.resolve_media_facts(
|
|
Path("x"), {"vcodec": "vp9", "acodec": "opus", "width": 1920, "height": 1080}
|
|
)
|
|
assert (v, a, w, h) == ("vp9", "opus", 1920, 1080)
|
|
|
|
def test_info_none_codec_string_is_treated_as_absent(self, monkeypatch):
|
|
# yt-dlp writes the literal "none" for a missing stream — it must not become a codec name.
|
|
self._ffprobe(monkeypatch, None) # probe failed → info path, where the "none" scrub applies
|
|
v, a, w, h = probemod.resolve_media_facts(Path("x"), {"vcodec": "none", "acodec": "aac"})
|
|
assert (v, a) == (None, "aac")
|
|
assert mediakind.derive(v, a) == mediakind.AUDIO
|