starred/shelfmark
1
0
Fork 0
mirror of https://github.com/calibrain/shelfmark.git synced 2026-10-05 13:11:13 +01:00
Code Issues Packages Projects Releases 10 Wiki Activity

feat(clients): add Debrid-Link as a torrent client (#1380)

Browse Source 
calibrain said in #1315 that he'd review and merge a Debrid-Link client
if someone took a stab at it, so here it is. It's the fourth debrid
service alongside AllDebrid, Real-Debrid and TorBox and it uses the same
DownloadClient interface as the other three. No frontend changes, the
settings UI picks it up from the backend field definitions.

**I can't test this against the live API.** I don't use debrid and I
don't have a Debrid-Link key. Everything here is built from their v2 API
docs and cross-checked against two independent reference
implementations, so the request shapes should be right, but nobody has
run a real magnet through it yet.

@Tatsu941 you opened the issue and mentioned they give developers free
premium. Would you be able to try it? The things most worth confirming
are the field names on a seedbox torrent, that `premiumLeft` is the
right premium check, and that an API error comes back the way the client
expects. Happy to fix whatever turns up.

Two things about their API shaped how this is written.

Every v2 endpoint wraps its payload in a `{success, value}` envelope and
a failure can arrive with an HTTP 200, so `_request_value` decides the
outcome from the body instead of the status code.

A completed seedbox torrent already carries a direct `downloadUrl` on
every file. The other three clients each need a separate unrestrict or
link-request call per file and this one doesn't, which is why it came
out shorter than them despite doing the same job.

Endpoints used:

| Endpoint | Used for |
|---|---|
| `GET account/infos` | test the key, confirm premium time remains |
| `POST seedbox/add` | magnet as JSON, .torrent as multipart |
| `GET seedbox/list` | poll progress |
| `DELETE seedbox/{id}/remove` | clean up |

Ruff and basedpyright are clean and the 36 new tests pass. The two
failures in `tests/e2e/test_proxy_auth_middleware.py` are already on
main and fail the same way on a pristine checkout. They assume the first
proxy-authenticated user becomes an admin, which only holds while the
user table has no admin in it, so they depend on how many users earlier
tests created.

One thing for review: the download loop uses `download_url` like the
other three clients, which holds each file in memory. I kept the
existing pattern rather than do something different in a new-provider
PR, but say so if you'd rather it streamed here.

Closes #1315
This commit is contained in:
splitsec2
2026-10-03 07:55:36 -04:00
committed by GitHub
parent f773ac6029
commit cad18b8019
8 changed files with 1673 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/environment-variables.md
+11 -1
View File
@@ -1703,6 +1703,7 @@ How long to keep cached search results before they expire.
| `PROWLARR_TORRENT_CLIENT` | Choose which torrent client to use | string (choice) | _empty string_ |
| `BLACKHOLE_DIRECTORY` | Directory where Shelfmark saves .torrent files for another downloader | string | _none_ |
| `ALLDEBRID_API_KEY` | AllDebrid API Key (apiv4) from your AllDebrid account settings | string (secret) | _none_ |
| `DEBRIDLINK_API_KEY` | Debrid-Link API Key from your Debrid-Link account settings | string (secret) | _none_ |
| `REALDEBRID_API_KEY` | Real-Debrid API Key (Secret Token) from your Real-Debrid account settings | string (secret) | _none_ |
| `TORBOX_API_KEY` | TorBox API Key from your TorBox account settings | string (secret) | _none_ |
| `QBITTORRENT_URL` | Web UI URL of your qBittorrent instance | string | _none_ |
@@ -1756,7 +1757,7 @@ Choose which torrent client to use
- **Type:** string (choice)
- **Default:** _empty string_
- **Options:** `""` (None), `alldebrid` (AllDebrid), `blackhole` (Blackhole), `qbittorrent` (qBittorrent), `realdebrid` (Real-Debrid), `torbox` (TorBox), `transmission` (Transmission), `deluge` (Deluge), `rtorrent` (rTorrent)
- **Options:** `""` (None), `alldebrid` (AllDebrid), `blackhole` (Blackhole), `debridlink` (Debrid-Link), `qbittorrent` (qBittorrent), `realdebrid` (Real-Debrid), `torbox` (TorBox), `transmission` (Transmission), `deluge` (Deluge), `rtorrent` (rTorrent)
#### `BLACKHOLE_DIRECTORY`
@@ -1776,6 +1777,15 @@ AllDebrid API Key (apiv4) from your AllDebrid account settings
- **Type:** string (secret)
- **Default:** _none_
#### `DEBRIDLINK_API_KEY`
**API Key**
Debrid-Link API Key from your Debrid-Link account settings
- **Type:** string (secret)
- **Default:** _none_
#### `REALDEBRID_API_KEY`
**API Key**
shelfmark/download/clients/__init__.py
+1
View File
@@ -378,6 +378,7 @@ ClientType = TypeVar("ClientType", bound=DownloadClient)
_BUILTIN_CLIENT_MODULES = (
"shelfmark.download.clients.alldebrid",
"shelfmark.download.clients.blackhole",
"shelfmark.download.clients.debridlink",
"shelfmark.download.clients.deluge",
"shelfmark.download.clients.nzbget",
"shelfmark.download.clients.qbittorrent",
shelfmark/download/clients/debridlink.py
+811
View File
@@ -0,0 +1,811 @@
"""Debrid-Link debrid service client for Shelfmark.
Routes magnet links through the Debrid-Link REST API (v2) to download torrent
content via Debrid-Link's seedbox infrastructure.
Unlike the other debrid services, a completed Debrid-Link torrent already
carries a direct ``downloadUrl`` on every file, so there is no per-file
unrestrict or link-request round trip.
Everything here follows the documented v2 fields only: ``status``,
``downloadPercent``, ``wait``, ``isZip``, ``srvMaint`` and the per-file
``downloadPercent``. Values of ``status`` outside the documented set (the docs'
own example shows 6) are tolerated and fall back to ``downloadPercent``.
"""
from __future__ import annotations
import shutil
import threading
import time
from dataclasses import dataclass, field
from typing import TYPE_CHECKING, Any, ClassVar, NoReturn
import requests
from shelfmark.config.env import TMP_DIR
from shelfmark.core.config import config
from shelfmark.core.logger import setup_logger
from shelfmark.download.clients import (
DownloadClient,
DownloadState,
DownloadStatus,
register_client,
)
from shelfmark.download.clients._coercion import config_text
from shelfmark.download.clients.torrent_utils import (
DebridMagnet,
DebridUpload,
resolve_debrid_upload,
safe_relative_path,
)
from shelfmark.download.http import download_url
from shelfmark.download.network import get_ssl_verify
if TYPE_CHECKING:
from pathlib import Path
logger = setup_logger(__name__)
_API_BASE = "https://debrid-link.com/api/v2"
_DEBRIDLINK_CLIENT_ERRORS = (
AttributeError,
OSError,
requests.exceptions.RequestException,
RuntimeError,
TypeError,
ValueError,
)
# Timeouts for API calls.
_API_TIMEOUT = 30
_STATUS_TIMEOUT = 15
# A torrent or file reports 0-100 in downloadPercent; 100 means it is ready.
_COMPLETE_PERCENT = 100
# Documented torrent status values (seedbox-list): 0 paused, 1 queued,
# 2 verification, 4 downloading, 8 seeding, 100 finished.
_STATUS_PAUSED = 0
_STATUS_QUEUED = 1
_STATUS_VERIFYING = 2
_STATUS_SEEDING = 8
_STATUS_FINISHED = 100
_DONE_STATUSES = frozenset({_STATUS_SEEDING, _STATUS_FINISHED})
# The download handler polls every couple of seconds. Debrid-Link rate-limits per
# endpoint and answers a breach with floodDetected, which blocks the API for an hour,
# so the seedbox is asked at most this often per torrent and cached status is served
# in between.
_MIN_REFRESH_SECONDS = 15.0
_FLOOD_BACKOFF_SECONDS = 3600.0
# Errors Debrid-Link documents as temporary. They, network failures and 5xx
# responses are retried until the torrent has been failing for this long.
_TRANSIENT_ERROR_CODES = frozenset({"internalError", "server_error", "freeServerOverload"})
_TRANSIENT_GRACE_SECONDS = 600.0
# A torrent whose progress has not moved for this long is treated as dead (no seeds,
# or left paused), because the API documents no error status for a torrent.
_STALL_TIMEOUT_SECONDS = 3600.0
# How long remove() waits for the retrieval thread to notice a cancel.
_WORKER_JOIN_TIMEOUT = 10.0
# File extensions recognised as book or audiobook content.
_BOOK_EXTENSIONS = (
".aac",
".azw",
".azw3",
".cbr",
".cbz",
".djvu",
".doc",
".docx",
".epub",
".fb2",
".flac",
".lit",
".m4a",
".m4b",
".mobi",
".mp3",
".mp4",
".ogg",
".opus",
".pdf",
".rtf",
".txt",
".wma",
# A torrent with many files can come back as a single ZIP of all of them; the
# post-processing step extracts it.
".zip",
)
class DebridLinkAPIError(RuntimeError):
"""An error Debrid-Link reported, carrying its documented error code."""
def __init__(self, message: str, code: str | None = None) -> None:
super().__init__(message)
self.code = code
def _raise_runtime_error(message: str) -> NoReturn:
raise RuntimeError(message)
@dataclass
class _DownloadState:
"""Internal mutable state for an in-progress Debrid-Link download."""
torrent_id: str
name: str
target_dir: Path
phase: str = "uploading"
error_message: str | None = None
progress: float = 0.0
download_thread: threading.Thread | None = None
lock: threading.Lock = field(default_factory=threading.Lock)
cancel_event: threading.Event = field(default_factory=threading.Event)
# Throttling: the last seedbox record and when it was fetched.
last_info: dict[str, Any] | None = None
last_fetch_at: float = 0.0
# Transient-failure tracking: when the current run of failures started.
failing_since: float | None = None
# Stall tracking: the highest progress seen and when it last moved.
best_percent: float = -1.0
# Set once a torrent held for file selection (``wait``) has been told to start.
start_requested: bool = False
progress_moved_at: float = field(default_factory=time.monotonic)
@register_client("torrent")
class DebridLinkClient(DownloadClient):
"""Debrid-Link debrid service client.
Downloads torrent content by handing a magnet to Debrid-Link's seedbox,
polling until it holds every file, then fetching each file over HTTP from
the ``downloadUrl`` the seedbox already returned.
API documentation: https://debrid-link.com/api_doc/v2/introduction
"""
protocol = "torrent"
name = "debridlink"
_downloads: ClassVar[dict[str, _DownloadState]] = {}
_downloads_lock = threading.Lock()
# floodDetected blocks the whole account's API use, so the back-off is shared.
_flood_until: ClassVar[float] = 0.0
def __init__(self) -> None:
self._api_key = config_text(config.get("DEBRIDLINK_API_KEY", ""))
def _auth_headers(self) -> dict[str, str]:
"""Return Authorization header dict for API requests."""
return {"Authorization": f"Bearer {self._api_key}"}
# ------------------------------------------------------------------
# DownloadClient interface
# ------------------------------------------------------------------
@staticmethod
def is_configured() -> bool:
"""Return True when Debrid-Link is selected and an API key exists."""
client = config_text(config.get("PROWLARR_TORRENT_CLIENT", ""))
api_key = config_text(config.get("DEBRIDLINK_API_KEY", ""))
return client == "debridlink" and bool(api_key)
def test_connection(self) -> tuple[bool, str]:
"""Validate the API key and check the account still has premium time."""
if not self._api_key:
return False, "Debrid-Link API Key is required"
try:
account = self._request_value(
"GET", "/account/infos", operation="account check", timeout=_STATUS_TIMEOUT
)
if not isinstance(account, dict):
return False, "Unexpected response from Debrid-Link account endpoint"
username = str(account.get("username") or account.get("email") or "Unknown")
# premiumLeft is seconds of premium remaining; 0 means the account expired.
if not account.get("premiumLeft"):
return (
False,
f"Debrid-Link user '{username}' does not have an active Premium subscription",
)
except _DEBRIDLINK_CLIENT_ERRORS as e:
return False, f"Connection failed: {e}"
else:
return True, f"Connected to Debrid-Link as '{username}' (Premium)"
def add_download(
self,
url: str,
name: str,
category: str | None = None,
expected_hash: str | None = None,
**kwargs: object,
) -> str:
"""Send a torrent to Debrid-Link's seedbox.
Accepts a magnet link, a .torrent URL, or an indexer proxy URL. Anything
that is not already a magnet is resolved first, then posted as a file,
because the seedbox endpoint takes a magnet or hash in ``url`` and a
.torrent only as multipart form data.
"""
if not self._api_key:
msg = "Debrid-Link API key is not configured"
raise RuntimeError(msg)
try:
upload = resolve_debrid_upload(url, expected_hash=expected_hash)
data = self._send_torrent(upload)
torrent_id = str(data.get("id", ""))
if not torrent_id:
msg = "No torrent ID returned from Debrid-Link"
_raise_runtime_error(msg)
target_dir = TMP_DIR / f"debridlink_{torrent_id}"
target_dir.mkdir(parents=True, exist_ok=True)
state = _DownloadState(
torrent_id=torrent_id,
name=name,
target_dir=target_dir,
phase="waiting_dl",
)
with self._downloads_lock:
self._downloads[torrent_id] = state
logger.info(
"Added torrent to Debrid-Link: ID %s (%s)",
torrent_id,
name,
)
except Exception:
logger.exception("Failed to add torrent to Debrid-Link")
raise
else:
return torrent_id
def _send_torrent(self, upload: DebridUpload) -> dict[str, Any]:
"""Hand the torrent to the seedbox, as a magnet or as a file upload."""
if isinstance(upload, DebridMagnet):
data = self._request_value(
"POST",
"/seedbox/add",
operation="torrent upload",
json={"url": upload.magnet_url},
timeout=_API_TIMEOUT,
)
else:
# A .torrent must go as multipart/form-data under the "file" field;
# the JSON body only accepts a magnet, a hash or a torrent URL.
data = self._request_value(
"POST",
"/seedbox/add",
operation="torrent upload",
files={"file": ("upload.torrent", upload.torrent_data, "application/x-bittorrent")},
timeout=_API_TIMEOUT,
)
if not isinstance(data, dict):
msg = "Unexpected response when adding a torrent to Debrid-Link"
_raise_runtime_error(msg)
return data
def get_status(self, download_id: str) -> DownloadStatus:
"""Poll Debrid-Link for torrent status and drive the download."""
state = self._ensure_state(download_id)
# Return cached terminal / in-flight states immediately.
with state.lock:
if state.phase == "error":
return DownloadStatus.error(state.error_message or "Debrid-Link error")
if state.phase == "complete":
return DownloadStatus(
progress=100.0,
state=DownloadState.COMPLETE,
message="Complete",
complete=True,
file_path=str(state.target_dir),
)
if state.phase == "downloading_http":
return DownloadStatus(
progress=state.progress,
state=DownloadState.DOWNLOADING,
message="Downloading files via HTTP...",
complete=False,
file_path=None,
)
cached = state.last_info
fresh = time.monotonic() - state.last_fetch_at < _MIN_REFRESH_SECONDS
now = time.time()
if now < type(self)._flood_until:
resume = time.strftime("%H:%M", time.localtime(type(self)._flood_until))
return self._waiting_status(
cached, f"Debrid-Link rate limit reached, retrying at {resume}"
)
if fresh and cached is not None:
return self._handle_torrent_info(cached, state, observed=False)
try:
torrent = self._fetch_torrent(download_id)
except _DEBRIDLINK_CLIENT_ERRORS as e:
return self._handle_fetch_failure(e, state)
with state.lock:
state.failing_since = None
state.last_fetch_at = time.monotonic()
state.last_info = torrent
if torrent is None:
return self._set_error(state, f"Torrent {download_id} is no longer on Debrid-Link")
return self._handle_torrent_info(torrent, state)
def remove(
self,
download_id: str,
*,
delete_files: bool = False,
) -> bool:
"""Stop any retrieval, delete the torrent from Debrid-Link and clean up locally."""
remote_removed = True
try:
self._request_value(
"DELETE",
f"/seedbox/{download_id}/remove",
operation="torrent deletion",
timeout=_STATUS_TIMEOUT,
)
except _DEBRIDLINK_CLIENT_ERRORS as e:
remote_removed = False
logger.warning("Failed to delete torrent from Debrid-Link: %s", e)
with self._downloads_lock:
state = self._downloads.get(download_id)
if state:
with state.lock:
state.cancel_event.set()
thread = state.download_thread
if thread and thread is not threading.current_thread():
thread.join(_WORKER_JOIN_TIMEOUT)
if thread.is_alive():
logger.warning(
"Debrid-Link retrieval thread did not stop; deferring cleanup of %s",
download_id,
)
return False
with self._downloads_lock:
self._downloads.pop(download_id, None)
target_dir = state.target_dir if state else TMP_DIR / f"debridlink_{download_id}"
local_removed = True
if target_dir.exists():
try:
shutil.rmtree(target_dir)
except OSError:
local_removed = False
logger.warning("Failed to remove Debrid-Link temporary files for %s", download_id)
return remote_removed and local_removed
def get_download_path(self, download_id: str) -> str | None:
"""Return the local directory containing downloaded files."""
with self._downloads_lock:
state = self._downloads.get(download_id)
if state and state.phase == "complete":
return str(state.target_dir)
target_dir = TMP_DIR / f"debridlink_{download_id}"
if target_dir.exists():
return str(target_dir)
return None
# ------------------------------------------------------------------
# Internal helpers
# ------------------------------------------------------------------
def _request_value(
self,
method: str,
path: str,
*,
operation: str,
timeout: int = _API_TIMEOUT,
**kwargs: Any,
) -> Any:
"""Call the API and unwrap Debrid-Link's ``{success, value}`` envelope.
Errors arrive as a 4xx or 5xx with ``{success: false, error: <code>}``, so
the body is read before the status code: raising on the status alone would
turn ``badToken`` into a bare "401 Client Error". Messages never include the
request URL or its parameters.
"""
url = f"{_API_BASE}{path}"
try:
resp = requests.request(
method,
url,
headers=self._auth_headers(),
timeout=timeout,
verify=get_ssl_verify(url),
**kwargs,
)
except requests.exceptions.RequestException as e:
msg = f"Debrid-Link {operation} failed: {type(e).__name__}"
raise RuntimeError(msg) from None
status_code = getattr(resp, "status_code", 200)
ok_status = isinstance(status_code, int) and 200 <= status_code < 300
try:
payload = resp.json()
except TypeError, ValueError:
payload = None
if not isinstance(payload, dict):
detail = "an invalid response" if ok_status else f"HTTP {status_code}"
msg = f"Debrid-Link {operation} failed: {detail}"
raise DebridLinkAPIError(msg, None if ok_status else f"http{status_code}")
if not ok_status or payload.get("success") is not True:
# The documented error body is just {success: false, error: <code>}.
code = payload.get("error")
code = code if isinstance(code, str) and code else None
if code == "floodDetected":
type(self)._flood_until = time.time() + _FLOOD_BACKOFF_SECONDS
detail = code or ("an unsuccessful response" if ok_status else f"HTTP {status_code}")
msg = f"Debrid-Link {operation} failed: {detail}"
raise DebridLinkAPIError(msg, code or (None if ok_status else f"http{status_code}"))
return payload.get("value")
def _fetch_torrent(self, download_id: str) -> dict[str, Any] | None:
"""Return the seedbox record for one torrent, or None when it is gone."""
try:
value = self._request_value(
"GET",
"/seedbox/list",
operation="status check",
params={"ids": download_id},
timeout=_STATUS_TIMEOUT,
)
except DebridLinkAPIError as e:
if e.code == "badId":
return None
raise
return self._pick_torrent(value, download_id)
def _fetch_file_list(self, download_id: str) -> list[dict[str, Any]]:
"""Return the torrent's individual files.
A torrent with many files is listed as one ZIP (``isZip``) under ``ids``;
Debrid-Link documents ``?id=TORRENT_ID`` as the way to see the files
themselves. Fetched once, after the torrent is complete, so the list is
final rather than the partial one an early status check can return.
"""
value = self._request_value(
"GET",
"/seedbox/list",
operation="file listing",
params={"id": download_id},
timeout=_STATUS_TIMEOUT,
)
torrent = self._pick_torrent(value, download_id)
if torrent is None:
msg = f"Torrent {download_id} is no longer on Debrid-Link"
_raise_runtime_error(msg)
return [f for f in torrent.get("files", []) if isinstance(f, dict)]
@staticmethod
def _pick_torrent(value: object, download_id: str) -> dict[str, Any] | None:
entries = value if isinstance(value, list) else [value]
for entry in entries:
if isinstance(entry, dict) and str(entry.get("id", "")) == download_id:
return entry
return None
def _handle_fetch_failure(self, error: Exception, state: _DownloadState) -> DownloadStatus:
"""Fail on a permanent error; ride out a temporary one for a grace period."""
code = getattr(error, "code", None)
if code == "floodDetected":
resume = time.strftime("%H:%M", time.localtime(type(self)._flood_until))
logger.warning("Debrid-Link rate limit reached; pausing status checks until %s", resume)
return self._waiting_status(
state.last_info, f"Debrid-Link rate limit reached, retrying at {resume}"
)
transient = (
not isinstance(error, DebridLinkAPIError)
or code in _TRANSIENT_ERROR_CODES
or (isinstance(code, str) and code.startswith("http5"))
)
if not transient:
return self._set_error(state, str(error))
now = time.monotonic()
with state.lock:
if state.failing_since is None:
state.failing_since = now
failing_for = now - state.failing_since
state.last_fetch_at = now
if failing_for >= _TRANSIENT_GRACE_SECONDS:
minutes = int(_TRANSIENT_GRACE_SECONDS // 60)
return self._set_error(state, f"{error} (still failing after {minutes} minutes)")
logger.warning("Debrid-Link status check failed, will retry: %s", error)
return self._waiting_status(state.last_info, "Debrid-Link unreachable, retrying")
@staticmethod
def _waiting_status(info: dict[str, Any] | None, message: str) -> DownloadStatus:
percent = _as_float(info.get("downloadPercent")) if info else 0.0
return DownloadStatus(
progress=min(percent, _COMPLETE_PERCENT) * 0.5,
state=DownloadState.DOWNLOADING,
message=message,
complete=False,
file_path=None,
)
@staticmethod
def _set_error(state: _DownloadState, message: str) -> DownloadStatus:
with state.lock:
state.phase = "error"
state.error_message = message
return DownloadStatus.error(message)
def _ensure_state(self, download_id: str) -> _DownloadState:
"""Get or create download state for the given torrent ID."""
with self._downloads_lock:
state = self._downloads.get(download_id)
if state:
return state
target_dir = TMP_DIR / f"debridlink_{download_id}"
state = _DownloadState(
torrent_id=download_id,
name=f"Download {download_id}",
target_dir=target_dir,
phase="waiting_dl",
)
with self._downloads_lock:
self._downloads[download_id] = state
return state
def _handle_torrent_info(
self,
info: dict[str, Any],
state: _DownloadState,
*,
observed: bool = True,
) -> DownloadStatus:
"""Map a Debrid-Link seedbox torrent to a DownloadStatus.
``observed`` is False when ``info`` is a cached record being re-served
between throttled fetches, so it doesn't count toward stall detection.
"""
percent = _as_float(info.get("downloadPercent"))
name = str(info.get("name") or state.name)
files = [f for f in info.get("files", []) if isinstance(f, dict)]
status = _as_status(info.get("status"))
ready = (
(status in _DONE_STATUSES or percent >= _COMPLETE_PERCENT)
and bool(files)
and all(_as_float(f.get("downloadPercent", 100)) >= _COMPLETE_PERCENT for f in files)
)
if ready:
self._maybe_start_download_thread(state)
return DownloadStatus(
progress=50.0,
state=DownloadState.DOWNLOADING,
message="Debrid-Link ready, retrieving files...",
complete=False,
file_path=None,
)
now = time.monotonic()
with state.lock:
if percent > state.best_percent:
state.best_percent = percent
state.progress_moved_at = now
stalled_for = now - state.progress_moved_at
if observed and stalled_for >= _STALL_TIMEOUT_SECONDS:
minutes = int(_STALL_TIMEOUT_SECONDS // 60)
return self._set_error(
state,
f"Debrid-Link made no progress on {name} for {minutes} minutes "
f"({int(_as_float(info.get('peersConnected')))} peers connected)",
)
if info.get("wait"):
# Held for file selection. Shelfmark wants every file, so start it.
self._start_waiting_torrent(state)
message = f"Starting on Debrid-Link ({name})"
elif info.get("srvMaint"):
message = "Debrid-Link server maintenance, waiting"
elif status == _STATUS_PAUSED:
message = f"Paused on Debrid-Link ({name})"
elif status == _STATUS_QUEUED:
message = f"Queued on Debrid-Link ({name})"
elif status == _STATUS_VERIFYING:
message = f"Debrid-Link verifying torrent ({name})"
else:
message = f"Debrid-Link downloading torrent ({name})"
# Still being fetched by the seedbox: the first half of the progress bar.
return DownloadStatus(
progress=min(percent, _COMPLETE_PERCENT) * 0.5,
state=DownloadState.DOWNLOADING,
message=message,
complete=False,
file_path=None,
download_speed=int(_as_float(info.get("downloadSpeed"))),
)
def _start_waiting_torrent(self, state: _DownloadState) -> None:
"""Release a torrent held for file selection, keeping every file.
Shelfmark never asks for ``wait``, but a torrent reported that way would
otherwise never start. The config endpoint with no unwanted files starts it.
"""
with state.lock:
if state.start_requested:
return
state.start_requested = True
try:
self._request_value(
"POST",
f"/seedbox/{state.torrent_id}/config",
operation="torrent start",
json={"files-unwanted": []},
timeout=_STATUS_TIMEOUT,
)
except _DEBRIDLINK_CLIENT_ERRORS as e:
logger.warning(
"Could not start waiting Debrid-Link torrent %s: %s", state.torrent_id, e
)
with state.lock:
state.start_requested = False
def _maybe_start_download_thread(self, state: _DownloadState) -> None:
"""Spawn a background thread to download the seedbox's files."""
with state.lock:
already_running = state.phase in ("downloading_http", "complete")
thread_alive = state.download_thread is not None and state.download_thread.is_alive()
if already_running or thread_alive or state.cancel_event.is_set():
return
state.phase = "downloading_http"
t = threading.Thread(
target=self._process_and_download,
args=(state,),
daemon=True,
)
state.download_thread = t
t.start()
# ------------------------------------------------------------------
# File download pipeline
# ------------------------------------------------------------------
def _process_and_download(
self,
state: _DownloadState,
files: list[dict[str, Any]] | None = None,
) -> None:
"""Download each file over HTTP.
Runs in a background thread spawned by ``_maybe_start_download_thread``.
Debrid-Link puts a ready-to-use ``downloadUrl`` on every file, so unlike
the other debrid clients there is nothing to unrestrict here. ``files`` is
fetched fresh when not given.
"""
try:
if files is None:
files = self._fetch_file_list(state.torrent_id)
if state.cancel_event.is_set():
return
downloadable = [f for f in files if f.get("downloadUrl")]
if not downloadable:
msg = "No download links returned by Debrid-Link"
_raise_runtime_error(msg)
relevant = [
f for f in downloadable if str(f.get("name", "")).lower().endswith(_BOOK_EXTENSIONS)
]
if not relevant:
relevant = downloadable
state.target_dir.mkdir(parents=True, exist_ok=True)
used: set[Path] = set()
total = len(relevant)
for idx, file_info in enumerate(relevant):
if state.cancel_event.is_set():
return
name = str(file_info.get("name") or f"file_{idx + 1}")
rel_path = _unique_path(
safe_relative_path(name, state.target_dir, "Debrid-Link"), used
)
dest = state.target_dir / rel_path
dest.parent.mkdir(parents=True, exist_ok=True)
logger.info("Downloading Debrid-Link file %d/%d: %s", idx + 1, total, rel_path)
buf = download_url(
str(file_info.get("downloadUrl", "")),
referer="https://debrid-link.com/",
cancel_flag=state.cancel_event,
)
if state.cancel_event.is_set():
return
if not buf:
# The URL is signed and stays out of the message, which ends up
# in logs and the UI.
msg = f"Failed to download {rel_path} from Debrid-Link"
_raise_runtime_error(msg)
with dest.open("wb") as fh:
buf.seek(0)
shutil.copyfileobj(buf, fh)
with state.lock:
state.progress = 50.0 + (idx + 1) / total * 50.0
with state.lock:
if state.cancel_event.is_set():
return
state.phase = "complete"
state.progress = 100.0
logger.info(
"Debrid-Link download complete for ID %s at %s",
state.torrent_id,
state.target_dir,
)
except Exception as e:
logger.exception("Error in Debrid-Link download for ID %s", state.torrent_id)
with state.lock:
state.phase = "error"
state.error_message = str(e) or "Download failed"
def _unique_path(path: Path, used: set[Path]) -> Path:
"""Keep two files with the same name from overwriting each other."""
candidate = path
counter = 2
while candidate in used:
candidate = path.with_name(f"{path.stem} ({counter}){path.suffix}")
counter += 1
used.add(candidate)
return candidate
def _as_status(value: object) -> int | None:
"""The torrent's numeric status, or None when it is missing or not a number."""
if isinstance(value, bool):
return None
if isinstance(value, int):
return value
if isinstance(value, str) and value.strip().isdigit():
return int(value)
return None
def _as_float(value: object) -> float:
"""Coerce an API numeric field to a float, treating anything odd as zero."""
if isinstance(value, bool):
return 0.0
if isinstance(value, (int, float)):
return float(value)
if isinstance(value, str):
try:
return float(value)
except ValueError:
return 0.0
return 0.0
shelfmark/download/clients/settings.py
+33
View File
@@ -550,6 +550,23 @@ def _test_alldebrid_connection(current_values: dict[str, Any] | None = None) ->
return {"success": success, "message": message}
def _test_debridlink_connection(current_values: dict[str, Any] | None = None) -> dict[str, Any]:
"""Test the Debrid-Link API connection using current form values."""
from shelfmark.core.config import config
from shelfmark.download.clients.debridlink import DebridLinkClient
current_values = current_values or {}
api_key = _resolve_string_setting(current_values, config.get, "DEBRIDLINK_API_KEY")
if not api_key:
return {"success": False, "message": "Debrid-Link API Key is required"}
client = DebridLinkClient()
client._api_key = api_key
success, message = client.test_connection()
return {"success": success, "message": message}
def _test_realdebrid_connection(current_values: dict[str, Any] | None = None) -> dict[str, Any]:
"""Test the Real-Debrid API connection using current form values."""
from shelfmark.core.config import config
@@ -610,6 +627,7 @@ def prowlarr_clients_settings() -> list[SettingsField]:
{"value": "", "label": "None"},
{"value": "alldebrid", "label": "AllDebrid"},
{"value": "blackhole", "label": "Blackhole"},
{"value": "debridlink", "label": "Debrid-Link"},
{"value": "qbittorrent", "label": "qBittorrent"},
{"value": "realdebrid", "label": "Real-Debrid"},
{"value": "torbox", "label": "TorBox"},
@@ -641,6 +659,21 @@ def prowlarr_clients_settings() -> list[SettingsField]:
callback=_test_alldebrid_connection,
show_when={"field": "PROWLARR_TORRENT_CLIENT", "value": "alldebrid"},
),
# --- Debrid-Link Settings ---
PasswordField(
key="DEBRIDLINK_API_KEY",
label="API Key",
description="Debrid-Link API Key from your Debrid-Link account settings",
show_when={"field": "PROWLARR_TORRENT_CLIENT", "value": "debridlink"},
),
ActionButton(
key="test_debridlink",
label="Test Connection",
description="Verify your Debrid-Link configuration",
style="primary",
callback=_test_debridlink_connection,
show_when={"field": "PROWLARR_TORRENT_CLIENT", "value": "debridlink"},
),
# --- Real-Debrid Settings ---
PasswordField(
key="REALDEBRID_API_KEY",
shelfmark/download/clients/torbox.py
+6 -25
View File
@@ -7,8 +7,7 @@ import shutil
import threading
import time
from dataclasses import dataclass, field
from pathlib import Path, PurePosixPath, PureWindowsPath
from typing import Any, ClassVar, NoReturn
from typing import TYPE_CHECKING, Any, ClassVar, NoReturn
from urllib.parse import urlparse
import requests
@@ -27,10 +26,14 @@ from shelfmark.download.clients.torrent_utils import (
DebridMagnet,
DebridUpload,
resolve_debrid_upload,
safe_relative_path,
)
from shelfmark.download.http import download_url
from shelfmark.download.network import get_ssl_verify
if TYPE_CHECKING:
from pathlib import Path
logger = setup_logger(__name__)
_API_BASE = "https://api.torbox.app/v1/api"
@@ -605,29 +608,7 @@ class TorBoxClient(DownloadClient):
@classmethod
def _safe_relative_path(cls, file_info: dict[str, Any], target_dir: Path) -> Path:
"""Validate external file metadata before writing below ``target_dir``."""
name = cls._file_name(file_info)
if not name:
_raise_runtime_error("TorBox returned a file without a name")
normalized = name.replace("\\", "/")
relative_path = PurePosixPath(normalized)
windows_path = PureWindowsPath(name)
if (
relative_path.is_absolute()
or windows_path.is_absolute()
or windows_path.drive
or ".." in relative_path.parts
):
_raise_runtime_error(f"TorBox returned an unsafe file path: {name}")
if relative_path == PurePosixPath("."):
_raise_runtime_error("TorBox returned a file without a usable name")
destination = (target_dir / Path(*relative_path.parts)).resolve()
try:
destination.relative_to(target_dir.resolve())
except ValueError:
_raise_runtime_error(f"TorBox returned an unsafe file path: {name}")
return Path(*relative_path.parts)
return safe_relative_path(cls._file_name(file_info), target_dir, "TorBox")
@staticmethod
def _set_error(state: _DownloadState, message: str) -> None:
shelfmark/download/clients/torrent_utils.py
+35
View File
@@ -8,6 +8,7 @@ import re
import time
from binascii import Error as BinasciiError
from dataclasses import dataclass
from pathlib import Path, PurePosixPath, PureWindowsPath
from threading import Lock
from urllib.parse import ParseResult, parse_qs, urljoin, urlparse, urlunparse
@@ -588,3 +589,37 @@ def extract_hash_from_magnet(magnet_url: str) -> str | None:
return btmh_hash
return None
def safe_relative_path(name: str, target_dir: Path, service: str) -> Path:
"""Validate a service-supplied file name before anything is written below ``target_dir``.
Debrid services pass torrent file names straight through, and a torrent can name a
file ``../../etc/cron.d/x``. Absolute paths, drive letters and ``..`` segments are
refused, and the resolved destination must stay inside ``target_dir``.
"""
if not name:
msg = f"{service} returned a file without a name"
raise RuntimeError(msg)
relative_path = PurePosixPath(name.replace("\\", "/"))
windows_path = PureWindowsPath(name)
if (
relative_path.is_absolute()
or windows_path.is_absolute()
or windows_path.drive
or ".." in relative_path.parts
):
msg = f"{service} returned an unsafe file path: {name}"
raise RuntimeError(msg)
if relative_path == PurePosixPath("."):
msg = f"{service} returned a file without a usable name"
raise RuntimeError(msg)
destination = (target_dir / Path(*relative_path.parts)).resolve()
try:
destination.relative_to(target_dir.resolve())
except ValueError:
msg = f"{service} returned an unsafe file path: {name}"
raise RuntimeError(msg) from None
return Path(*relative_path.parts)
tests/prowlarr/test_debridlink_client.py
+727
View File
@@ -0,0 +1,727 @@
"""Debrid-Link seedbox client tests.
Two things set this client apart from the other debrid services and both are
covered here: every v2 endpoint wraps its payload in a ``{success, value}``
envelope that can report failure over an HTTP 200, and a completed torrent
already carries a direct ``downloadUrl`` per file, so there is no unrestrict
step to mock.
"""
import hashlib
import io
from unittest.mock import MagicMock
import pytest
import requests
from shelfmark.download.clients import DownloadState, debridlink
from shelfmark.download.clients.debridlink import (
DebridLinkAPIError,
DebridLinkClient,
_as_float,
_DownloadState,
)
from shelfmark.download.clients.torrent_utils import bencode_encode
_PROWLARR_PROXY_URL = "https://prowlarr.example/api/v1/indexer/1/download?apikey=k&link=abc"
_MAGNET = "magnet:?xt=urn:btih:0123456789abcdef0123456789abcdef01234567&dn=Dune"
def _valid_torrent() -> tuple[bytes, str]:
info_dict = {
b"name": b"book.epub",
b"length": 100,
b"piece length": 16384,
b"pieces": b"\x00" * 20,
}
return (
bencode_encode({b"info": info_dict}),
hashlib.sha1(bencode_encode(info_dict)).hexdigest().lower(),
)
def _mock_fetch(monkeypatch, *, content=b"", status_code=200, error=None):
"""Stand in for the .torrent prefetch inside extract_torrent_info."""
if error is not None:
mock_get = MagicMock(side_effect=error)
else:
response = MagicMock(status_code=status_code, content=content)
response.raise_for_status = MagicMock()
mock_get = MagicMock(return_value=response)
monkeypatch.setattr("shelfmark.download.clients.torrent_utils.requests.get", mock_get)
return mock_get
def _mock_request(monkeypatch, payload, *, status_code=200):
"""Patch requests.request with a single canned Debrid-Link envelope."""
response = MagicMock(status_code=status_code)
response.raise_for_status = MagicMock()
response.json = MagicMock(return_value=payload)
request = MagicMock(return_value=response)
monkeypatch.setattr("shelfmark.download.clients.debridlink.requests.request", request)
return request
@pytest.fixture(autouse=True)
def _reset_shared_state():
"""The flood back-off and the download table are shared across instances."""
DebridLinkClient._flood_until = 0.0
DebridLinkClient._downloads.clear()
yield
DebridLinkClient._flood_until = 0.0
DebridLinkClient._downloads.clear()
def _state(tmp_path, phase="waiting_dl"):
return _DownloadState(torrent_id="DL1", name="Dune", target_dir=tmp_path, phase=phase)
def _buffer(data=b"data"):
return io.BytesIO(data)
def _client(monkeypatch, api_key="dl-key"):
monkeypatch.setattr(
"shelfmark.download.clients.debridlink.config.get",
lambda key, default="": {"DEBRIDLINK_API_KEY": api_key}.get(key, default),
)
return DebridLinkClient()
class TestEnvelope:
"""The body decides the outcome, and error codes survive a 4xx or 5xx."""
def test_success_false_raises_even_on_http_200(self, monkeypatch):
_mock_request(monkeypatch, {"success": False, "error": "badToken"})
client = _client(monkeypatch)
with pytest.raises(DebridLinkAPIError, match="failed: badToken") as err:
client._request_value("GET", "/account/infos", operation="account check")
assert err.value.code == "badToken"
def test_the_error_code_is_read_from_a_4xx_body(self, monkeypatch):
_mock_request(monkeypatch, {"success": False, "error": "badToken"}, status_code=401)
client = _client(monkeypatch)
with pytest.raises(DebridLinkAPIError, match="badToken") as err:
client._request_value("GET", "/account/infos", operation="account check")
assert err.value.code == "badToken"
def test_a_5xx_without_a_body_keeps_the_status(self, monkeypatch):
response = MagicMock(status_code=503)
response.json = MagicMock(side_effect=ValueError("no json"))
monkeypatch.setattr(
"shelfmark.download.clients.debridlink.requests.request",
MagicMock(return_value=response),
)
client = _client(monkeypatch)
with pytest.raises(DebridLinkAPIError, match="HTTP 503") as err:
client._request_value("GET", "/seedbox/list", operation="status check")
assert err.value.code == "http503"
def test_success_unwraps_the_value(self, monkeypatch):
_mock_request(monkeypatch, {"success": True, "value": {"id": "DL1"}})
client = _client(monkeypatch)
assert client._request_value("GET", "/seedbox/list", operation="status check") == {
"id": "DL1"
}
def test_a_non_object_body_is_rejected(self, monkeypatch):
_mock_request(monkeypatch, ["unexpected"])
client = _client(monkeypatch)
with pytest.raises(RuntimeError, match="invalid response"):
client._request_value("GET", "/account/infos", operation="account check")
def test_a_network_failure_does_not_leak_the_url(self, monkeypatch):
monkeypatch.setattr(
"shelfmark.download.clients.debridlink.requests.request",
MagicMock(side_effect=requests.ConnectionError("https://debrid-link.com/secret")),
)
client = _client(monkeypatch)
with pytest.raises(RuntimeError) as err:
client._request_value("GET", "/seedbox/list", operation="status check")
assert "debrid-link.com" not in str(err.value)
assert "ConnectionError" in str(err.value)
def test_flood_detected_starts_the_shared_back_off(self, monkeypatch):
_mock_request(monkeypatch, {"success": False, "error": "floodDetected"}, status_code=429)
client = _client(monkeypatch)
with pytest.raises(DebridLinkAPIError):
client._request_value("GET", "/seedbox/list", operation="status check")
assert DebridLinkClient._flood_until > 0
class TestAdd:
def test_magnet_is_sent_as_json(self, monkeypatch, tmp_path):
monkeypatch.setattr("shelfmark.download.clients.debridlink.TMP_DIR", tmp_path)
request = _mock_request(monkeypatch, {"success": True, "value": {"id": "DL1"}})
client = _client(monkeypatch)
assert client.add_download(_MAGNET, "Dune") == "DL1"
request.assert_called_once()
assert request.call_args.args[0] == "POST"
assert request.call_args.args[1].endswith("/seedbox/add")
assert request.call_args.kwargs["json"] == {"url": _MAGNET}
assert "files" not in request.call_args.kwargs
def test_proxy_url_is_uploaded_as_a_torrent_file(self, monkeypatch, tmp_path):
monkeypatch.setattr("shelfmark.download.clients.debridlink.TMP_DIR", tmp_path)
torrent_data, _ = _valid_torrent()
_mock_fetch(monkeypatch, content=torrent_data)
request = _mock_request(monkeypatch, {"success": True, "value": {"id": "DL2"}})
client = _client(monkeypatch)
assert client.add_download(_PROWLARR_PROXY_URL, "Dune") == "DL2"
# A .torrent only goes up as multipart; the JSON body takes a magnet.
assert "json" not in request.call_args.kwargs
assert request.call_args.kwargs["files"]["file"][1] == torrent_data
def test_missing_id_is_an_error_not_a_silent_success(self, monkeypatch, tmp_path):
monkeypatch.setattr("shelfmark.download.clients.debridlink.TMP_DIR", tmp_path)
_mock_request(monkeypatch, {"success": True, "value": {}})
client = _client(monkeypatch)
with pytest.raises(RuntimeError, match="No torrent ID returned"):
client.add_download(_MAGNET, "Dune")
def test_unresolvable_url_reports_the_reason(self, monkeypatch, tmp_path):
monkeypatch.setattr("shelfmark.download.clients.debridlink.TMP_DIR", tmp_path)
_mock_fetch(monkeypatch, error=OSError("tracker unreachable"))
request = _mock_request(monkeypatch, {"success": True, "value": {"id": "DL3"}})
client = _client(monkeypatch)
with pytest.raises(ValueError, match="Could not resolve a torrent to send"):
client.add_download(_PROWLARR_PROXY_URL, "Dune")
request.assert_not_called()
def test_without_an_api_key_it_refuses_before_any_call(self, monkeypatch):
request = _mock_request(monkeypatch, {"success": True, "value": {"id": "DL4"}})
client = _client(monkeypatch, api_key="")
with pytest.raises(RuntimeError, match="API key is not configured"):
client.add_download(_MAGNET, "Dune")
request.assert_not_called()
class TestStatus:
@staticmethod
def _bare_client():
return DebridLinkClient.__new__(DebridLinkClient)
def test_partial_progress_is_halved_and_stays_downloading(self, tmp_path):
client = self._bare_client()
state = _state(tmp_path)
status = client._handle_torrent_info(
{"downloadPercent": 40, "name": "Dune.epub", "downloadSpeed": 1234},
state,
)
# The seedbox fetch is the first half of the bar; HTTP is the second.
assert status.progress == 20.0
assert status.state == DownloadState.DOWNLOADING
assert status.complete is False
assert status.download_speed == 1234
assert state.phase == "waiting_dl"
def test_completion_hands_off_only_when_every_file_is_ready(self, tmp_path, monkeypatch):
client = self._bare_client()
state = _state(tmp_path)
started = MagicMock()
monkeypatch.setattr(client, "_maybe_start_download_thread", started)
partial = {
"downloadPercent": 100,
"files": [
{"name": "a.mp3", "downloadPercent": 100},
{"name": "b.mp3", "downloadPercent": 80},
],
}
client._handle_torrent_info(partial, state)
started.assert_not_called()
partial["files"][1]["downloadPercent"] = 100
status = client._handle_torrent_info(partial, state)
assert status.progress == 50.0
started.assert_called_once_with(state)
def test_an_empty_file_list_is_not_treated_as_ready(self, tmp_path, monkeypatch):
client = self._bare_client()
started = MagicMock()
monkeypatch.setattr(client, "_maybe_start_download_thread", started)
client._handle_torrent_info({"downloadPercent": 100, "files": []}, _state(tmp_path))
started.assert_not_called()
@pytest.mark.parametrize(
("info", "expected"),
[
({"status": 0}, "Paused on Debrid-Link"),
({"status": 1}, "Queued on Debrid-Link"),
({"status": 2}, "verifying"),
({"status": 4}, "downloading torrent"),
({"status": 6}, "downloading torrent"), # the docs' own example value
({"srvMaint": True}, "server maintenance"),
],
)
def test_documented_states_are_reported(self, tmp_path, info, expected):
status = self._bare_client()._handle_torrent_info(
{"downloadPercent": 10, **info}, _state(tmp_path)
)
assert status.state == DownloadState.DOWNLOADING
assert expected in (status.message or "")
@pytest.mark.parametrize("done_status", [8, 100])
def test_seeding_or_finished_counts_as_done(self, tmp_path, monkeypatch, done_status):
client = self._bare_client()
started = MagicMock()
monkeypatch.setattr(client, "_maybe_start_download_thread", started)
client._handle_torrent_info(
{
"status": done_status,
"downloadPercent": 99,
"files": [{"name": "a.epub", "downloadPercent": 100}],
},
_state(tmp_path),
)
started.assert_called_once()
def test_a_torrent_held_for_file_selection_is_started_once(self, tmp_path, monkeypatch):
# ``wait`` means "hold for file selection", not "queued": without a config
# call the torrent never starts.
client = _client(monkeypatch)
request = _mock_request(monkeypatch, {"success": True, "value": []})
state = _state(tmp_path)
first = client._handle_torrent_info({"wait": True, "downloadPercent": 0}, state)
client._handle_torrent_info({"wait": True, "downloadPercent": 0}, state)
assert "Starting on Debrid-Link" in (first.message or "")
assert request.call_count == 1
args, kwargs = request.call_args
assert args[0] == "POST"
assert args[1].endswith("/seedbox/DL1/config")
assert kwargs["json"] == {"files-unwanted": []}
def test_a_torrent_that_stops_moving_fails_after_the_stall_timeout(self, tmp_path, monkeypatch):
client = self._bare_client()
state = _state(tmp_path)
clock = [1000.0]
monkeypatch.setattr(debridlink.time, "monotonic", lambda: clock[0])
state.progress_moved_at = clock[0]
client._handle_torrent_info({"downloadPercent": 12, "peersConnected": 0}, state)
clock[0] += debridlink._STALL_TIMEOUT_SECONDS - 1
assert client._handle_torrent_info({"downloadPercent": 12}, state).state == (
DownloadState.DOWNLOADING
)
clock[0] += 2
status = client._handle_torrent_info({"downloadPercent": 12, "peersConnected": 0}, state)
assert status.state == DownloadState.ERROR
assert "no progress" in (status.message or "")
def test_progress_resets_the_stall_clock(self, tmp_path, monkeypatch):
client = self._bare_client()
state = _state(tmp_path)
clock = [1000.0]
monkeypatch.setattr(debridlink.time, "monotonic", lambda: clock[0])
state.progress_moved_at = clock[0]
client._handle_torrent_info({"downloadPercent": 12}, state)
clock[0] += debridlink._STALL_TIMEOUT_SECONDS - 1
client._handle_torrent_info({"downloadPercent": 13}, state)
clock[0] += debridlink._STALL_TIMEOUT_SECONDS - 1
assert client._handle_torrent_info({"downloadPercent": 13}, state).state == (
DownloadState.DOWNLOADING
)
def test_a_vanished_torrent_becomes_an_error_rather_than_a_stall(self, monkeypatch, tmp_path):
monkeypatch.setattr("shelfmark.download.clients.debridlink.TMP_DIR", tmp_path)
_mock_request(monkeypatch, {"success": True, "value": []})
client = _client(monkeypatch)
status = client.get_status("GONE")
assert status.state == DownloadState.ERROR
assert "no longer on Debrid-Link" in (status.message or "")
class TestPolling:
"""The handler polls every couple of seconds; Debrid-Link must not see that."""
def test_status_is_served_from_cache_between_refreshes(self, monkeypatch, tmp_path):
monkeypatch.setattr("shelfmark.download.clients.debridlink.TMP_DIR", tmp_path)
request = _mock_request(
monkeypatch, {"success": True, "value": [{"id": "DL1", "downloadPercent": 30}]}
)
client = _client(monkeypatch)
for _ in range(5):
status = client.get_status("DL1")
assert request.call_count == 1
assert status.progress == 15.0
def test_a_rate_limit_pauses_checks_instead_of_failing(self, monkeypatch, tmp_path):
monkeypatch.setattr("shelfmark.download.clients.debridlink.TMP_DIR", tmp_path)
request = _mock_request(
monkeypatch, {"success": False, "error": "floodDetected"}, status_code=429
)
client = _client(monkeypatch)
first = client.get_status("DL1")
second = client.get_status("DL1")
assert first.state == DownloadState.DOWNLOADING
assert "rate limit" in (first.message or "")
assert second.state == DownloadState.DOWNLOADING
assert request.call_count == 1
def test_a_temporary_error_is_ridden_out_then_fails(self, monkeypatch, tmp_path):
monkeypatch.setattr("shelfmark.download.clients.debridlink.TMP_DIR", tmp_path)
_mock_request(monkeypatch, {"success": False, "error": "internalError"}, status_code=500)
clock = [1000.0]
monkeypatch.setattr(debridlink.time, "monotonic", lambda: clock[0])
client = _client(monkeypatch)
assert client.get_status("DL1").state == DownloadState.DOWNLOADING
clock[0] += debridlink._TRANSIENT_GRACE_SECONDS + debridlink._MIN_REFRESH_SECONDS
status = client.get_status("DL1")
assert status.state == DownloadState.ERROR
assert "internalError" in (status.message or "")
def test_a_permanent_error_fails_at_once(self, monkeypatch, tmp_path):
monkeypatch.setattr("shelfmark.download.clients.debridlink.TMP_DIR", tmp_path)
_mock_request(monkeypatch, {"success": False, "error": "badToken"}, status_code=401)
client = _client(monkeypatch)
status = client.get_status("DL1")
assert status.state == DownloadState.ERROR
assert "badToken" in (status.message or "")
class TestFetchTorrent:
def test_the_matching_id_is_picked_out_of_the_page(self, monkeypatch):
_mock_request(
monkeypatch,
{
"success": True,
"value": [{"id": "OTHER"}, {"id": "DL1", "downloadPercent": 12}],
},
)
client = _client(monkeypatch)
assert client._fetch_torrent("DL1") == {"id": "DL1", "downloadPercent": 12}
def test_an_absent_id_returns_none(self, monkeypatch):
_mock_request(monkeypatch, {"success": True, "value": [{"id": "OTHER"}]})
client = _client(monkeypatch)
assert client._fetch_torrent("DL1") is None
def test_a_bad_id_error_means_the_torrent_is_gone(self, monkeypatch):
_mock_request(monkeypatch, {"success": False, "error": "badId"}, status_code=400)
client = _client(monkeypatch)
assert client._fetch_torrent("DL1") is None
def test_the_status_check_asks_for_just_that_torrent(self, monkeypatch):
request = _mock_request(monkeypatch, {"success": True, "value": []})
client = _client(monkeypatch)
client._fetch_torrent("DL1")
assert request.call_args.kwargs["params"] == {"ids": "DL1"}
def test_the_file_list_uses_the_singular_id_so_zips_are_expanded(self, monkeypatch):
# Debrid-Link lists a many-file torrent as one ZIP under ``ids``; ``?id=``
# returns the individual files.
request = _mock_request(
monkeypatch,
{
"success": True,
"value": [{"id": "DL1", "files": [{"name": "a.mp3"}, {"name": "b.mp3"}]}],
},
)
client = _client(monkeypatch)
files = client._fetch_file_list("DL1")
assert request.call_args.kwargs["params"] == {"id": "DL1"}
assert [f["name"] for f in files] == ["a.mp3", "b.mp3"]
class TestConnection:
def test_missing_key_is_reported_without_a_call(self, monkeypatch):
request = _mock_request(monkeypatch, {"success": True, "value": {}})
client = _client(monkeypatch, api_key="")
assert client.test_connection() == (False, "Debrid-Link API Key is required")
request.assert_not_called()
def test_an_expired_account_is_rejected(self, monkeypatch):
_mock_request(
monkeypatch,
{"success": True, "value": {"username": "rob", "premiumLeft": 0}},
)
client = _client(monkeypatch)
ok, message = client.test_connection()
assert ok is False
assert "does not have an active" in message
def test_a_premium_account_is_accepted(self, monkeypatch):
_mock_request(
monkeypatch,
{"success": True, "value": {"username": "rob", "premiumLeft": 86400}},
)
client = _client(monkeypatch)
ok, message = client.test_connection()
assert ok is True
assert "rob" in message
def test_an_api_failure_is_surfaced_not_raised(self, monkeypatch):
_mock_request(monkeypatch, {"success": False, "error": "badToken"})
client = _client(monkeypatch)
ok, message = client.test_connection()
assert ok is False
assert "badToken" in message
class TestIsConfigured:
@staticmethod
def _config(monkeypatch, values):
monkeypatch.setattr(
"shelfmark.download.clients.debridlink.config.get",
lambda key, default="": values.get(key, default),
)
def test_requires_both_the_selection_and_the_key(self, monkeypatch):
self._config(
monkeypatch,
{"PROWLARR_TORRENT_CLIENT": "debridlink", "DEBRIDLINK_API_KEY": "dl-key"},
)
assert DebridLinkClient.is_configured() is True
def test_another_client_selected_means_not_configured(self, monkeypatch):
self._config(
monkeypatch,
{"PROWLARR_TORRENT_CLIENT": "torbox", "DEBRIDLINK_API_KEY": "dl-key"},
)
assert DebridLinkClient.is_configured() is False
def test_no_key_means_not_configured(self, monkeypatch):
self._config(
monkeypatch,
{"PROWLARR_TORRENT_CLIENT": "debridlink", "DEBRIDLINK_API_KEY": ""},
)
assert DebridLinkClient.is_configured() is False
class TestDownloadPipeline:
@staticmethod
def _run(tmp_path, monkeypatch, files, payload=b"data"):
client = DebridLinkClient.__new__(DebridLinkClient)
state = _state(tmp_path, phase="downloading_http")
fetched: list[str] = []
def fake_download(url, referer=None, cancel_flag=None):
fetched.append(url)
return _buffer(payload)
monkeypatch.setattr("shelfmark.download.clients.debridlink.download_url", fake_download)
client._process_and_download(state, files)
return state, fetched
def test_book_files_are_preferred_over_the_rest(self, tmp_path, monkeypatch):
state, fetched = self._run(
tmp_path,
monkeypatch,
[
{"name": "cover.jpg", "downloadUrl": "https://dl.example/cover.jpg"},
{"name": "Dune.epub", "downloadUrl": "https://dl.example/Dune.epub"},
],
)
assert fetched == ["https://dl.example/Dune.epub"]
assert (tmp_path / "Dune.epub").read_bytes() == b"data"
assert state.phase == "complete"
assert state.progress == 100.0
def test_everything_is_taken_when_nothing_looks_like_a_book(self, tmp_path, monkeypatch):
state, _ = self._run(
tmp_path, monkeypatch, [{"name": "readme.nfo", "downloadUrl": "https://dl.example/x"}]
)
assert (tmp_path / "readme.nfo").exists()
assert state.phase == "complete"
def test_a_zip_bundle_is_downloaded_as_a_book_file(self, tmp_path, monkeypatch):
state, fetched = self._run(
tmp_path,
monkeypatch,
[
{"name": "notes.nfo", "downloadUrl": "https://dl.example/n"},
{"name": "Dune.zip", "downloadUrl": "https://dl.example/z"},
],
)
assert fetched == ["https://dl.example/z"]
assert state.phase == "complete"
def test_no_links_is_an_error(self, tmp_path):
client = DebridLinkClient.__new__(DebridLinkClient)
state = _state(tmp_path, phase="downloading_http")
client._process_and_download(state, [{"name": "Dune.epub"}])
assert state.phase == "error"
assert "No download links" in (state.error_message or "")
@pytest.mark.parametrize("name", ["../escape.epub", "/etc/escape.epub", "a/../../x.epub"])
def test_a_file_name_cannot_escape_the_download_folder(self, tmp_path, monkeypatch, name):
target = tmp_path / "dl"
target.mkdir()
client = DebridLinkClient.__new__(DebridLinkClient)
state = _state(target, phase="downloading_http")
monkeypatch.setattr(
"shelfmark.download.clients.debridlink.download_url",
lambda url, referer=None, cancel_flag=None: _buffer(),
)
client._process_and_download(state, [{"name": name, "downloadUrl": "https://dl.example/x"}])
assert state.phase == "error"
assert "unsafe file path" in (state.error_message or "")
assert not (tmp_path / "escape.epub").exists()
def test_two_files_with_the_same_name_are_both_kept(self, tmp_path, monkeypatch):
payloads = iter([b"one", b"two"])
client = DebridLinkClient.__new__(DebridLinkClient)
state = _state(tmp_path, phase="downloading_http")
monkeypatch.setattr(
"shelfmark.download.clients.debridlink.download_url",
lambda url, referer=None, cancel_flag=None: _buffer(next(payloads)),
)
client._process_and_download(
state,
[
{"name": "chapter.mp3", "downloadUrl": "https://dl.example/1"},
{"name": "chapter.mp3", "downloadUrl": "https://dl.example/2"},
],
)
assert (tmp_path / "chapter.mp3").read_bytes() == b"one"
assert (tmp_path / "chapter (2).mp3").read_bytes() == b"two"
def test_a_failed_file_is_reported_without_its_signed_url(self, tmp_path, monkeypatch):
client = DebridLinkClient.__new__(DebridLinkClient)
state = _state(tmp_path, phase="downloading_http")
monkeypatch.setattr(
"shelfmark.download.clients.debridlink.download_url",
lambda url, referer=None, cancel_flag=None: None,
)
client._process_and_download(
state, [{"name": "Dune.epub", "downloadUrl": "https://seed1.debrid.link/dl/SIGNED"}]
)
assert state.phase == "error"
assert "Dune.epub" in (state.error_message or "")
assert "SIGNED" not in (state.error_message or "")
def test_the_file_list_is_fetched_fresh_when_none_is_given(self, tmp_path, monkeypatch):
client = DebridLinkClient.__new__(DebridLinkClient)
state = _state(tmp_path, phase="downloading_http")
listing = MagicMock(return_value=[{"name": "Dune.epub", "downloadUrl": "https://x/y"}])
monkeypatch.setattr(client, "_fetch_file_list", listing)
monkeypatch.setattr(
"shelfmark.download.clients.debridlink.download_url",
lambda url, referer=None, cancel_flag=None: _buffer(),
)
client._process_and_download(state)
listing.assert_called_once_with("DL1")
assert state.phase == "complete"
def test_cancelling_stops_before_the_next_file(self, tmp_path, monkeypatch):
client = DebridLinkClient.__new__(DebridLinkClient)
state = _state(tmp_path, phase="downloading_http")
fetched: list[str] = []
def fake_download(url, referer=None, cancel_flag=None):
fetched.append(url)
state.cancel_event.set()
return _buffer()
monkeypatch.setattr("shelfmark.download.clients.debridlink.download_url", fake_download)
client._process_and_download(
state,
[
{"name": "a.mp3", "downloadUrl": "https://dl.example/a"},
{"name": "b.mp3", "downloadUrl": "https://dl.example/b"},
],
)
assert fetched == ["https://dl.example/a"]
assert state.phase != "complete"
class TestRemove:
def test_remove_cancels_the_retrieval_and_cleans_up(self, monkeypatch, tmp_path):
monkeypatch.setattr("shelfmark.download.clients.debridlink.TMP_DIR", tmp_path)
_mock_request(monkeypatch, {"success": True, "value": None})
client = _client(monkeypatch)
state = _state(tmp_path / "debridlink_DL1", phase="downloading_http")
state.target_dir.mkdir()
DebridLinkClient._downloads["DL1"] = state
assert client.remove("DL1") is True
assert state.cancel_event.is_set()
assert not state.target_dir.exists()
assert "DL1" not in DebridLinkClient._downloads
class TestAsFloat:
@pytest.mark.parametrize(
("value", "expected"),
[
(42, 42.0),
(4.5, 4.5),
("17", 17.0),
("not a number", 0.0),
(None, 0.0),
(True, 0.0),
],
)
def test_api_numbers_are_coerced_without_raising(self, value, expected):
assert _as_float(value) == expected
tests/prowlarr/test_debridlink_settings.py
+49
View File
@@ -0,0 +1,49 @@
"""Debrid-Link Download Client settings and connection-action tests."""
from unittest.mock import MagicMock
from shelfmark.core.settings_registry import ActionButton, PasswordField
def _field(fields, key):
return next(field for field in fields if field.key == key)
def test_debridlink_fields_are_registered_with_conditional_visibility():
from shelfmark.download.clients.settings import prowlarr_clients_settings
fields = prowlarr_clients_settings()
client_field = _field(fields, "PROWLARR_TORRENT_CLIENT")
api_key_field = _field(fields, "DEBRIDLINK_API_KEY")
test_action = _field(fields, "test_debridlink")
assert {option["value"] for option in client_field.options} >= {"debridlink"}
assert isinstance(api_key_field, PasswordField)
assert api_key_field.show_when == {"field": "PROWLARR_TORRENT_CLIENT", "value": "debridlink"}
assert isinstance(test_action, ActionButton)
assert test_action.show_when == {"field": "PROWLARR_TORRENT_CLIENT", "value": "debridlink"}
def test_debridlink_connection_action_uses_unsaved_api_key(monkeypatch):
from shelfmark.core.config import config
from shelfmark.download.clients import settings as settings_module
client = MagicMock()
client.test_connection.return_value = (True, "Connected")
client_type = MagicMock(return_value=client)
monkeypatch.setattr("shelfmark.download.clients.debridlink.DebridLinkClient", client_type)
monkeypatch.setattr(config, "get", lambda _key, default="": "saved-key")
result = settings_module._test_debridlink_connection({"DEBRIDLINK_API_KEY": "unsaved-key"})
assert result == {"success": True, "message": "Connected"}
assert client._api_key == "unsaved-key"
def test_debridlink_connection_action_requires_api_key():
from shelfmark.download.clients.settings import _test_debridlink_connection
assert _test_debridlink_connection({"DEBRIDLINK_API_KEY": ""}) == {
"success": False,
"message": "Debrid-Link API Key is required",
}