diff --git a/CHANGELOG.md b/CHANGELOG.md index 1924c6c..583b016 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,27 @@ All notable changes to this project will be documented in this file. The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). +## [0.28.0] - 2026-08-29 + +### Added +- **Freeleech (FL) token spending on RED/OPS** — opt-in via `RED_USE_FL_TOKEN` / + `OPS_USE_FL_TOKEN` env vars (or `--red-use-token` / `--ops-use-token` CLI flags), + off by default. When enabled, flacfetch spends a Freeleech token on eligible + downloads (`ajax.php?action=download&id=&usetoken=1`) so the download + doesn't count against ratio. + - **Driven by the tracker's `canUseToken` flag** (captured from the `browse` + search response, which the code previously discarded). It's true only when + the account currently holds a spendable token for that torrent (≤ 5 GB, not + already free), so tokens are used automatically whenever available and it + stops once they run out. There is no RED API endpoint that returns a raw + token *count* — `canUseToken` is the authoritative signal. + - Already-free torrents are never charged a token; cached `.torrent` files are + reused without spending one. + - **Graceful fallback:** a failed token spend transparently retries as a normal + (ratio-counted) download, so a token failure never blocks a download. + - Hardened `.torrent` fetch to validate the response is bencoded before caching, + so a JSON error body (e.g. "no tokens left") is never cached as a bogus torrent. + ## [0.27.0] - 2026-08-26 ### Added diff --git a/README.md b/README.md index 1cf122c..9a92206 100644 --- a/README.md +++ b/README.md @@ -125,6 +125,31 @@ export OPS_API_URL="your_tracker_url_here" flacfetch "..." --ops-key "your_key" --ops-url "your_url" ``` +**Freeleech Tokens** (Optional) + +Gazelle trackers (RED/OPS) let you spend a **Freeleech (FL) token** to download a +torrent without it counting against your ratio. flacfetch can spend tokens +automatically on eligible downloads: + +```bash +export RED_USE_FL_TOKEN=true # or: flacfetch "..." --red-use-token +export OPS_USE_FL_TOKEN=true # or: flacfetch "..." --ops-use-token +``` + +Behaviour (opt-in, **off by default**): +- Spends a token only when the tracker reports one is spendable for that torrent + (its `canUseToken` flag — which is only true when you currently hold a token, + the torrent is ≤ 5 GB, and it isn't already freeleech). This means tokens are + used automatically whenever available and it simply stops once you run out. +- Already-free torrents are never charged a token, and cached `.torrent` files are + reused without spending one. +- If a token spend fails for any reason, flacfetch transparently falls back to a + normal (ratio-counted) download — a failed token never blocks a download. + +> Note: there is no RED API endpoint that returns your raw token *count*; the +> per-torrent `canUseToken` flag from search results is the authoritative +> "is a token spendable here" signal, and is what drives this behaviour. + **Spotify Configuration** (Optional - requires Premium account) Spotify provides CD-quality audio (44.1kHz/16-bit) captured via librespot and converted to FLAC. This uses the official Spotify Web API for authentication (OAuth) and librespot for audio capture. diff --git a/flacfetch/api/routes/health.py b/flacfetch/api/routes/health.py index be92f8c..9da9225 100644 --- a/flacfetch/api/routes/health.py +++ b/flacfetch/api/routes/health.py @@ -238,8 +238,10 @@ async def debug_providers(): "env_vars": { "RED_API_KEY": bool(os.environ.get("RED_API_KEY")), "RED_API_URL": bool(os.environ.get("RED_API_URL")), + "RED_USE_FL_TOKEN": os.environ.get("RED_USE_FL_TOKEN", "false").lower() in ("true", "1", "yes"), "OPS_API_KEY": bool(os.environ.get("OPS_API_KEY")), "OPS_API_URL": bool(os.environ.get("OPS_API_URL")), + "OPS_USE_FL_TOKEN": os.environ.get("OPS_USE_FL_TOKEN", "false").lower() in ("true", "1", "yes"), "SPOTIPY_CLIENT_ID": bool(os.environ.get("SPOTIPY_CLIENT_ID")), "SPOTIPY_CLIENT_SECRET": bool(os.environ.get("SPOTIPY_CLIENT_SECRET")), }, diff --git a/flacfetch/api/services/download_manager.py b/flacfetch/api/services/download_manager.py index 4f853c4..cbf7892 100644 --- a/flacfetch/api/services/download_manager.py +++ b/flacfetch/api/services/download_manager.py @@ -224,12 +224,15 @@ def _get_fetch_manager(self): # Add RED provider if configured (requires both key and URL) red_key = os.environ.get("RED_API_KEY") red_url = os.environ.get("RED_API_URL") + red_use_token = os.environ.get("RED_USE_FL_TOKEN", "false").lower() in ("true", "1", "yes") if red_key and red_url: try: from flacfetch.downloaders.torrent import TorrentDownloader from flacfetch.providers.red import REDProvider - self._fetch_manager.add_provider(REDProvider(api_key=red_key, base_url=red_url)) + self._fetch_manager.add_provider(REDProvider(api_key=red_key, base_url=red_url, use_fl_token=red_use_token)) + if red_use_token: + logger.info("RED Freeleech token spending ENABLED (RED_USE_FL_TOKEN)") self._fetch_manager.register_downloader( "RED", TorrentDownloader(keep_seeding=self.keep_seeding) @@ -241,12 +244,15 @@ def _get_fetch_manager(self): # Add OPS provider if configured (requires both key and URL) ops_key = os.environ.get("OPS_API_KEY") ops_url = os.environ.get("OPS_API_URL") + ops_use_token = os.environ.get("OPS_USE_FL_TOKEN", "false").lower() in ("true", "1", "yes") if ops_key and ops_url: try: from flacfetch.downloaders.torrent import TorrentDownloader from flacfetch.providers.ops import OPSProvider - self._fetch_manager.add_provider(OPSProvider(api_key=ops_key, base_url=ops_url)) + self._fetch_manager.add_provider(OPSProvider(api_key=ops_key, base_url=ops_url, use_fl_token=ops_use_token)) + if ops_use_token: + logger.info("OPS Freeleech token spending ENABLED (OPS_USE_FL_TOKEN)") self._fetch_manager.register_downloader( "OPS", TorrentDownloader(keep_seeding=self.keep_seeding) diff --git a/flacfetch/core/models.py b/flacfetch/core/models.py index 404950d..e8d158a 100644 --- a/flacfetch/core/models.py +++ b/flacfetch/core/models.py @@ -116,6 +116,10 @@ class Release: release_type: Optional[str] = None # e.g. "Album", "Single" seeders: Optional[int] = None + # Private-tracker freeleech info (RED/OPS) + is_freeleech: bool = False # Torrent is already free (no ratio hit) -- don't spend a token on it + can_use_token: Optional[bool] = None # Server says a Freeleech token can be spent (has tokens, <=5GB, not already free) + # YouTube / Streaming Metadata channel: Optional[str] = None view_count: Optional[int] = None @@ -153,6 +157,8 @@ def to_dict(self) -> dict: "catalogue_number": self.catalogue_number, "release_type": self.release_type, "seeders": self.seeders, + "is_freeleech": self.is_freeleech, + "can_use_token": self.can_use_token, "channel": self.channel, "view_count": self.view_count, "duration_seconds": self.duration_seconds, @@ -214,6 +220,8 @@ def from_dict(cls, data: dict) -> "Release": catalogue_number=data.get("catalogue_number"), release_type=data.get("release_type"), seeders=data.get("seeders"), + is_freeleech=data.get("is_freeleech", False), + can_use_token=data.get("can_use_token"), channel=data.get("channel"), view_count=data.get("view_count"), duration_seconds=data.get("duration_seconds"), diff --git a/flacfetch/interface/cli.py b/flacfetch/interface/cli.py index b949f22..994b574 100644 --- a/flacfetch/interface/cli.py +++ b/flacfetch/interface/cli.py @@ -1320,8 +1320,10 @@ def __init__(self, prog, max_help_position=35, width=100): Environment Variables: RED_API_KEY API key for RED (lossless FLAC source) RED_API_URL Base URL for RED API (required if using RED) + RED_USE_FL_TOKEN Spend a RED Freeleech token on eligible downloads (true/false) OPS_API_KEY API key for OPS (lossless FLAC source) OPS_API_URL Base URL for OPS API (required if using OPS) + OPS_USE_FL_TOKEN Spend an OPS Freeleech token on eligible downloads (true/false) SPOTIPY_CLIENT_ID Spotify app client ID SPOTIPY_CLIENT_SECRET Spotify app client secret SPOTIPY_REDIRECT_URI OAuth redirect URI (http://127.0.0.1:8888/callback) @@ -1418,6 +1420,16 @@ def __init__(self, prog, max_help_position=35, width=100): metavar="URL", help="OPS API base URL (or use OPS_API_URL env var)" ) + provider_group.add_argument( + "--red-use-token", + action="store_true", + help="Spend a RED Freeleech token on eligible downloads (or use RED_USE_FL_TOKEN env var)" + ) + provider_group.add_argument( + "--ops-use-token", + action="store_true", + help="Spend an OPS Freeleech token on eligible downloads (or use OPS_USE_FL_TOKEN env var)" + ) provider_group.add_argument( "--no-spotify", action="store_true", @@ -1488,9 +1500,10 @@ def __init__(self, prog, max_help_position=35, width=100): # Register RED provider red_key = args.red_key or os.environ.get("RED_API_KEY") red_url = args.red_url or os.environ.get("RED_API_URL") + red_use_token = args.red_use_token or os.environ.get("RED_USE_FL_TOKEN", "false").lower() in ("true", "1", "yes") if red_key and red_url: if artist: - rp = REDProvider(api_key=red_key, base_url=red_url) + rp = REDProvider(api_key=red_key, base_url=red_url, use_fl_token=red_use_token) rp.search_limit = search_limit rp.early_termination = use_early_termination manager.add_provider(rp) @@ -1509,9 +1522,10 @@ def __init__(self, prog, max_help_position=35, width=100): # Register OPS provider ops_key = args.ops_key or os.environ.get("OPS_API_KEY") ops_url = args.ops_url or os.environ.get("OPS_API_URL") + ops_use_token = args.ops_use_token or os.environ.get("OPS_USE_FL_TOKEN", "false").lower() in ("true", "1", "yes") if ops_key and ops_url: if artist: - ops = OPSProvider(api_key=ops_key, base_url=ops_url) + ops = OPSProvider(api_key=ops_key, base_url=ops_url, use_fl_token=ops_use_token) ops.search_limit = search_limit ops.early_termination = use_early_termination manager.add_provider(ops) diff --git a/flacfetch/providers/gazelle.py b/flacfetch/providers/gazelle.py index b2a749b..df3c567 100644 --- a/flacfetch/providers/gazelle.py +++ b/flacfetch/providers/gazelle.py @@ -36,19 +36,27 @@ class GazelleProvider(Provider): - Torrent artifact fetching with caching """ - def __init__(self, api_key: str, base_url: str, cache_subdir: str): + # Freeleech (FL) tokens can only be spent on torrents up to 5 GB. + # See the RED wiki: a token makes the whole torrent personal-freeleech. + FL_TOKEN_MAX_BYTES = 5 * 1024 ** 3 + + def __init__(self, api_key: str, base_url: str, cache_subdir: str, use_fl_token: bool = False): """Initialize the Gazelle provider. Args: api_key: API key for authentication base_url: Base URL of the tracker API cache_subdir: Subdirectory name for cache (e.g., "red" or "ops") + use_fl_token: When True, spend a Freeleech token on eligible downloads + (torrent <= 5 GB and not already freeleech). Falls back to a normal + download if the token cannot be spent. Default False (opt-in). """ if not base_url: raise ValueError("base_url is required. Set the appropriate environment variable.") self.api_key = api_key self.base_url = base_url.rstrip('/') + self.use_fl_token = use_fl_token self.session = requests.Session() self.session.headers.update({"Authorization": self.api_key}) self.search_limit = 10 @@ -264,6 +272,21 @@ def _fetch_torrent_from_url(self, url: str, torrent_id: Optional[str] = None, ma resp = self.session.get(redirect_url, timeout=10) if resp.status_code == 200: + # The download endpoint returns raw bencoded .torrent bytes on + # success (content-type application/x-bittorrent) but a JSON + # error body on failure (e.g. a token that can't be spent, or + # "already downloaded"). A JSON body is still HTTP 200, so guard + # against caching/returning it as a bogus torrent. + content_type = resp.headers.get("Content-Type", "").lower() + is_torrent = resp.content[:1] == b"d" and "json" not in content_type + if not is_torrent: + try: + err = resp.json() + logger.warning(f"{self.name} download returned an error instead of a torrent: {err.get('error', err)}") + except ValueError: + logger.warning(f"{self.name} download returned non-torrent content ({len(resp.content)} bytes): {resp.content[:100]}") + return None + logger.debug(f"Artifact fetched successfully ({len(resp.content)} bytes)") if len(resp.content) < 1000: logger.warning(f"Artifact seems too small ({len(resp.content)} bytes). Content sample: {resp.content[:100]}") @@ -297,11 +320,72 @@ def _fetch_torrent_from_url(self, url: str, torrent_id: Optional[str] = None, ma return None - def fetch_artifact_by_id(self, source_id: str) -> Optional[bytes]: + def _fetch_with_optional_token(self, base_url: str, torrent_id: Optional[str], use_token: bool) -> Optional[bytes]: + """Fetch a torrent, optionally spending a Freeleech token. + + When ``use_token`` is True, first attempt the download with ``usetoken=1``. + If that fails for any reason (torrent too large, no tokens left, server + error), transparently fall back to a normal (ratio-counted) download so a + failed token spend never blocks the download. + + Args: + base_url: Download URL without the ``usetoken`` parameter + torrent_id: Torrent ID for caching/logging (may be None) + use_token: Whether to attempt spending a Freeleech token + + Returns: + Torrent file contents as bytes, or None on failure + """ + if use_token: + sep = "&" if "?" in base_url else "?" + token_url = f"{base_url}{sep}usetoken=1" + logger.info(f"{self.name}: attempting Freeleech token download for torrent {torrent_id}") + content = self._fetch_torrent_from_url(token_url, torrent_id) + if content: + logger.info(f"{self.name}: spent a Freeleech token on torrent {torrent_id}") + return content + logger.warning(f"{self.name}: Freeleech token download failed for torrent {torrent_id}; retrying without token") + + return self._fetch_torrent_from_url(base_url, torrent_id) + + def _is_token_eligible(self, release: Release) -> bool: + """Decide whether it's worth spending a Freeleech token on a release. + + Prefers the server's ``canUseToken`` flag (captured from the browse + response), which is true only when the account currently holds a + spendable token for that torrent -- this is what lets us automatically + use tokens whenever any are available and stop once they run out. + + When ``canUseToken`` is unknown (e.g. the direct torrent-ID path), falls + back to a local heuristic: skip already-free torrents and torrents over + the 5 GB token limit. The attempt-and-fallback in _fetch_with_optional_token + remains the final safety net. + """ + if release.can_use_token is not None: + if not release.can_use_token: + logger.debug(f"{self.name}: server reports no spendable token for this torrent") + return release.can_use_token + + if getattr(release, "is_freeleech", False): + logger.debug(f"{self.name}: torrent already freeleech; not spending a token") + return False + if release.size_bytes and release.size_bytes > self.FL_TOKEN_MAX_BYTES: + logger.info( + f"{self.name}: torrent size {release.size_bytes} bytes exceeds the " + f"{self.FL_TOKEN_MAX_BYTES}-byte FL token limit; not spending a token" + ) + return False + return True + + def fetch_artifact_by_id(self, source_id: str, use_token: Optional[bool] = None) -> Optional[bytes]: """Fetch .torrent file by torrent ID directly. Args: source_id: The torrent ID + use_token: Whether to spend a Freeleech token. Defaults to the + provider-level ``use_fl_token`` setting when None. No torrent size + is known via this path, so eligibility relies on the graceful + fallback in ``_fetch_with_optional_token``. Returns: Torrent file contents as bytes, or None on failure @@ -309,7 +393,11 @@ def fetch_artifact_by_id(self, source_id: str) -> Optional[bytes]: if not source_id: return None - # Check Cache first + if use_token is None: + use_token = self.use_fl_token + + # Check Cache first (a cached torrent means we already have it -- never + # spend a token to re-fetch something on disk). if self.cache_dir: cache_path = self.cache_dir / f"{source_id}.torrent" if cache_path.exists(): @@ -324,7 +412,7 @@ def fetch_artifact_by_id(self, source_id: str) -> Optional[bytes]: # Construct download URL from torrent ID url = f"{self.base_url}/ajax.php?action=download&id={source_id}" - return self._fetch_torrent_from_url(url, source_id) + return self._fetch_with_optional_token(url, source_id, use_token) def fetch_artifact(self, release: Release) -> Optional[bytes]: """Fetch .torrent file for a release. @@ -338,6 +426,11 @@ def fetch_artifact(self, release: Release) -> Optional[bytes]: if not release.download_url: return None + # Decide whether to spend a Freeleech token on this specific release. + # Unlike fetch_artifact_by_id, we have the torrent size and freeleech + # status here, so we can avoid wasting tokens on ineligible torrents. + use_token = self.use_fl_token and self._is_token_eligible(release) + # Extract Torrent ID for caching torrent_id = None try: @@ -348,10 +441,10 @@ def fetch_artifact(self, release: Release) -> Optional[bytes]: # Check Cache first (reuse fetch_artifact_by_id if we have a torrent_id) if torrent_id: - return self.fetch_artifact_by_id(torrent_id) + return self.fetch_artifact_by_id(torrent_id, use_token=use_token) # Fallback: fetch directly from URL without caching - return self._fetch_torrent_from_url(release.download_url) + return self._fetch_with_optional_token(release.download_url, None, use_token) @abstractmethod def search(self, query: TrackQuery) -> list[Release]: @@ -365,16 +458,25 @@ def search(self, query: TrackQuery) -> list[Release]: """ pass - def _fetch_group_details(self, group_id: int, track_title: str) -> list[Release]: + def _fetch_group_details( + self, + group_id: int, + track_title: str, + token_eligibility: Optional[dict[int, bool]] = None, + ) -> list[Release]: """Fetch detailed torrent information for a group. Args: group_id: The group ID to fetch details for track_title: The track title to match against files + token_eligibility: Optional map of {torrentId: canUseToken} captured + from the browse response. The torrentgroup endpoint does not + return canUseToken, so we carry it over from the search results. Returns: List of releases matching the track title """ + token_eligibility = token_eligibility or {} url = f"{self.base_url}/ajax.php" params = {"action": "torrentgroup", "id": group_id} @@ -440,6 +542,20 @@ def _fetch_group_details(self, group_id: int, track_title: str) -> list[Release] torrent_id = str(torrent.get("id", "")) + # Already-free torrents don't need (and shouldn't waste) a token. + is_freeleech = bool( + torrent.get("isFreeleech") + or torrent.get("isPersonalFreeleech") + or torrent.get("isNeutralLeech") + or torrent.get("isFreeload") + or torrent.get("freeTorrent") + ) + + # canUseToken comes from the browse response (torrentgroup omits + # it). It's the authoritative "a token can be spent here" signal: + # true only when the account currently has a spendable token. + can_use_token = token_eligibility.get(torrent.get("id")) + r = Release( title=group_name, artist=artist, @@ -453,6 +569,8 @@ def _fetch_group_details(self, group_id: int, track_title: str) -> list[Release] catalogue_number=cat_num, release_type=release_type_str, seeders=torrent.get("seeders", 0), + is_freeleech=is_freeleech, + can_use_token=can_use_token, target_file=target_file, target_file_size=target_size, match_score=match_score, @@ -511,14 +629,20 @@ def _search_browse(self, query: TrackQuery) -> list[Release]: browse_results = data.get("response", {}).get("results", []) logger.debug(f"Found {len(browse_results)} groups in {self.name} response") - # Extract ordered group IDs + # Extract ordered group IDs, and capture per-torrent canUseToken from + # the browse response (the torrentgroup endpoint doesn't return it). ordered_group_ids = [] seen = set() + token_eligibility: dict[int, bool] = {} for g in browse_results: gid = g.get("groupId") if gid and gid not in seen: ordered_group_ids.append(gid) seen.add(gid) + for t in g.get("torrents", []): + tid = t.get("torrentId") + if tid is not None and "canUseToken" in t: + token_eligibility[tid] = bool(t.get("canUseToken")) limited_group_ids = ordered_group_ids[:self.search_limit] @@ -535,7 +659,7 @@ def _search_browse(self, query: TrackQuery) -> list[Release]: # Rate limit: sleep before each request time.sleep(1.1) - group_releases = self._fetch_group_details(gid, query.title) + group_releases = self._fetch_group_details(gid, query.title, token_eligibility) releases.extend(group_releases) groups_fetched += 1 diff --git a/flacfetch/providers/ops.py b/flacfetch/providers/ops.py index 3b73f7e..2d99ae2 100644 --- a/flacfetch/providers/ops.py +++ b/flacfetch/providers/ops.py @@ -12,16 +12,18 @@ class OPSProvider(GazelleProvider): for security reasons (to avoid hardcoding tracker URLs in source code). """ - def __init__(self, api_key: str, base_url: str): + def __init__(self, api_key: str, base_url: str, use_fl_token: bool = False): """Initialize the OPS provider. Args: api_key: API key for authentication base_url: Base URL of the tracker API (e.g., from OPS_API_URL env var) + use_fl_token: When True, spend a Freeleech token on eligible downloads + (from OPS_USE_FL_TOKEN env var). Default False. """ if not base_url: raise ValueError("base_url is required for OPSProvider. Set OPS_API_URL environment variable.") - super().__init__(api_key, base_url, cache_subdir="ops") + super().__init__(api_key, base_url, cache_subdir="ops", use_fl_token=use_fl_token) @property def name(self) -> str: diff --git a/flacfetch/providers/red.py b/flacfetch/providers/red.py index 6fd3c4b..fd7e0d9 100644 --- a/flacfetch/providers/red.py +++ b/flacfetch/providers/red.py @@ -12,16 +12,18 @@ class REDProvider(GazelleProvider): for security reasons (to avoid hardcoding tracker URLs in source code). """ - def __init__(self, api_key: str, base_url: str): + def __init__(self, api_key: str, base_url: str, use_fl_token: bool = False): """Initialize the RED provider. Args: api_key: API key for authentication base_url: Base URL of the tracker API (e.g., from RED_API_URL env var) + use_fl_token: When True, spend a Freeleech token on eligible downloads + (from RED_USE_FL_TOKEN env var). Default False. """ if not base_url: raise ValueError("base_url is required for REDProvider. Set RED_API_URL environment variable.") - super().__init__(api_key, base_url, cache_subdir="red") + super().__init__(api_key, base_url, cache_subdir="red", use_fl_token=use_fl_token) @property def name(self) -> str: diff --git a/pyproject.toml b/pyproject.toml index 8ba2156..0a6ee8e 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "flacfetch" -version = "0.27.0" +version = "0.28.0" description = "Search and download high-quality audio from multiple sources" readme = "README.md" requires-python = ">=3.10,<4.0" diff --git a/tests/test_gazelle.py b/tests/test_gazelle.py index c5b4d38..cde389e 100644 --- a/tests/test_gazelle.py +++ b/tests/test_gazelle.py @@ -1,5 +1,7 @@ """Tests for GazelleProvider base class - Sphinx query sanitization and shared functionality.""" +from unittest.mock import MagicMock + import pytest from flacfetch.core.models import AudioFormat, MediaSource @@ -472,3 +474,150 @@ def test_empty_torrent_data(self, provider): assert quality.media == MediaSource.OTHER assert quality.bit_depth is None assert quality.bitrate is None + + +class TestFreeleechTokens: + """Test Freeleech (FL) token spending behavior.""" + + @pytest.fixture + def provider(self): + """Token-enabled concrete GazelleProvider.""" + return ConcreteGazelleProvider( + api_key="test", + base_url="https://test.example", + cache_subdir="test", + use_fl_token=True, + ) + + @pytest.fixture + def no_token_provider(self): + """Default provider (token spending off).""" + return ConcreteGazelleProvider( + api_key="test", + base_url="https://test.example", + cache_subdir="test", + ) + + def _release(self, **kwargs): + from flacfetch.core.models import AudioFormat, Quality, Release + defaults = dict( + title="Album", + artist="Artist", + quality=Quality(format=AudioFormat.FLAC), + source_name="TEST", + download_url="https://test.example/ajax.php?action=download&id=42", + source_id="42", + ) + defaults.update(kwargs) + return Release(**defaults) + + # ---- default off ------------------------------------------------------- + + def test_use_fl_token_defaults_off(self, no_token_provider): + assert no_token_provider.use_fl_token is False + + def test_use_fl_token_flag_on(self, provider): + assert provider.use_fl_token is True + + # ---- eligibility ------------------------------------------------------- + + def test_eligible_when_can_use_token_true(self, provider): + assert provider._is_token_eligible(self._release(can_use_token=True)) is True + + def test_ineligible_when_can_use_token_false(self, provider): + # Server authoritatively says no token can be spent (e.g. none left). + assert provider._is_token_eligible(self._release(can_use_token=False)) is False + + def test_ineligible_when_already_freeleech(self, provider): + # canUseToken unknown -> fall back to heuristic. + assert provider._is_token_eligible(self._release(is_freeleech=True)) is False + + def test_ineligible_when_over_5gb(self, provider): + big = provider.FL_TOKEN_MAX_BYTES + 1 + assert provider._is_token_eligible(self._release(size_bytes=big)) is False + + def test_eligible_when_small_and_not_free(self, provider): + small = 300 * 1024 * 1024 + assert provider._is_token_eligible(self._release(size_bytes=small)) is True + + def test_can_use_token_overrides_size_heuristic(self, provider): + # Explicit server signal wins over the local size guess. + r = self._release(can_use_token=True, size_bytes=provider.FL_TOKEN_MAX_BYTES + 1) + assert provider._is_token_eligible(r) is True + + # ---- token URL + fallback --------------------------------------------- + + def test_fetch_with_token_appends_usetoken(self, provider): + provider._fetch_torrent_from_url = MagicMock(return_value=b"d...torrent") + result = provider._fetch_with_optional_token( + "https://test.example/ajax.php?action=download&id=42", "42", use_token=True + ) + assert result == b"d...torrent" + called_url = provider._fetch_torrent_from_url.call_args[0][0] + assert "usetoken=1" in called_url + + def test_fetch_without_token_has_no_usetoken(self, provider): + provider._fetch_torrent_from_url = MagicMock(return_value=b"d...torrent") + provider._fetch_with_optional_token( + "https://test.example/ajax.php?action=download&id=42", "42", use_token=False + ) + called_url = provider._fetch_torrent_from_url.call_args[0][0] + assert "usetoken" not in called_url + + def test_token_failure_falls_back_to_normal_download(self, provider): + # First call (with token) fails, second (without) succeeds. + provider._fetch_torrent_from_url = MagicMock(side_effect=[None, b"d...torrent"]) + result = provider._fetch_with_optional_token( + "https://test.example/ajax.php?action=download&id=42", "42", use_token=True + ) + assert result == b"d...torrent" + assert provider._fetch_torrent_from_url.call_count == 2 + first_url = provider._fetch_torrent_from_url.call_args_list[0][0][0] + second_url = provider._fetch_torrent_from_url.call_args_list[1][0][0] + assert "usetoken=1" in first_url + assert "usetoken" not in second_url + + # ---- fetch_artifact drives eligibility -------------------------------- + + def test_fetch_artifact_spends_token_when_eligible(self, provider): + provider.cache_dir = None # skip cache path + provider._fetch_torrent_from_url = MagicMock(return_value=b"d...torrent") + provider.fetch_artifact(self._release(can_use_token=True)) + called_url = provider._fetch_torrent_from_url.call_args[0][0] + assert "usetoken=1" in called_url + + def test_fetch_artifact_no_token_when_flag_off(self, no_token_provider): + no_token_provider.cache_dir = None + no_token_provider._fetch_torrent_from_url = MagicMock(return_value=b"d...torrent") + no_token_provider.fetch_artifact(self._release(can_use_token=True)) + called_url = no_token_provider._fetch_torrent_from_url.call_args[0][0] + assert "usetoken" not in called_url + + def test_fetch_artifact_no_token_when_ineligible(self, provider): + provider.cache_dir = None + provider._fetch_torrent_from_url = MagicMock(return_value=b"d...torrent") + provider.fetch_artifact(self._release(can_use_token=False)) + called_url = provider._fetch_torrent_from_url.call_args[0][0] + assert "usetoken" not in called_url + + # ---- content validation guards the cache ------------------------------ + + def test_json_error_body_not_treated_as_torrent(self, provider): + provider.cache_dir = None + resp = MagicMock() + resp.status_code = 200 + resp.headers = {"Content-Type": "application/json"} + resp.content = b'{"status":"failure","error":"You do not have any freeleech tokens left."}' + resp.json.return_value = {"status": "failure", "error": "You do not have any freeleech tokens left."} + provider.session.get = MagicMock(return_value=resp) + assert provider._fetch_torrent_from_url("https://test.example/ajax.php?action=download&id=42&usetoken=1", "42") is None + + def test_valid_bencoded_torrent_accepted(self, provider): + provider.cache_dir = None + resp = MagicMock() + resp.status_code = 200 + resp.headers = {"Content-Type": "application/x-bittorrent"} + resp.content = b"d8:announce" + b"x" * 2000 # bencoded dict, > 1000 bytes + provider.session.get = MagicMock(return_value=resp) + result = provider._fetch_torrent_from_url("https://test.example/ajax.php?action=download&id=42", "42") + assert result == resp.content diff --git a/tests/test_red.py b/tests/test_red.py index 9ec3b7a..d9c9071 100644 --- a/tests/test_red.py +++ b/tests/test_red.py @@ -189,3 +189,42 @@ def test_red_provider_name(): """Test that REDProvider returns 'RED' as its name.""" provider = REDProvider(api_key="test", base_url=MOCK_BASE_URL) assert provider.name == "RED" + + +def test_red_captures_can_use_token_from_browse(): + """canUseToken from the browse response should land on matching releases.""" + provider = REDProvider(api_key="test", base_url=MOCK_BASE_URL) + provider.session.get = MagicMock() + + # Browse response carries per-torrent canUseToken (torrentgroup does not). + browse_resp = MagicMock() + browse_resp.status_code = 200 + browse_resp.json.return_value = { + "status": "success", + "response": { + "results": [{ + "groupId": 123, + "torrents": [ + {"torrentId": 29991962, "canUseToken": True}, + {"torrentId": 12345678, "canUseToken": False}, # 24bit, >5GB-ish -> server says no + ], + }], + }, + } + + details_resp = MagicMock() + details_resp.status_code = 200 + details_resp.json.return_value = SAMPLE_GROUP_RESPONSE + + provider.session.get.side_effect = [browse_resp, details_resp] + + releases = provider.search(TrackQuery(artist="Logistics", title="Fear Not")) + by_id = {r.source_id: r for r in releases} + assert by_id["29991962"].can_use_token is True + assert by_id["12345678"].can_use_token is False + + +def test_red_use_fl_token_flag_propagates(): + """use_fl_token passes through REDProvider to the Gazelle base.""" + assert REDProvider(api_key="k", base_url=MOCK_BASE_URL).use_fl_token is False + assert REDProvider(api_key="k", base_url=MOCK_BASE_URL, use_fl_token=True).use_fl_token is True