122 lines
4.1 KiB
Python
122 lines
4.1 KiB
Python
import tempfile
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from app.services.downloader import (
|
|
HttpxDownloader,
|
|
UnsupportedURLError,
|
|
YtDlpDownloader,
|
|
_create_unique,
|
|
_unique_path,
|
|
_ytdlp_cookie_opts,
|
|
list_strategies,
|
|
pick_downloader,
|
|
)
|
|
from app.services.extractors import load_extractors
|
|
from app.services.extractors.google_drive import GoogleDriveExtractor
|
|
|
|
|
|
def test_unique_path_appends_counter():
|
|
d = Path(tempfile.mkdtemp())
|
|
assert _unique_path(d, "a.zip") == d / "a.zip"
|
|
(d / "a.zip").write_bytes(b"")
|
|
assert _unique_path(d, "a.zip") == d / "a (1).zip"
|
|
(d / "a (1).zip").write_bytes(b"")
|
|
assert _unique_path(d, "a.zip") == d / "a (2).zip"
|
|
|
|
|
|
def test_create_unique_reserves_atomically():
|
|
d = Path(tempfile.mkdtemp())
|
|
p1, f1 = _create_unique(d, "file.bin")
|
|
p2, f2 = _create_unique(d, "file.bin")
|
|
try:
|
|
assert p1 == d / "file.bin"
|
|
assert p2 == d / "file (1).bin" # имя занято — следующий уходит в (1)
|
|
assert p1.exists() and p2.exists() # оба зарезервированы сразу, не на старте записи
|
|
finally:
|
|
f1.close()
|
|
f2.close()
|
|
|
|
|
|
def test_create_unique_without_extension():
|
|
d = Path(tempfile.mkdtemp())
|
|
p1, f1 = _create_unique(d, "noext")
|
|
p2, f2 = _create_unique(d, "noext")
|
|
try:
|
|
assert p1 == d / "noext"
|
|
assert p2 == d / "noext (1)"
|
|
finally:
|
|
f1.close()
|
|
f2.close()
|
|
|
|
|
|
# --- Реестр стратегий и расширения ---
|
|
|
|
|
|
def test_pick_downloader_chain():
|
|
load_extractors() # подгружаем экстракторы (как делает lifespan)
|
|
|
|
gd = pick_downloader("https://drive.google.com/file/d/ABC123_xyz/view?usp=sharing")
|
|
assert isinstance(gd, GoogleDriveExtractor) # расширение перехватывает раньше всех
|
|
|
|
assert isinstance(pick_downloader("https://e.com/archive.zip"), HttpxDownloader)
|
|
assert isinstance(pick_downloader("https://youtube.com/watch?v=x"), YtDlpDownloader)
|
|
|
|
|
|
def test_pick_downloader_raises_for_unsupported():
|
|
# Не прямой файл и не из загрузчиков (example.com — резервный домен, yt-dlp
|
|
# его не знает) -> явная ошибка, а не молчаливый фолбэк.
|
|
with pytest.raises(UnsupportedURLError):
|
|
pick_downloader("https://example.com/just-a-page")
|
|
|
|
|
|
def test_ytdlp_cookie_opts():
|
|
assert _ytdlp_cookie_opts(None, None) == {}
|
|
assert _ytdlp_cookie_opts("chrome", None) == {
|
|
"cookiesfrombrowser": ("chrome", None, None, None)
|
|
}
|
|
assert _ytdlp_cookie_opts("chrome:Work", None)["cookiesfrombrowser"] == (
|
|
"chrome",
|
|
"Work",
|
|
None,
|
|
None,
|
|
)
|
|
assert _ytdlp_cookie_opts(None, "/tmp/c.txt") == {"cookiefile": "/tmp/c.txt"}
|
|
both = _ytdlp_cookie_opts("firefox", "/tmp/c.txt")
|
|
assert both["cookiefile"] == "/tmp/c.txt"
|
|
assert both["cookiesfrombrowser"][0] == "firefox"
|
|
|
|
|
|
def test_list_strategies_sorted_with_builtins():
|
|
strategies = list_strategies()
|
|
kinds = {s["kind"] for s in strategies}
|
|
assert {"direct", "fallback"} <= kinds
|
|
prios = [s["priority"] for s in strategies]
|
|
assert prios == sorted(prios, reverse=True) # по убыванию приоритета
|
|
|
|
|
|
def test_load_extractors_idempotent():
|
|
from app.services.downloader import _REGISTRY
|
|
|
|
load_extractors()
|
|
before = len(_REGISTRY)
|
|
load_extractors() # повторный вызов не должен задваивать реестр
|
|
assert len(_REGISTRY) == before
|
|
|
|
|
|
def test_gdrive_matches():
|
|
assert GoogleDriveExtractor.matches("https://drive.google.com/file/d/XyZ_1/view")
|
|
assert not GoogleDriveExtractor.matches("https://example.com/file.zip")
|
|
|
|
|
|
async def test_gdrive_resolves_to_direct_url():
|
|
extractor = GoogleDriveExtractor()
|
|
resolved = await extractor.resolve(
|
|
"https://drive.google.com/file/d/ABC123_xyz/view?usp=sharing"
|
|
)
|
|
assert (
|
|
resolved.download_url
|
|
== "https://drive.google.com/uc?export=download&id=ABC123_xyz"
|
|
)
|