152 lines
4.8 KiB
Python
152 lines
4.8 KiB
Python
"""GalleryDLService tests — subprocess is mocked so no real gallery-dl runs."""
|
|
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
|
|
from backend.app.services.gallery_dl import (
|
|
ErrorType,
|
|
GalleryDLService,
|
|
SourceConfig,
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def gdl(tmp_path):
|
|
return GalleryDLService(
|
|
images_root=tmp_path / "images",
|
|
rate_limit=3.0,
|
|
validate_files=False,
|
|
)
|
|
|
|
|
|
def _proc(stdout="", stderr="", returncode=0):
|
|
return SimpleNamespace(stdout=stdout, stderr=stderr, returncode=returncode)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_download_happy_path(gdl, monkeypatch):
|
|
out_path = gdl.images_root / "alice" / "patreon" / "post1" / "img.jpg"
|
|
out_path.parent.mkdir(parents=True, exist_ok=True)
|
|
out_path.write_bytes(b"\xff\xd8\xff\xe0" + b"\x00" * 32 + b"\xff\xd9")
|
|
|
|
monkeypatch.setattr(
|
|
"backend.app.services.gallery_dl.subprocess.run",
|
|
lambda *a, **k: _proc(stdout=str(out_path) + "\n", returncode=0),
|
|
)
|
|
result = await gdl.download(
|
|
url="https://patreon.com/alice",
|
|
artist_slug="alice",
|
|
platform="patreon",
|
|
source_config=SourceConfig(),
|
|
)
|
|
assert result.success is True
|
|
assert result.files_downloaded == 1
|
|
assert result.error_type is None
|
|
assert str(out_path) in result.written_paths
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_download_no_new_content(gdl, monkeypatch):
|
|
monkeypatch.setattr(
|
|
"backend.app.services.gallery_dl.subprocess.run",
|
|
lambda *a, **k: _proc(stdout="# /images/alice/patreon/old.jpg\n", returncode=0),
|
|
)
|
|
result = await gdl.download(
|
|
url="https://patreon.com/alice", artist_slug="alice", platform="patreon",
|
|
source_config=SourceConfig(),
|
|
)
|
|
assert result.success is True
|
|
assert result.files_downloaded == 0
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_download_auth_error_categorization(gdl, monkeypatch):
|
|
monkeypatch.setattr(
|
|
"backend.app.services.gallery_dl.subprocess.run",
|
|
lambda *a, **k: _proc(
|
|
stdout="", stderr='[patreon][error] "GET /api/foo HTTP/1.1" 401 None\n',
|
|
returncode=1,
|
|
),
|
|
)
|
|
result = await gdl.download(
|
|
url="https://patreon.com/alice", artist_slug="alice", platform="patreon",
|
|
source_config=SourceConfig(),
|
|
)
|
|
assert result.success is False
|
|
assert result.error_type == ErrorType.AUTH_ERROR
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_download_timeout(gdl, monkeypatch):
|
|
import subprocess as sp
|
|
|
|
def _raise(*a, **k):
|
|
raise sp.TimeoutExpired(cmd="gallery-dl", timeout=10)
|
|
|
|
monkeypatch.setattr("backend.app.services.gallery_dl.subprocess.run", _raise)
|
|
result = await gdl.download(
|
|
url="https://patreon.com/alice", artist_slug="alice", platform="patreon",
|
|
source_config=SourceConfig(timeout=10),
|
|
)
|
|
assert result.success is False
|
|
assert result.error_type == ErrorType.TIMEOUT
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_validation_quarantines_truncated_files(tmp_path, monkeypatch):
|
|
gdl = GalleryDLService(
|
|
images_root=tmp_path / "images", rate_limit=3.0, validate_files=True,
|
|
)
|
|
bad = gdl.images_root / "alice" / "patreon" / "post" / "bad.jpg"
|
|
bad.parent.mkdir(parents=True, exist_ok=True)
|
|
bad.write_bytes(b"\xff\xd8\xff" + b"\x00" * 100) # SOI but no EOI
|
|
|
|
monkeypatch.setattr(
|
|
"backend.app.services.gallery_dl.subprocess.run",
|
|
lambda *a, **k: _proc(stdout=str(bad) + "\n", returncode=0),
|
|
)
|
|
result = await gdl.download(
|
|
url="https://patreon.com/alice", artist_slug="alice", platform="patreon",
|
|
source_config=SourceConfig(),
|
|
)
|
|
assert result.files_quarantined == 1
|
|
assert len(result.quarantined_paths) == 1
|
|
assert not bad.exists()
|
|
assert result.error_type == ErrorType.VALIDATION_FAILED
|
|
|
|
|
|
def test_compute_run_stats(gdl):
|
|
stats = gdl._compute_run_stats(
|
|
return_code=0,
|
|
stdout="/path/a.jpg\n/path/b.jpg\n# /path/c.jpg\n",
|
|
stderr="[patreon][warning] foo\n[patreon][download][error] failed to download /x\n",
|
|
)
|
|
assert stats["downloaded_count"] == 2
|
|
assert stats["skipped_count"] == 1
|
|
assert stats["warning_count"] >= 1
|
|
assert stats["per_item_failures"] == 1
|
|
|
|
|
|
def test_extract_errors_warnings(gdl):
|
|
stderr = (
|
|
"[patreon][debug] foo\n[patreon][error] bar\n"
|
|
"[urllib3][debug] baz\n[patreon][warning] qux\n"
|
|
)
|
|
result = gdl._extract_errors_warnings(stderr)
|
|
assert "[error]" in result
|
|
assert "[warning]" in result
|
|
assert "[debug]" not in result
|
|
|
|
|
|
def test_truncate_log_caps_large_text(gdl):
|
|
huge = "line\n" * 200_000
|
|
result = gdl._truncate_log(huge, max_bytes=10_000)
|
|
assert len(result.encode()) < 11_000
|
|
assert "lines elided" in result
|
|
|
|
|
|
def test_truncate_log_passes_short_text(gdl):
|
|
text = "small\n"
|
|
assert gdl._truncate_log(text) == text
|