Roman
4e1faf4cd9
- /api/v2/* : login SID (mot de passe = clé Torznab), app/version, torrents/info (progression temps réel), properties (content_path), add (rejoue le grab encodé dans le .torrent de service, dédupliqué par infohash SHA-1), delete (± fichiers), pause/resume - Les grabs Sonarr sont marqués « sonarr:<hash>| » dans source_key → suivis de bout en bout : Sonarr importe, renomme et range les épisodes dans sa bibliothèque, puis retire le torrent de la file Ohm - L'indexeur Torznab embarque les paramètres du grab dans l'announce - README : nouveau mode « client de téléchargement » recommandé (Remote Path Mapping documenté), blackhole en variante minimale - 3 nouveaux tests (flux complet add → suivi → import → delete)
240 lines
8.0 KiB
Python
240 lines
8.0 KiB
Python
"""Tests de l'indexeur Torznab (compatibilité Sonarr / Prowlarr)."""
|
|
|
|
import xml.etree.ElementTree as ET
|
|
|
|
import pytest
|
|
from httpx import ASGITransport, AsyncClient
|
|
|
|
from app.db import db
|
|
from app.main import app
|
|
from app.scrapers.base import Episode, SearchResult, VideoLink
|
|
from app.services.settings import get_torznab_apikey
|
|
from app.services.torznab import torznab as service
|
|
|
|
|
|
@pytest.fixture
|
|
async def client() -> AsyncClient:
|
|
transport = ASGITransport(app=app)
|
|
async with AsyncClient(transport=transport, base_url="http://test") as c:
|
|
yield c
|
|
|
|
|
|
@pytest.fixture
|
|
async def apikey() -> str:
|
|
return await get_torznab_apikey()
|
|
|
|
|
|
class FakeSource:
|
|
name = "fake"
|
|
label = "Fake"
|
|
base_url = "https://fake.example"
|
|
|
|
async def search(self, query: str) -> list[SearchResult]:
|
|
return [
|
|
SearchResult(
|
|
source="fake",
|
|
source_id="frieren-1",
|
|
title="Frieren",
|
|
url="https://fake.example/frieren",
|
|
)
|
|
]
|
|
|
|
async def list_episodes(self, source_id: str) -> list[Episode]:
|
|
return [
|
|
Episode(number=1, title="Épisode 1", url="https://fake.example/ep1", season=1),
|
|
Episode(number=2, title="Épisode 2", url="https://fake.example/ep2", season=1),
|
|
Episode(number=2.5, title="OAV", url="https://fake.example/oav", season=1),
|
|
Episode(number=1, title="Épisode 1", url="https://fake.example/s2ep1", season=2),
|
|
]
|
|
|
|
async def extract_embed_links(self, episode_url: str) -> list[str]:
|
|
return ["https://embed.example/player"]
|
|
|
|
async def get_details(self, source_id): ...
|
|
async def latest(self): return []
|
|
|
|
|
|
@pytest.fixture
|
|
def fake_source(monkeypatch):
|
|
source = FakeSource()
|
|
|
|
async def _sources():
|
|
return [source]
|
|
|
|
monkeypatch.setattr(service, "_enabled_sources", _sources)
|
|
return source
|
|
|
|
|
|
# ---------------------------------------------------------------- auth + caps
|
|
|
|
|
|
async def test_torznab_rejects_missing_apikey(client):
|
|
r = await client.get("/torznab/api", params={"t": "caps"})
|
|
assert r.status_code == 401
|
|
assert 'code="100"' in r.text
|
|
|
|
|
|
async def test_torznab_rejects_wrong_apikey(client):
|
|
r = await client.get("/torznab/api", params={"t": "caps", "apikey": "mauvaise"})
|
|
assert r.status_code == 401
|
|
|
|
|
|
async def test_torznab_caps(client, apikey):
|
|
r = await client.get("/torznab/api", params={"t": "caps", "apikey": apikey})
|
|
assert r.status_code == 200
|
|
root = ET.fromstring(r.text)
|
|
assert root.tag == "caps"
|
|
tv = root.find(".//tv-search")
|
|
assert tv is not None and tv.get("available") == "yes"
|
|
assert "q,season,ep" in (tv.get("supportedParams") or "")
|
|
assert any(c.get("id") == "5070" for c in root.findall(".//subcat"))
|
|
|
|
|
|
async def test_torznab_apikey_via_header(client, apikey):
|
|
r = await client.get("/torznab/api", params={"t": "caps"}, headers={"X-Api-Key": apikey})
|
|
assert r.status_code == 200
|
|
|
|
|
|
async def test_torznab_unknown_function(client, apikey):
|
|
r = await client.get("/torznab/api", params={"t": "music", "apikey": apikey})
|
|
assert r.status_code == 400
|
|
assert 'code="203"' in r.text
|
|
|
|
|
|
# ---------------------------------------------------------------- recherche
|
|
|
|
|
|
async def test_torznab_tvsearch_filters_season_ep(client, apikey, fake_source):
|
|
r = await client.get(
|
|
"/torznab/api",
|
|
params={"t": "tvsearch", "q": "Frieren", "season": 1, "ep": 2, "apikey": apikey},
|
|
)
|
|
assert r.status_code == 200
|
|
root = ET.fromstring(r.text)
|
|
items = root.findall(".//item")
|
|
assert len(items) == 1
|
|
assert items[0].find("title").text == "Frieren S01E02 VOSTFR WEB-DL"
|
|
link = items[0].find("link").text
|
|
assert "/torznab/download" in link and "season=1" in link and "ep=2" in link
|
|
|
|
|
|
async def test_torznab_tvsearch_all_episodes(client, apikey, fake_source):
|
|
r = await client.get(
|
|
"/torznab/api", params={"t": "tvsearch", "q": "Frieren", "apikey": apikey}
|
|
)
|
|
root = ET.fromstring(r.text)
|
|
titles = [i.find("title").text for i in root.findall(".//item")]
|
|
assert len(titles) == 3 # OAV 2.5 exclue : S01E01, S01E02, S02E01
|
|
assert "Frieren S02E01 VOSTFR WEB-DL" in titles
|
|
|
|
|
|
async def test_torznab_search_requires_query(client, apikey):
|
|
r = await client.get("/torznab/api", params={"t": "search", "apikey": apikey})
|
|
assert r.status_code == 400
|
|
|
|
|
|
# ---------------------------------------------------------------- grab
|
|
|
|
|
|
class _FakeManager:
|
|
def __init__(self) -> None:
|
|
self.calls: list[tuple[str, str | None]] = []
|
|
|
|
async def enqueue(
|
|
self, video_url: str, page_url: str, title: str, source_key: str | None = None
|
|
) -> dict:
|
|
self.calls.append((video_url, source_key))
|
|
return {"id": 7, "title": title, "status": "pending"}
|
|
|
|
|
|
class _FakeExtractor:
|
|
name = "fakehost"
|
|
|
|
async def extract(self, embed_url: str) -> VideoLink:
|
|
return VideoLink(url="https://cdn.example/video.mp4", hoster="fakehost")
|
|
|
|
|
|
async def test_torznab_download_enqueues_and_returns_torrent(client, apikey, fake_source, monkeypatch):
|
|
manager = _FakeManager()
|
|
monkeypatch.setattr("app.scrapers.base.get_source", lambda name: fake_source)
|
|
monkeypatch.setattr("app.services.torznab.resolve_hoster", lambda url: _FakeExtractor())
|
|
monkeypatch.setattr("app.services.torznab.download_manager", manager)
|
|
|
|
r = await client.get(
|
|
"/torznab/download",
|
|
params={
|
|
"apikey": apikey,
|
|
"source": "fake",
|
|
"sid": "frieren-1",
|
|
"season": 1,
|
|
"ep": 2,
|
|
"series": "Frieren",
|
|
},
|
|
)
|
|
assert r.status_code == 200
|
|
assert r.headers["content-type"].startswith("application/x-bittorrent")
|
|
assert r.content.startswith(b"d") and b"OhmStreaming" in r.content # bencode valide
|
|
video_url, source_key = manager.calls[0]
|
|
assert video_url == "https://cdn.example/video.mp4"
|
|
assert source_key and source_key.startswith("sonarr:") and video_url in source_key
|
|
|
|
|
|
async def test_torznab_download_rejects_bad_key(client):
|
|
r = await client.get("/torznab/download", params={"apikey": "mauvaise"})
|
|
assert r.status_code == 401
|
|
|
|
|
|
async def test_torznab_download_unknown_episode(client, apikey, fake_source, monkeypatch):
|
|
monkeypatch.setattr("app.scrapers.base.get_source", lambda name: fake_source)
|
|
r = await client.get(
|
|
"/torznab/download",
|
|
params={
|
|
"apikey": apikey,
|
|
"source": "fake",
|
|
"sid": "frieren-1",
|
|
"season": 9,
|
|
"ep": 9,
|
|
"series": "Frieren",
|
|
},
|
|
)
|
|
assert r.status_code == 502
|
|
assert 'code="300"' in r.text
|
|
|
|
|
|
# ---------------------------------------------------------------- admin
|
|
|
|
|
|
async def _admin(client: AsyncClient):
|
|
r = await client.post("/auth/register", data={"username": "admin", "password": "secret123"})
|
|
assert r.status_code == 303
|
|
return r.cookies
|
|
|
|
|
|
async def test_admin_integrations(client, apikey):
|
|
cookies = await _admin(client)
|
|
r = await client.get("/api/admin/integrations", cookies=cookies)
|
|
assert r.status_code == 200
|
|
data = r.json()
|
|
assert data["torznab"]["apikey"] == apikey
|
|
assert data["torznab"]["endpoint"].endswith("/torznab/api")
|
|
assert data["sonarr"] == {"url": "", "apikey": ""}
|
|
|
|
r = await client.put(
|
|
"/api/admin/integrations/sonarr",
|
|
json={"url": "http://sonarr:8989/", "apikey": "abc"},
|
|
cookies=cookies,
|
|
)
|
|
assert r.status_code == 200
|
|
row = await db.fetchone("SELECT value FROM settings WHERE key = 'sonarr:url'")
|
|
assert row["value"] == '"http://sonarr:8989"' # slash final retiré
|
|
|
|
|
|
async def test_admin_regenerate_torznab_key(client, apikey):
|
|
cookies = await _admin(client)
|
|
r = await client.post("/api/admin/integrations/torznab/regenerate", cookies=cookies)
|
|
assert r.status_code == 200
|
|
new_key = r.json()["apikey"]
|
|
assert new_key != apikey and len(new_key) == 32
|
|
r = await client.get("/torznab/api", params={"t": "caps", "apikey": apikey})
|
|
assert r.status_code == 401 # l'ancienne clé est révoquée
|