" in body
+ assert "Źródło:" in body and "Embed:" in body
+ assert p.source == source and resolved.embed_url == embed_url
+
def test_no_info_renders_empty_message(self):
title, body = format_player_source(_PLAYER, None)
assert "CDA" in title
From 5cfce38932ad366ab4bf4c9b07594612cc99f4e4 Mon Sep 17 00:00:00 2001
From: Aleksander <71835339+Cybernetic-Ransomware@users.noreply.github.com>
Date: Sun, 4 Oct 2026 21:43:36 +0200
Subject: [PATCH 10/11] Feat/download source pin (#26)
* feat(download): pin the chosen source for the remaining episodes of a download batch
* fix(download): let --select-nth override the source pin and require a manual pick when the pin has no match
---
src/alt_ani_cli/content/ui_content_pl-pl.yaml | 11 +-
src/alt_ani_cli/flow/handlers.py | 70 +++-
src/alt_ani_cli/flow/pin.py | 50 +++
src/alt_ani_cli/flow/state.py | 5 +
src/alt_ani_cli/ui/menus.py | 24 +-
tests/flow/test_handlers.py | 381 ++++++++++++++++++
tests/flow/test_pin.py | 95 +++++
tests/ui/test_menus.py | 38 ++
8 files changed, 654 insertions(+), 20 deletions(-)
create mode 100644 src/alt_ani_cli/flow/pin.py
create mode 100644 tests/flow/test_pin.py
diff --git a/src/alt_ani_cli/content/ui_content_pl-pl.yaml b/src/alt_ani_cli/content/ui_content_pl-pl.yaml
index c7569c9..95e9103 100644
--- a/src/alt_ani_cli/content/ui_content_pl-pl.yaml
+++ b/src/alt_ani_cli/content/ui_content_pl-pl.yaml
@@ -54,6 +54,7 @@ progress:
player_failed_short: "Player {player} nie zadziałał — wybierz inny:"
no_player_worked: "Żaden player nie zadziałał dla odcinka {number:g} — pomijam."
download_failed: "Pobieranie z playera {player} nie powiodło się."
+ pin_no_match: "Przypiętego źródła nie ma w odcinku {number:g}."
host_deferred: "Host {host} jest chwilowo niedostępny w tej sesji — odkładam próbę."
host_last_resort: "Pozostałe źródła zawiodły — próbuję odłożonego hosta {host}."
host_dropped: "Host {host} nadal jest niedostępny po próbie last-resort — pomijam kolejnego playera na tym hoście."
@@ -166,10 +167,18 @@ menu:
options:
play: "Oglądaj w mpv/vlc"
download: "Pobierz na dysk"
+ download_pin: "Pobierz i używaj tego źródła dla kolejnych odcinków"
debug: "Pokaż linki (debug)"
instruction: "↑↓ = nawigacja | Enter = wybierz | ESC = wstecz"
fallback_prompt: "Akcja [1-3, Enter=wstecz]: "
- fallback_invalid: " Wpisz liczbę 1–3."
+ fallback_invalid: " Wpisz liczbę 1–{n}."
+ pin_fallback:
+ message: "Ten player różni się od przypiętego źródła. Co zrobić?"
+ options:
+ keep: "Pobierz ten odcinek i zachowaj przypięte źródło"
+ repin: "Pobierz i używaj tego źródła dla kolejnych odcinków"
+ instruction: "↑↓ = nawigacja | Enter = wybierz | ESC = wstecz"
+ fallback_invalid: " Wpisz liczbę 1–2."
existing_file:
header: "Plik już istnieje:
{path}"
diff --git a/src/alt_ani_cli/flow/handlers.py b/src/alt_ani_cli/flow/handlers.py
index 1f90203..35d840f 100644
--- a/src/alt_ani_cli/flow/handlers.py
+++ b/src/alt_ani_cli/flow/handlers.py
@@ -21,6 +21,7 @@
from alt_ani_cli import __version__, diagnostics, download, history
from alt_ani_cli.content import CONTENT
from alt_ani_cli.errors import DownloadFailedError, DownloadTargetError, ShindenError
+from alt_ani_cli.flow.pin import find_pinned, player_fingerprint
from alt_ani_cli.flow.state import BACK, FlowState, Screen, ScreenResult
from alt_ani_cli.models import EmbedURL, EpisodeRow, PlayerSource, SeriesHit, SeriesMetadata, SeriesRef
from alt_ani_cli.shinden import api as shinden_api
@@ -110,6 +111,10 @@ def _reset_download_resume(state: FlowState) -> None:
state.downloaded_eps = set()
+def _has_remaining_targets(state: FlowState) -> bool:
+ return state.ep_idx < len(state.targets) - 1
+
+
def _first_missing_index(episodes: list[EpisodeRow], downloaded: set[float]) -> int:
return next((i for i, ep in enumerate(episodes) if ep.number not in downloaded), len(episodes) - 1)
@@ -295,6 +300,7 @@ def handle_fetch_episodes(state: FlowState) -> ScreenResult:
state.targets = []
state.ep_idx = 0
state.overwrite_existing_batch = False
+ state.pinned_player = None
state.episode_arg_consumed = False
if not episodes:
progress.error(_PROG["no_episodes"])
@@ -318,6 +324,7 @@ def handle_episodes_pick(state: FlowState) -> ScreenResult:
state.targets = targets
state.ep_idx = 0
state.overwrite_existing_batch = False
+ state.pinned_player = None
state.episode_arg_consumed = True
return Screen.EPISODE_DISPATCH
@@ -349,6 +356,7 @@ def handle_episodes_pick(state: FlowState) -> ScreenResult:
state.targets = result
state.ep_idx = 0
state.overwrite_existing_batch = False
+ state.pinned_player = None
return Screen.EPISODE_DISPATCH
@@ -388,22 +396,32 @@ def handle_episode_dispatch(state: FlowState) -> ScreenResult:
else:
use_full = menus.confirm(_M["filter_confirm"]["question"].format(filters=filters))
if not use_full: # False or None (ESC) → back to episode selection
+ state.pinned_player = None
return Screen.EPISODES_PICK
else:
players = filtered
state.players = players
state.chosen_player = None
+ state.player_picked_manually = False
state.failed_ids = set()
state.stream = None
state.embed = None
state.player_sources = {}
state.player_embeds = {}
- if args.select_nth or len(players) == 1:
- state.chosen_player = players[0]
- diagnostics.player_selected(state.chosen_player.online_id, state.chosen_player.player, None)
- return Screen.RESOLVE_STREAM
+ if args.select_nth:
+ return _auto_pick_player(state, players[0])
+
+ # matched on the final sorted/filtered list from the episode HTML alone — no resolve_embed needed
+ if state.pinned_player is not None:
+ pinned = find_pinned(players, state.pinned_player)
+ if pinned is not None:
+ return _auto_pick_player(state, pinned)
+ # a missing pinned source needs a deliberate fallback, even with a single player left
+ progress.info(_PROG["pin_no_match"].format(number=ep.number))
+ elif len(players) == 1:
+ return _auto_pick_player(state, players[0])
if args.show_sources:
_prefetch_player_sources(state)
@@ -411,6 +429,12 @@ def handle_episode_dispatch(state: FlowState) -> ScreenResult:
return Screen.PLAYER_PICK
+def _auto_pick_player(state: FlowState, player) -> ScreenResult:
+ state.chosen_player = player
+ diagnostics.player_selected(player.online_id, player.player, None)
+ return Screen.RESOLVE_STREAM
+
+
def handle_player_pick(state: FlowState) -> ScreenResult:
ep = state.current_ep
ep_label = f"{ep.number:g}" if ep is not None else "?"
@@ -426,9 +450,11 @@ def handle_player_pick(state: FlowState) -> ScreenResult:
sources=state.player_sources,
)
if action == "back":
+ state.pinned_player = None # leaving the batch
return Screen.EPISODES_PICK # ESC → back to episode selection
if action == "pick":
state.chosen_player = payload
+ state.player_picked_manually = True
source = state.player_sources.get(payload.online_id)
diagnostics.player_selected(payload.online_id, payload.player, source.host if source else None)
return Screen.RESOLVE_STREAM
@@ -505,18 +531,40 @@ def handle_action_pick(state: FlowState) -> ScreenResult:
state.episode_action = "debug"
return Screen.RUN_ACTION
if state.episode_action is not None:
+ if state.episode_action == "download" and _offers_repin(state):
+ choice = menus.select_pin_fallback_action()
+ if choice is None:
+ return _back_from_action(state)
+ if choice == "repin":
+ state.pinned_player = player_fingerprint(state.chosen_player)
return Screen.RUN_ACTION
- action = menus.select_action()
+ action = menus.select_action(offer_pin=_has_remaining_targets(state))
if action is None:
state.episode_action = None # reset cache
- # ESC → back to quality pick if there were qualities, otherwise player
- if state.stream and state.stream.qualities:
- return Screen.QUALITY_PICK
- return Screen.PLAYER_PICK
+ return _back_from_action(state)
+ if action == "download_pin":
+ state.pinned_player = player_fingerprint(state.chosen_player)
+ action = "download"
state.episode_action = action
return Screen.RUN_ACTION
+def _offers_repin(state: FlowState) -> bool:
+ """A manual fallback must not silently replace the pin, so ask when the picked player differs from it."""
+ return (
+ state.pinned_player is not None
+ and state.player_picked_manually
+ and _has_remaining_targets(state)
+ and player_fingerprint(state.chosen_player) != state.pinned_player
+ )
+
+
+def _back_from_action(state: FlowState) -> ScreenResult:
+ if state.stream and state.stream.qualities:
+ return Screen.QUALITY_PICK
+ return Screen.PLAYER_PICK
+
+
def handle_run_action(state: FlowState) -> ScreenResult:
if state.stream is None:
raise AssertionError
@@ -543,8 +591,7 @@ def handle_run_action(state: FlowState) -> ScreenResult:
def confirm_overwrite(path: Path) -> bool:
if state.overwrite_existing_batch:
return True
- has_remaining = state.ep_idx < len(state.targets) - 1
- choice = menus.select_existing_download_action(str(path), has_remaining=has_remaining)
+ choice = menus.select_existing_download_action(str(path), has_remaining=_has_remaining_targets(state))
existing.update(choice=choice, path=path)
if choice == "overwrite_remaining":
state.overwrite_existing_batch = True
@@ -566,6 +613,7 @@ def confirm_overwrite(path: Path) -> bool:
return Screen.PLAYER_PICK
if not downloaded and existing["choice"] == "cancel":
state.overwrite_existing_batch = False
+ state.pinned_player = None
state.stream = None
state.embed = None
progress.warn(CONTENT["download"]["batch_cancelled"])
diff --git a/src/alt_ani_cli/flow/pin.py b/src/alt_ani_cli/flow/pin.py
new file mode 100644
index 0000000..0110476
--- /dev/null
+++ b/src/alt_ani_cli/flow/pin.py
@@ -0,0 +1,50 @@
+"""Session-only download source pin: a player identity that survives across episodes of one batch."""
+
+from dataclasses import dataclass
+from urllib.parse import urlparse
+
+from alt_ani_cli.models import PlayerEntry
+
+
+@dataclass(frozen=True)
+class PlayerFingerprint:
+ """Episode-independent player identity; online_id and date_added change between episodes, so they are excluded."""
+
+ player: str
+ lang_audio: str
+ lang_subs: str
+ max_res: str
+ subs_author: str
+ source: str
+
+
+def _norm_text(value: str | None) -> str:
+ return " ".join((value or "").split()).casefold()
+
+
+def source_identity(source: str | None) -> str:
+ """Hostname without www. for URLs (scheme, path, query and fragment ignored); normalized text otherwise."""
+ text = (source or "").strip()
+ try:
+ host = urlparse(text).hostname
+ except ValueError:
+ host = None
+ if host:
+ return host.casefold().removeprefix("www.")
+ return _norm_text(text)
+
+
+def player_fingerprint(p: PlayerEntry) -> PlayerFingerprint:
+ return PlayerFingerprint(
+ player=_norm_text(p.player),
+ lang_audio=_norm_text(p.lang_audio),
+ lang_subs=_norm_text(p.lang_subs),
+ max_res=_norm_text(p.max_res),
+ subs_author=_norm_text(p.subs_author),
+ source=source_identity(p.source),
+ )
+
+
+def find_pinned(players: list[PlayerEntry], pin: PlayerFingerprint) -> PlayerEntry | None:
+ """First player matching the pin, in the given (already sorted and filtered) order."""
+ return next((p for p in players if player_fingerprint(p) == pin), None)
diff --git a/src/alt_ani_cli/flow/state.py b/src/alt_ani_cli/flow/state.py
index da104e4..5437dc7 100644
--- a/src/alt_ani_cli/flow/state.py
+++ b/src/alt_ani_cli/flow/state.py
@@ -4,6 +4,7 @@
from enum import Enum, auto
from typing import Any, Literal
+from alt_ani_cli.flow.pin import PlayerFingerprint
from alt_ani_cli.health import ResolverHealth
@@ -71,10 +72,14 @@ class FlowState:
# replace already-downloaded files without asking — scoped to the current targets batch only
overwrite_existing_batch: bool = False
episode_arg_consumed: bool = False
+ # download source pin — scoped to the current targets batch, never persisted
+ pinned_player: PlayerFingerprint | None = None
# player / stream
players: list = field(default_factory=list)
chosen_player: Any = None
+ # True only when chosen_player came from PLAYER_PICK (not pin / --select-nth / single-player auto-pick)
+ player_picked_manually: bool = False
failed_ids: set[str] = field(default_factory=set)
stream: Any = None
embed: Any = None
diff --git a/src/alt_ani_cli/ui/menus.py b/src/alt_ani_cli/ui/menus.py
index e4a3b8f..f2e2d17 100644
--- a/src/alt_ani_cli/ui/menus.py
+++ b/src/alt_ani_cli/ui/menus.py
@@ -712,19 +712,27 @@ def _label(opt: str) -> str:
return _run_simple_picker(all_options, _label, prompt=prompt, instruction=_q["instruction"], mode="select")
-def select_action() -> Literal["play", "download", "debug"] | None:
+def select_action(*, offer_pin: bool = False) -> Literal["play", "download", "download_pin", "debug"] | None:
+ """offer_pin adds "download and keep using this source" — only meaningful when later episodes remain."""
_ac = _M["action"]
_ac_opts = _ac["options"]
- _options: list[tuple[str, str]] = [
- ("play", _ac_opts["play"]),
- ("download", _ac_opts["download"]),
- ("debug", _ac_opts["debug"]),
- ]
+ keys = ["play", "download", "download_pin", "debug"] if offer_pin else ["play", "download", "debug"]
return _run_keyed_picker( # type: ignore[return-value]
- _options,
+ [(k, _ac_opts[k]) for k in keys],
prompt=_ac["message"],
instruction=_ac["instruction"],
- fallback_invalid=_ac["fallback_invalid"],
+ fallback_invalid=_ac["fallback_invalid"].format(n=len(keys)),
+ )
+
+
+def select_pin_fallback_action() -> Literal["keep", "repin"] | None:
+ _pf = _M["pin_fallback"]
+ _opts = _pf["options"]
+ return _run_keyed_picker( # type: ignore[return-value]
+ [("keep", _opts["keep"]), ("repin", _opts["repin"])],
+ prompt=_pf["message"],
+ instruction=_pf["instruction"],
+ fallback_invalid=_pf["fallback_invalid"],
)
diff --git a/tests/flow/test_handlers.py b/tests/flow/test_handlers.py
index d7ee305..f11b232 100644
--- a/tests/flow/test_handlers.py
+++ b/tests/flow/test_handlers.py
@@ -23,6 +23,7 @@
_sorted_by_date_desc,
handle_run_action,
)
+from alt_ani_cli.flow.pin import player_fingerprint
from alt_ani_cli.flow.state import BACK, FlowState, Screen, _BackSentinel
from alt_ani_cli.models import EmbedURL, PlayerSource, RelatedSeries, SeriesMetadata
from alt_ani_cli.player.runner import PlayResult
@@ -1959,3 +1960,383 @@ def test_player_pick_offers_same_player_again(self):
assert HANDLERS[Screen.PLAYER_PICK](state) is Screen.RESOLVE_STREAM
assert mock_pick.call_args.kwargs["failed"] == set()
assert state.chosen_player is _PLAYER
+
+
+_PIN_A1 = PlayerEntry(
+ online_id="a1", player="CDA", lang_audio="jp", lang_subs="pl", subs_author="Mioro", source="https://www.miorosubs.com/?ep=4"
+)
+_PIN_A2 = PlayerEntry(
+ online_id="a2", player="CDA", lang_audio="jp", lang_subs="pl", subs_author="Mioro", source="http://miorosubs.com/?ep=5"
+)
+_PIN_A3 = PlayerEntry(
+ online_id="a3", player="CDA", lang_audio="jp", lang_subs="pl", subs_author="Mioro", source="https://miorosubs.com/x"
+)
+_OTHER = PlayerEntry(online_id="o1", player="Sibnet", lang_audio="jp", lang_subs="pl", source="https://anisubs.pl/")
+_OTHER_EN = PlayerEntry(online_id="o2", player="CDA", lang_audio="jp", lang_subs="en")
+_PIN_FP = player_fingerprint(_PIN_A1)
+
+
+def _pin_batch_state(**overrides) -> FlowState:
+ defaults = dict(
+ targets=[_EP4, _EP5, _EP6],
+ ep_idx=1,
+ players=[_PIN_A1, _OTHER],
+ chosen_player=_PIN_A1,
+ episode_action="download",
+ pinned_player=_PIN_FP,
+ )
+ defaults.update(overrides)
+ return _download_state(**defaults)
+
+
+def _pin_dispatch(players, *, state=None, confirm=True, **arg_overrides):
+ """Run EPISODE_DISPATCH with parse/sort mocked; sort_players returns `players` as the final order."""
+ if state is None:
+ state = _pin_batch_state(args=_make_args(**arg_overrides))
+ resp = MagicMock()
+ resp.text = ""
+ state.client.get.return_value = resp
+ with (
+ patch("alt_ani_cli.shinden.episode.parse_players", return_value=list(players)),
+ patch("alt_ani_cli.shinden.episode.sort_players", return_value=list(players)) as mock_sort,
+ patch("alt_ani_cli.shinden.api.resolve_embed") as mock_embed,
+ patch("alt_ani_cli.diagnostics.player_selected") as mock_diag,
+ patch("alt_ani_cli.ui.menus.confirm", return_value=confirm),
+ patch("alt_ani_cli.ui.progress.info") as mock_info,
+ patch("alt_ani_cli.ui.progress.warn"),
+ ):
+ result = HANDLERS[Screen.EPISODE_DISPATCH](state)
+ return state, result, {"sort": mock_sort, "embed": mock_embed, "diag": mock_diag, "info": mock_info}
+
+
+@pytest.mark.unit
+class TestPinFirstSelection:
+ def test_download_and_pin_sets_pin_and_downloads_current_episode(self):
+ state = _pin_batch_state(ep_idx=0, episode_action=None, pinned_player=None)
+ with patch("alt_ani_cli.ui.menus.select_action", return_value="download_pin") as mock_action:
+ assert HANDLERS[Screen.ACTION_PICK](state) is Screen.RUN_ACTION
+ assert mock_action.call_args.kwargs == {"offer_pin": True}
+ assert state.episode_action == "download"
+ assert state.pinned_player == _PIN_FP
+
+ fake = _FakeDownloads()
+ result, _, record = _run_download(state, fake)
+ assert fake.events == [("downloaded", 4.0)]
+ record.assert_called_once_with(_SERIES_REF, 4.0)
+ assert (result, state.ep_idx) == (Screen.EPISODE_DISPATCH, 1)
+ assert state.pinned_player == _PIN_FP
+
+ @pytest.mark.parametrize(("targets", "ep_idx"), [([_EP4], 0), ([_EP4, _EP5], 1)])
+ def test_option_hidden_for_last_or_only_target(self, targets, ep_idx):
+ state = _pin_batch_state(targets=targets, ep_idx=ep_idx, episode_action=None, pinned_player=None)
+ with patch("alt_ani_cli.ui.menus.select_action", return_value="download") as mock_action:
+ HANDLERS[Screen.ACTION_PICK](state)
+ assert mock_action.call_args.kwargs == {"offer_pin": False}
+ assert state.pinned_player is None
+
+ def test_plain_download_does_not_pin(self):
+ state = _pin_batch_state(ep_idx=0, episode_action=None, pinned_player=None)
+ with patch("alt_ani_cli.ui.menus.select_action", return_value="download"):
+ HANDLERS[Screen.ACTION_PICK](state)
+ assert state.pinned_player is None
+
+ def test_pinning_does_not_touch_overwrite_policy(self):
+ state = _pin_batch_state(ep_idx=0, episode_action=None, pinned_player=None)
+ with patch("alt_ani_cli.ui.menus.select_action", return_value="download_pin"):
+ HANDLERS[Screen.ACTION_PICK](state)
+ assert state.overwrite_existing_batch is False
+
+
+@pytest.mark.unit
+class TestPinAutoMatch:
+ def test_same_fingerprint_new_online_id_skips_player_pick(self):
+ state, result, mocks = _pin_dispatch([_OTHER, _PIN_A2])
+ assert result is Screen.RESOLVE_STREAM
+ assert state.chosen_player is _PIN_A2
+ assert state.player_picked_manually is False
+ mocks["diag"].assert_called_once_with(_PIN_A2.online_id, _PIN_A2.player, None)
+ assert state.pinned_player == _PIN_FP
+
+ def test_matching_issues_no_extra_requests(self):
+ state, _, mocks = _pin_dispatch([_OTHER, _PIN_A2])
+ mocks["embed"].assert_not_called()
+ state.client.get.assert_called_once_with(_EP5.url)
+ assert state.player_sources == {}
+
+ def test_cached_download_action_keeps_download_sort(self):
+ _, _, mocks = _pin_dispatch([_OTHER, _PIN_A2])
+ assert mocks["sort"].call_args.kwargs == {"download": True}
+
+ def test_first_of_several_identical_fingerprints_in_sorted_order(self):
+ state, _, _ = _pin_dispatch([_OTHER, _PIN_A3, _PIN_A2])
+ assert state.chosen_player is _PIN_A3
+
+ def test_match_runs_on_filtered_list(self):
+ state, result, _ = _pin_dispatch([_OTHER_EN, _OTHER, _PIN_A2], subs="pl")
+ assert result is Screen.RESOLVE_STREAM
+ assert state.players == [_OTHER, _PIN_A2]
+ assert state.chosen_player is _PIN_A2
+
+ def test_pinned_candidate_filtered_out_shows_normal_picker(self):
+ state, result, _ = _pin_dispatch([_OTHER, _PIN_A2, _OTHER_EN], player_name="CDA", subs="en")
+ assert result is Screen.PLAYER_PICK
+ assert state.players == [_OTHER_EN]
+ assert state.chosen_player is None
+ assert state.pinned_player == _PIN_FP
+
+ def test_pinned_candidate_filtered_out_of_several_shows_picker(self):
+ other_sibnet = PlayerEntry(online_id="o3", player="Sibnet", lang_audio="jp", lang_subs="en")
+ state, result, _ = _pin_dispatch([_OTHER, _PIN_A2, other_sibnet], player_name="Sibnet")
+ assert result is Screen.PLAYER_PICK
+ assert state.players == [_OTHER, other_sibnet]
+ assert state.chosen_player is None
+ assert state.pinned_player == _PIN_FP
+
+ def test_filter_fallback_full_list_can_match_pin(self):
+ state, result, _ = _pin_dispatch([_OTHER, _PIN_A2], lang="xx", allow_fallback=True)
+ assert result is Screen.RESOLVE_STREAM
+ assert state.chosen_player is _PIN_A2
+
+ def test_no_match_shows_picker_and_keeps_pin(self):
+ state, result, mocks = _pin_dispatch([_OTHER, _OTHER_EN])
+ assert result is Screen.PLAYER_PICK
+ assert state.chosen_player is None
+ assert state.pinned_player == _PIN_FP
+ mocks["info"].assert_called_with(CONTENT["progress"]["pin_no_match"].format(number=5.0))
+
+ def test_without_pin_dispatch_is_unchanged(self):
+ state, result, mocks = _pin_dispatch([_OTHER, _PIN_A2], state=_pin_batch_state(pinned_player=None))
+ assert result is Screen.PLAYER_PICK
+ mocks["diag"].assert_not_called()
+ no_match = CONTENT["progress"]["pin_no_match"].format(number=5.0)
+ assert all(c.args[0] != no_match for c in mocks["info"].call_args_list)
+
+ def test_select_nth_takes_precedence_over_pin(self):
+ state, result, mocks = _pin_dispatch([_OTHER, _PIN_A2], select_nth=1)
+ assert result is Screen.RESOLVE_STREAM
+ assert state.chosen_player is _OTHER
+ assert state.pinned_player == _PIN_FP
+ mocks["diag"].assert_called_once_with(_OTHER.online_id, _OTHER.player, None)
+
+ def test_pin_without_match_and_single_player_shows_picker(self):
+ state, result, mocks = _pin_dispatch([_OTHER])
+ assert result is Screen.PLAYER_PICK
+ assert state.chosen_player is None
+ assert state.pinned_player == _PIN_FP
+ mocks["diag"].assert_not_called()
+
+ def test_pin_with_match_and_single_player_auto_picks_match(self):
+ state, result, _ = _pin_dispatch([_PIN_A2])
+ assert result is Screen.RESOLVE_STREAM
+ assert state.chosen_player is _PIN_A2
+
+ def test_without_pin_single_player_is_still_auto_selected(self):
+ state, result, mocks = _pin_dispatch([_OTHER], state=_pin_batch_state(pinned_player=None))
+ assert result is Screen.RESOLVE_STREAM
+ assert state.chosen_player is _OTHER
+ mocks["diag"].assert_called_once_with(_OTHER.online_id, _OTHER.player, None)
+
+ def test_select_nth_still_applies_when_pin_has_no_match(self):
+ state, result, _ = _pin_dispatch([_OTHER, _OTHER_EN], select_nth=1)
+ assert result is Screen.RESOLVE_STREAM
+ assert state.chosen_player is _OTHER
+
+
+@pytest.mark.unit
+class TestPinnedPlayerFailure:
+ def test_resolve_failure_returns_to_player_pick_and_keeps_pin(self):
+ state, _, _ = _pin_dispatch([_OTHER, _PIN_A2])
+ with patch("alt_ani_cli.cli._resolve_with_fallback", return_value=(None, None)):
+ assert HANDLERS[Screen.RESOLVE_STREAM](state) is Screen.PLAYER_PICK
+ assert state.failed_ids == {_PIN_A2.online_id}
+ assert state.ep_idx == 1
+ assert state.pinned_player == _PIN_FP
+
+ with (
+ patch("alt_ani_cli.ui.menus.select_player_once", return_value=("pick", _OTHER)) as mock_pick,
+ patch("alt_ani_cli.ui.progress.warn"),
+ ):
+ assert HANDLERS[Screen.PLAYER_PICK](state) is Screen.RESOLVE_STREAM
+ assert mock_pick.call_args.kwargs["failed"] == {_PIN_A2.online_id}
+ assert state.chosen_player is _OTHER
+
+ def test_download_failed_returns_to_player_pick_and_keeps_pin(self):
+ state = _pin_batch_state(players=[_OTHER, _PIN_A2], chosen_player=_PIN_A2)
+ result, _, record = _run_download(state, _FakeDownloads(fail_urls={_STREAM.url}))
+ assert result is Screen.PLAYER_PICK
+ assert state.failed_ids == {_PIN_A2.online_id}
+ assert state.ep_idx == 1
+ assert state.pinned_player == _PIN_FP
+ record.assert_not_called()
+
+ def test_download_target_error_keeps_pin_and_player(self):
+ state = _pin_batch_state(players=[_OTHER, _PIN_A2], chosen_player=_PIN_A2)
+ with (
+ patch("alt_ani_cli.download.run", side_effect=DownloadTargetError("locked")),
+ patch("alt_ani_cli.history.record_download") as mock_record,
+ patch("alt_ani_cli.ui.progress.error"),
+ ):
+ assert handle_run_action(state) is Screen.PLAYER_PICK
+ assert state.failed_ids == set()
+ assert state.ep_idx == 1
+ assert state.pinned_player == _PIN_FP
+ mock_record.assert_not_called()
+
+
+def _manual_fallback(picked, *, state=None, pin_choice=None, qualities=None):
+ """PLAYER_PICK (manual pick) → RESOLVE_STREAM → ACTION_PICK with download cached and pin active."""
+ if state is None:
+ state = _pin_batch_state(players=[_PIN_A2, _OTHER, _PIN_A3], chosen_player=None)
+ stream = Stream(url=_STREAM.url, ext="m3u8", qualities=qualities or {})
+ with (
+ patch("alt_ani_cli.ui.menus.select_player_once", return_value=("pick", picked)),
+ patch("alt_ani_cli.cli._resolve_with_fallback", return_value=(stream, _EMBED)),
+ patch("alt_ani_cli.ui.menus.select_pin_fallback_action", return_value=pin_choice) as mock_pin,
+ patch("alt_ani_cli.ui.menus.select_action") as mock_action,
+ patch("alt_ani_cli.ui.progress.warn"),
+ ):
+ assert HANDLERS[Screen.PLAYER_PICK](state) is Screen.RESOLVE_STREAM
+ state.quality = "best"
+ assert HANDLERS[Screen.RESOLVE_STREAM](state) is Screen.ACTION_PICK
+ result = HANDLERS[Screen.ACTION_PICK](state)
+ mock_action.assert_not_called()
+ return state, result, mock_pin
+
+
+@pytest.mark.unit
+class TestPinManualFallback:
+ def test_keep_downloads_without_changing_pin(self):
+ state, result, mock_pin = _manual_fallback(_OTHER, pin_choice="keep")
+ mock_pin.assert_called_once()
+ assert result is Screen.RUN_ACTION
+ assert state.pinned_player == _PIN_FP
+ assert state.episode_action == "download"
+
+ def test_repin_replaces_pin_with_current_player(self):
+ state, result, _ = _manual_fallback(_OTHER, pin_choice="repin")
+ assert result is Screen.RUN_ACTION
+ assert state.pinned_player == player_fingerprint(_OTHER)
+
+ def test_same_fingerprint_manual_pick_does_not_ask(self):
+ state, result, mock_pin = _manual_fallback(_PIN_A3)
+ mock_pin.assert_not_called()
+ assert result is Screen.RUN_ACTION
+ assert state.pinned_player == _PIN_FP
+
+ def test_last_target_does_not_ask(self):
+ state = _pin_batch_state(ep_idx=2, players=[_PIN_A2, _OTHER], chosen_player=None)
+ state, result, mock_pin = _manual_fallback(_OTHER, state=state)
+ mock_pin.assert_not_called()
+ assert result is Screen.RUN_ACTION
+ assert state.pinned_player == _PIN_FP
+
+ def test_esc_returns_to_player_pick_keeping_pin_and_action(self):
+ state, result, _ = _manual_fallback(_OTHER, pin_choice=None)
+ assert result is Screen.PLAYER_PICK
+ assert state.pinned_player == _PIN_FP
+ assert state.episode_action == "download"
+
+ def test_esc_with_qualities_returns_to_quality_pick(self):
+ _, result, _ = _manual_fallback(_OTHER, pin_choice=None, qualities={"720p": "u"})
+ assert result is Screen.QUALITY_PICK
+
+ def test_auto_picked_player_with_other_fingerprint_does_not_ask(self):
+ state, _, _ = _pin_dispatch([_OTHER, _OTHER_EN], select_nth=1)
+ state.stream = _STREAM
+ with patch("alt_ani_cli.ui.menus.select_pin_fallback_action") as mock_pin:
+ assert HANDLERS[Screen.ACTION_PICK](state) is Screen.RUN_ACTION
+ mock_pin.assert_not_called()
+
+ def test_manual_flag_resets_on_next_episode(self):
+ state, _, _ = _manual_fallback(_OTHER, pin_choice="keep")
+ assert state.player_picked_manually is True
+ state.ep_idx = 2
+ _pin_dispatch([_OTHER, _PIN_A2], state=state)
+ assert state.player_picked_manually is False
+
+
+@pytest.mark.unit
+class TestPinReset:
+ def test_new_episode_batch_clears_pin(self):
+ state = _make_state(ref=_SERIES_REF, episodes=[_EP4, _EP5], pinned_player=_PIN_FP)
+ with patch("alt_ani_cli.ui.menus.select_episodes", return_value=[_EP5]):
+ HANDLERS[Screen.EPISODES_PICK](state)
+ assert state.pinned_player is None
+
+ def test_cli_episode_range_batch_clears_pin(self):
+ state = _make_state(ref=_SERIES_REF, episodes=[_EP4, _EP5], pinned_player=_PIN_FP, args=_make_args(episode="4-5"))
+ HANDLERS[Screen.EPISODES_PICK](state)
+ assert state.pinned_player is None
+
+ def test_cancel_batch_clears_pin(self):
+ state = _pin_batch_state(chosen_player=_PIN_A2)
+ result, _, _ = _run_download(state, _FakeDownloads(existing={5.0}), "cancel")
+ assert result is Screen.EPISODES_PICK
+ assert state.pinned_player is None
+
+ def test_esc_from_player_pick_clears_pin(self):
+ state = _pin_batch_state()
+ with patch("alt_ani_cli.ui.menus.select_player_once", return_value=("back", None)):
+ assert HANDLERS[Screen.PLAYER_PICK](state) is Screen.EPISODES_PICK
+ assert state.pinned_player is None
+
+ def test_new_series_clears_pin(self):
+ state = _make_state(ref=_SERIES_REF, pinned_player=_PIN_FP)
+ with patch("alt_ani_cli.shinden.series.list_episodes", return_value=(_SERIES_REF, [_EP1])):
+ HANDLERS[Screen.FETCH_EPISODES](state)
+ assert state.pinned_player is None
+
+ @pytest.mark.parametrize("answer", [False, None])
+ def test_filter_mismatch_back_to_episode_pick_clears_pin(self, answer):
+ state = _pin_batch_state(args=_make_args(lang="xx"))
+ state, result, _ = _pin_dispatch([_OTHER, _PIN_A2], state=state, confirm=answer)
+ assert result is Screen.EPISODES_PICK
+ assert state.pinned_player is None
+
+ def test_quality_pick_keeps_pin(self):
+ state = _pin_batch_state(stream=Stream(url=_STREAM.url, ext="m3u8", qualities={"720p": "u"}))
+ with patch("alt_ani_cli.ui.menus.select_quality", return_value="720p"):
+ HANDLERS[Screen.QUALITY_PICK](state)
+ assert state.pinned_player == _PIN_FP
+
+ def test_next_episode_in_batch_keeps_pin(self):
+ state = _pin_batch_state(chosen_player=_PIN_A2)
+ result, _, _ = _run_download(state, _FakeDownloads())
+ assert (result, state.ep_idx) == (Screen.EPISODE_DISPATCH, 2)
+ assert state.pinned_player == _PIN_FP
+
+
+@pytest.mark.unit
+class TestPinIndependentOfOverwritePolicy:
+ def test_overwrite_remaining_does_not_pin(self):
+ state = _pin_batch_state(pinned_player=None, chosen_player=_PIN_A2)
+ _run_download(state, _FakeDownloads(existing={5.0}), "overwrite_remaining")
+ assert state.overwrite_existing_batch is True
+ assert state.pinned_player is None
+
+ def test_policy_and_pin_both_survive_pinned_download_failure(self):
+ state = _pin_batch_state(players=[_OTHER, _PIN_A2], chosen_player=_PIN_A2, overwrite_existing_batch=True)
+ _run_download(state, _FakeDownloads(existing={5.0}, fail_urls={_STREAM.url}))
+ assert state.overwrite_existing_batch is True
+ assert state.pinned_player == _PIN_FP
+
+ def test_repin_keeps_overwrite_policy(self):
+ state = _pin_batch_state(players=[_PIN_A2, _OTHER], chosen_player=None, overwrite_existing_batch=True)
+ state, _, _ = _manual_fallback(_OTHER, state=state, pin_choice="repin")
+ assert state.overwrite_existing_batch is True
+
+
+@pytest.mark.unit
+class TestPinNotPersisted:
+ def test_history_file_has_no_pin_data(self):
+ from alt_ani_cli import history
+
+ state = _pin_batch_state(chosen_player=_PIN_A2)
+ with (
+ patch("alt_ani_cli.download.run", side_effect=_FakeDownloads()),
+ patch("alt_ani_cli.ui.progress.info"),
+ ):
+ handle_run_action(state)
+ text = history.HISTORY_FILE.read_text(encoding="utf-8")
+ assert "miorosubs" not in text
+ assert "pinned" not in text.casefold()
diff --git a/tests/flow/test_pin.py b/tests/flow/test_pin.py
new file mode 100644
index 0000000..0d916a1
--- /dev/null
+++ b/tests/flow/test_pin.py
@@ -0,0 +1,95 @@
+"""Tests for the download source pin fingerprint (flow/pin.py)."""
+
+import dataclasses
+
+import pytest
+
+from alt_ani_cli.flow.pin import PlayerFingerprint, find_pinned, player_fingerprint, source_identity
+from alt_ani_cli.models import PlayerEntry
+
+_BASE = PlayerEntry(
+ online_id="111",
+ player="CDA",
+ lang_audio="jp",
+ lang_subs="pl",
+ max_res="1080p",
+ date_added="2024-01-01",
+ subs_author="Mioro-Subs",
+ source="https://www.miorosubs.com/?ep=4",
+)
+
+
+@pytest.mark.unit
+class TestPlayerFingerprint:
+ def test_ignores_online_id_and_date_added(self):
+ other = dataclasses.replace(_BASE, online_id="222", date_added="2024-02-02")
+ assert player_fingerprint(other) == player_fingerprint(_BASE)
+
+ @pytest.mark.parametrize(
+ ("field", "value"),
+ [
+ ("player", "Sibnet"),
+ ("lang_audio", "pl"),
+ ("lang_subs", "en"),
+ ("max_res", "720p"),
+ ("subs_author", "Other-Subs"),
+ ],
+ )
+ def test_identity_field_change_gives_different_fingerprint(self, field, value):
+ other = dataclasses.replace(_BASE, **{field: value})
+ assert player_fingerprint(other) != player_fingerprint(_BASE)
+
+ def test_different_source_hosts_give_different_fingerprint(self):
+ other = dataclasses.replace(_BASE, source="https://anisubs.pl/?ep=4")
+ assert player_fingerprint(other) != player_fingerprint(_BASE)
+
+ def test_source_url_episode_query_does_not_split_identity(self):
+ other = dataclasses.replace(_BASE, online_id="222", source="http://miorosubs.com/?ep=5")
+ assert player_fingerprint(other) == player_fingerprint(_BASE)
+
+ def test_text_fields_normalize_whitespace_and_case(self):
+ other = dataclasses.replace(_BASE, player=" cda ", subs_author="mioro-subs", max_res="1080P")
+ assert player_fingerprint(other) == player_fingerprint(_BASE)
+
+ def test_none_and_empty_optional_fields_are_equal(self):
+ a = PlayerEntry(online_id="1", player="CDA", lang_audio="jp", lang_subs="")
+ b = PlayerEntry(online_id="2", player="CDA", lang_audio="jp", lang_subs="", max_res="", subs_author=" ", source="")
+ assert player_fingerprint(a) == player_fingerprint(b)
+
+ def test_is_hashable_value(self):
+ fp = player_fingerprint(_BASE)
+ assert isinstance(fp, PlayerFingerprint)
+ assert {fp} == {player_fingerprint(_BASE)}
+
+
+@pytest.mark.unit
+class TestSourceIdentity:
+ def test_url_variants_share_host_identity(self):
+ assert source_identity("https://www.miorosubs.com/?ep=4") == "miorosubs.com"
+ assert source_identity("http://miorosubs.com/?ep=5") == "miorosubs.com"
+
+ def test_hostname_is_casefolded_without_www(self):
+ assert source_identity("https://www.MioroSubs.com/?episode=4#top") == "miorosubs.com"
+
+ def test_non_url_text_collapses_whitespace_and_case(self):
+ assert source_identity(" Grupa Mioro\tSubs ") == "grupa mioro subs"
+
+ def test_none_and_blank_are_empty(self):
+ assert source_identity(None) == ""
+ assert source_identity(" ") == ""
+
+ def test_malformed_url_falls_back_to_text(self):
+ assert source_identity("http://[broken") == "http://[broken"
+
+
+@pytest.mark.unit
+class TestFindPinned:
+ def test_returns_first_match_in_given_order(self):
+ a = dataclasses.replace(_BASE, online_id="a")
+ b = dataclasses.replace(_BASE, online_id="b")
+ other = dataclasses.replace(_BASE, online_id="c", player="Sibnet")
+ assert find_pinned([other, b, a], player_fingerprint(_BASE)) is b
+
+ def test_no_match_returns_none(self):
+ other = dataclasses.replace(_BASE, player="Sibnet")
+ assert find_pinned([other], player_fingerprint(_BASE)) is None
diff --git a/tests/ui/test_menus.py b/tests/ui/test_menus.py
index 08f729c..9128d60 100644
--- a/tests/ui/test_menus.py
+++ b/tests/ui/test_menus.py
@@ -19,6 +19,7 @@
select_action,
select_episodes,
select_existing_download_action,
+ select_pin_fallback_action,
select_player_once,
select_quality,
select_series_from_download_history,
@@ -154,6 +155,42 @@ def test_empty_enter_returns_none(self, monkeypatch):
with patch("builtins.input", return_value=""):
assert select_action() is None
+ def test_pin_option_hidden_by_default(self, monkeypatch):
+ monkeypatch.setattr("alt_ani_cli.ui.menus._USE_INQUIRER", False)
+ printed: list[str] = []
+ with (
+ patch("builtins.input", return_value="1"),
+ patch("builtins.print", side_effect=lambda *a, **k: printed.append(" ".join(str(x) for x in a))),
+ ):
+ select_action()
+ assert not any(_AC_OPTS["download_pin"] in line for line in printed)
+
+ @pytest.mark.parametrize(("answer", "expected"), [("1", "play"), ("2", "download"), ("3", "download_pin"), ("4", "debug")])
+ def test_offer_pin_adds_option_after_download(self, monkeypatch, answer, expected):
+ monkeypatch.setattr("alt_ani_cli.ui.menus._USE_INQUIRER", False)
+ printed: list[str] = []
+ with (
+ patch("builtins.input", return_value=answer),
+ patch("builtins.print", side_effect=lambda *a, **k: printed.append(" ".join(str(x) for x in a))),
+ ):
+ assert select_action(offer_pin=True) == expected
+ assert printed[2] == f" 3. {_AC_OPTS['download_pin']}"
+
+
+@pytest.mark.unit
+class TestSelectPinFallbackAction:
+ @pytest.mark.parametrize(("answer", "expected"), [("1", "keep"), ("2", "repin"), ("", None)])
+ def test_fallback_maps_options(self, monkeypatch, answer, expected):
+ monkeypatch.setattr("alt_ani_cli.ui.menus._USE_INQUIRER", False)
+ printed: list[str] = []
+ with (
+ patch("builtins.input", return_value=answer),
+ patch("builtins.print", side_effect=lambda *a, **k: printed.append(" ".join(str(x) for x in a))),
+ ):
+ assert select_pin_fallback_action() == expected
+ opts = CONTENT["menu"]["pin_fallback"]["options"]
+ assert printed[:2] == [f" 1. {opts['keep']}", f" 2. {opts['repin']}"]
+
@pytest.mark.unit
class TestSelectEpisodes:
@@ -515,6 +552,7 @@ def test_empty_enter_returns_none(self, monkeypatch):
assert confirm("Kontynuować?") is None
+_AC_OPTS = CONTENT["menu"]["action"]["options"]
_EF = CONTENT["menu"]["existing_file"]
_EXISTING = r"C:\dl\Show - ep4.mp4"
From e339c3b2e5bef7cf84d94adef8fd5c4cb8436c42 Mon Sep 17 00:00:00 2001
From: Aleksander <71835339+Cybernetic-Ransomware@users.noreply.github.com>
Date: Mon, 5 Oct 2026 03:08:53 +0200
Subject: [PATCH 11/11] test(extract): add fixture-based regression tests for
custom extractors (#27)
---
tests/extract/conftest.py | 54 +++++++++++++++++++
tests/extract/test_dood.py | 23 ++++++++
tests/extract/test_fixture_files.py | 34 ++++++++++++
tests/extract/test_flyf.py | 23 ++++++++
tests/extract/test_jwplayer.py | 21 ++++++++
tests/extract/test_lycoris.py | 26 +++++++++
tests/extract/test_mp4upload.py | 17 ++++++
tests/extract/test_playmate.py | 23 ++++++++
tests/extract/test_streamtape.py | 18 +++++++
tests/extract/test_uqload.py | 19 +++++++
tests/extract/test_vidara.py | 21 ++++++++
tests/fixtures/extract/dood_embed.html | 23 ++++++++
tests/fixtures/extract/dood_pass_md5.txt | 1 +
tests/fixtures/extract/flyf_assign.json | 4 ++
.../extract/jwplayer_morencius_packed.html | 17 ++++++
tests/fixtures/extract/lycoris_embed.json | 11 ++++
tests/fixtures/extract/mp4upload_embed.html | 25 +++++++++
tests/fixtures/extract/playmate_stream.json | 3 ++
tests/fixtures/extract/streamtape_embed.html | 18 +++++++
tests/fixtures/extract/uqload_dl_packed.html | 16 ++++++
tests/fixtures/extract/vidara_stream.json | 7 +++
21 files changed, 404 insertions(+)
create mode 100644 tests/extract/conftest.py
create mode 100644 tests/extract/test_fixture_files.py
create mode 100644 tests/fixtures/extract/dood_embed.html
create mode 100644 tests/fixtures/extract/dood_pass_md5.txt
create mode 100644 tests/fixtures/extract/flyf_assign.json
create mode 100644 tests/fixtures/extract/jwplayer_morencius_packed.html
create mode 100644 tests/fixtures/extract/lycoris_embed.json
create mode 100644 tests/fixtures/extract/mp4upload_embed.html
create mode 100644 tests/fixtures/extract/playmate_stream.json
create mode 100644 tests/fixtures/extract/streamtape_embed.html
create mode 100644 tests/fixtures/extract/uqload_dl_packed.html
create mode 100644 tests/fixtures/extract/vidara_stream.json
diff --git a/tests/extract/conftest.py b/tests/extract/conftest.py
new file mode 100644
index 0000000..eb670a7
--- /dev/null
+++ b/tests/extract/conftest.py
@@ -0,0 +1,54 @@
+import json
+import socket
+from pathlib import Path
+from unittest.mock import MagicMock
+
+import pytest
+from curl_cffi.curl import Curl
+
+FIXTURES_DIR = Path(__file__).parent.parent / "fixtures" / "extract"
+
+
+@pytest.fixture(autouse=True)
+def _no_outbound_network(monkeypatch):
+ """Fail fast if a mis-targeted Session patch would let a test reach libcurl or a raw socket."""
+
+ def _blocked(*args, **kwargs):
+ raise AssertionError("outbound network attempted in tests/extract")
+
+ monkeypatch.setattr(Curl, "perform", _blocked)
+ monkeypatch.setattr(socket.socket, "connect", _blocked)
+
+
+@pytest.fixture
+def load_fixture():
+ def _load(name: str):
+ text = (FIXTURES_DIR / name).read_text(encoding="utf-8")
+ return json.loads(text) if name.endswith(".json") else text
+
+ return _load
+
+
+def _fake_response(body):
+ resp = MagicMock(status_code=200, **{"raise_for_status.return_value": None})
+ if isinstance(body, str):
+ resp.text = body
+ else:
+ resp.json.return_value = body
+ return resp
+
+
+@pytest.fixture
+def fake_session(monkeypatch):
+ """Serve bodies in call order: ``str`` as ``.text``, anything else as ``.json()``."""
+
+ def _install(module, *, get=(), post=()):
+ session = MagicMock()
+ session.get.side_effect = [_fake_response(b) for b in get]
+ session.post.side_effect = [_fake_response(b) for b in post]
+ session.__enter__.return_value = session
+ session.__exit__.return_value = False
+ monkeypatch.setattr(module.cffi_requests, "Session", MagicMock(return_value=session))
+ return session
+
+ return _install
diff --git a/tests/extract/test_dood.py b/tests/extract/test_dood.py
index 93e660b..5c3c998 100644
--- a/tests/extract/test_dood.py
+++ b/tests/extract/test_dood.py
@@ -5,6 +5,7 @@
import pytest
+from alt_ani_cli.extract import dood
from alt_ani_cli.extract.dood import resolve
_EMBED = "https://dood.la/e/xyz"
@@ -97,3 +98,25 @@ def test_bad_base_url_raises_before_http(self):
resolve("not-a-url", _REFERER)
session.get.assert_not_called()
assert exc_info.value.category == "parser_drift"
+
+
+@pytest.mark.unit
+class TestResolveDoodFixture:
+ _EMBED = "https://dood.example/e/testfilecode"
+
+ def test_embed_and_pass_md5_fixtures_resolve_to_tokenized_url(self, fake_session, load_fixture):
+ session = fake_session(dood, get=[load_fixture("dood_embed.html"), load_fixture("dood_pass_md5.txt")])
+ with patch("alt_ani_cli.extract.dood.time.time", return_value=1700000000.0):
+ stream = resolve(self._EMBED, _REFERER)
+
+ assert stream.url == "https://cdn.example/stream/testfilecode~?token=testtokenAbc123&expiry=1700000000000"
+ assert stream.ext == "mp4"
+ assert stream.headers["Referer"] == "https://dood.example/"
+ assert "User-Agent" in stream.headers
+
+ embed_call, pass_call = session.get.call_args_list
+ assert embed_call.args == (self._EMBED,)
+ assert embed_call.kwargs["headers"]["Referer"] == _REFERER
+ assert pass_call.args == ("https://dood.example/pass_md5/42-1700000000-testhash/testfilecode",)
+ assert pass_call.kwargs["headers"]["Referer"] == self._EMBED
+ assert pass_call.kwargs["headers"]["X-Requested-With"] == "XMLHttpRequest"
diff --git a/tests/extract/test_fixture_files.py b/tests/extract/test_fixture_files.py
new file mode 100644
index 0000000..b7307e0
--- /dev/null
+++ b/tests/extract/test_fixture_files.py
@@ -0,0 +1,34 @@
+"""Invariants for the host response fixtures in tests/fixtures/extract/."""
+
+import json
+import re
+from pathlib import Path
+
+import pytest
+
+FIXTURES_DIR = Path(__file__).parent.parent / "fixtures" / "extract"
+
+# Scheme-ful URLs (JSON-escaped slashes included) and protocol-relative string literals.
+_URL_HOST_RE = re.compile(r"https?:\\?/\\?/([^/\\\"'\s:?]+)|[\"']//([^/\"'\s:?]+)")
+
+
+def _fixture_files():
+ return sorted(p for p in FIXTURES_DIR.iterdir() if p.is_file())
+
+
+@pytest.mark.unit
+class TestExtractFixtureFiles:
+ def test_directory_exists_and_is_not_empty(self):
+ assert FIXTURES_DIR.is_dir()
+ assert _fixture_files()
+
+ @pytest.mark.parametrize("path", _fixture_files(), ids=lambda p: p.name)
+ def test_fixture_is_utf8_and_json_fixtures_parse(self, path):
+ text = path.read_bytes().decode("utf-8")
+ if path.suffix == ".json":
+ json.loads(text)
+
+ @pytest.mark.parametrize("path", _fixture_files(), ids=lambda p: p.name)
+ def test_fixture_urls_point_only_at_reserved_example_hosts(self, path):
+ hosts = {a or b for a, b in _URL_HOST_RE.findall(path.read_text(encoding="utf-8"))}
+ assert all(h.endswith(".example") for h in hosts), hosts
diff --git a/tests/extract/test_flyf.py b/tests/extract/test_flyf.py
index 1cabb1a..f4dbb95 100644
--- a/tests/extract/test_flyf.py
+++ b/tests/extract/test_flyf.py
@@ -5,6 +5,7 @@
import pytest
+from alt_ani_cli.extract import flyf
from alt_ani_cli.extract.flyf import resolve
_EMBED = "https://flyf.lat/embed/Gdbv8BEkZhOFZ2O"
@@ -83,3 +84,25 @@ def test_bad_embed_url_raises_before_http(self):
resolve("not-a-url", _REFERER)
session.get.assert_not_called()
assert exc_info.value.category == "parser_drift"
+
+
+@pytest.mark.unit
+class TestResolveFlyfFixture:
+ _EMBED = "https://flyf.example/embed/testfilecode"
+
+ def test_assign_fixture_resolves_to_raw_file(self, fake_session, load_fixture):
+ session = fake_session(flyf, get=[load_fixture("flyf_assign.json")])
+ stream = resolve(self._EMBED, _REFERER)
+
+ assert stream.url == "https://cdn.example/raw/testtoken0000"
+ assert stream.ext == "mp4"
+ assert stream.headers["Referer"] == self._EMBED
+ assert "User-Agent" in stream.headers
+
+ call = session.get.call_args
+ assert call.args == ("https://api.flyfile.app/api/streaming/assign/testfilecode",)
+ headers = call.kwargs["headers"]
+ assert headers["Referer"] == self._EMBED
+ assert headers["X-FlyFile-View"] == "embed"
+ assert headers["X-Embed-Referrer"] == self._EMBED
+ assert headers["X-Adblock-Detected"] == "0"
diff --git a/tests/extract/test_jwplayer.py b/tests/extract/test_jwplayer.py
index 34de9bd..af50a67 100644
--- a/tests/extract/test_jwplayer.py
+++ b/tests/extract/test_jwplayer.py
@@ -5,6 +5,7 @@
import pytest
+from alt_ani_cli.extract import jwplayer
from alt_ani_cli.extract.jwplayer import _best_hls_url, _decode_base_n, _normalize_stream_url, resolve, unpack_packer
_SIMPLE_PACKED = (
@@ -171,3 +172,23 @@ def test_no_video_url_raises_value_error(self):
with pytest.raises(ValueError, match="jwplayer") as exc_info:
resolve(_EMBED, _REFERER)
assert exc_info.value.category == "parser_drift"
+
+
+@pytest.mark.unit
+class TestResolveJwplayerFixture:
+ def test_morencius_packed_page_resolves_hls4_relative_path(self, fake_session, load_fixture):
+ html = load_fixture("jwplayer_morencius_packed.html")
+ # Words live only in the packer key table, so the stream URL can only come out of the real unpacker.
+ assert '"hls4"' not in html and "master.m3u8" not in html
+ session = fake_session(jwplayer, get=[html])
+
+ stream = resolve(_EMBED, _REFERER)
+
+ assert stream.url == "https://morencius.com/stream/testfilecode/testtoken/1700000000/master.m3u8"
+ assert stream.ext == "m3u8"
+ assert stream.headers["Referer"] == _EMBED
+ assert "User-Agent" in stream.headers
+
+ call = session.get.call_args
+ assert call.args == (_EMBED,)
+ assert call.kwargs["headers"]["Referer"] == _REFERER
diff --git a/tests/extract/test_lycoris.py b/tests/extract/test_lycoris.py
index 7868db8..9161b63 100644
--- a/tests/extract/test_lycoris.py
+++ b/tests/extract/test_lycoris.py
@@ -5,6 +5,7 @@
import pytest
+from alt_ani_cli.extract import lycoris
from alt_ani_cli.extract.lycoris import resolve
_EMBED = "https://www.lycoris.cafe/embed?id=180136&episode=1"
@@ -98,3 +99,28 @@ def test_bad_embed_url_raises_before_http(self):
resolve("not-a-url", _REFERER)
session.get.assert_not_called()
assert exc_info.value.category == "parser_drift"
+
+
+@pytest.mark.unit
+class TestResolveLycorisFixture:
+ _EMBED = "https://lycoris.example/embed?id=4242&episode=1"
+
+ def test_embed_api_fixture_resolves_best_url_and_qualities(self, fake_session, load_fixture):
+ session = fake_session(lycoris, get=[load_fixture("lycoris_embed.json")])
+ stream = resolve(self._EMBED, _REFERER)
+
+ base = "https://cdn.example/d/testfilecode"
+ assert stream.url == f"{base}/episode-1-1080p.mp4"
+ assert stream.ext == "mp4"
+ assert stream.qualities == {
+ "1080p": f"{base}/episode-1-1080p.mp4",
+ "720p": f"{base}/episode-1-720p.mp4",
+ "480p": f"{base}/episode-1-480p.mp4",
+ "source-mkv": f"{base}/episode-1-1080p.mkv",
+ }
+ assert stream.headers["Referer"] == self._EMBED
+
+ call = session.get.call_args
+ assert call.args == ("https://lycoris.example/api/embed",)
+ assert call.kwargs["params"] == {"id": "4242", "episode": "1"}
+ assert call.kwargs["headers"]["Referer"] == self._EMBED
diff --git a/tests/extract/test_mp4upload.py b/tests/extract/test_mp4upload.py
index e8ec4a7..45552f1 100644
--- a/tests/extract/test_mp4upload.py
+++ b/tests/extract/test_mp4upload.py
@@ -6,6 +6,7 @@
import pytest
+from alt_ani_cli.extract import mp4upload
from alt_ani_cli.extract.mp4upload import resolve
_EMBED = "https://www.mp4upload.com/embed-abc123.html"
@@ -67,3 +68,19 @@ def test_referer_in_stream_headers(self):
with _make_session_patch(""):
stream = resolve(_EMBED, _REFERER)
assert stream.headers.get("Referer") == _EMBED
+
+
+@pytest.mark.unit
+class TestResolveMp4uploadFixture:
+ def test_videojs_object_src_fixture_resolves(self, fake_session, load_fixture):
+ session = fake_session(mp4upload, get=[load_fixture("mp4upload_embed.html")])
+ stream = resolve(_EMBED, _REFERER)
+
+ assert stream.url == "https://cdn.example:183/d/testfilecodetesttoken/video.mp4"
+ assert stream.ext == "mp4"
+ assert stream.headers["Referer"] == _EMBED
+ assert "User-Agent" in stream.headers
+
+ call = session.get.call_args
+ assert call.args == (_EMBED,)
+ assert call.kwargs["headers"]["Referer"] == _REFERER
diff --git a/tests/extract/test_playmate.py b/tests/extract/test_playmate.py
index 6b1b9ed..d75bf3b 100644
--- a/tests/extract/test_playmate.py
+++ b/tests/extract/test_playmate.py
@@ -5,6 +5,7 @@
import pytest
+from alt_ani_cli.extract import playmate
from alt_ani_cli.extract.playmate import resolve
_EMBED = "https://playmate.to/embed/MFyuwmxvBGiUE"
@@ -76,3 +77,25 @@ def test_mp4_sx_gets_mp4_ext(self):
with _make_session_patch({"sx": "https://cdn.example.com/video.mp4"}):
stream = resolve(_EMBED, _REFERER)
assert stream.ext == "mp4"
+
+
+@pytest.mark.unit
+class TestResolvePlaymateFixture:
+ _EMBED = "https://playmate.example/embed/testfilecode"
+
+ def test_api_fixture_master_txt_is_recognised_as_hls(self, fake_session, load_fixture):
+ session = fake_session(playmate, post=[load_fixture("playmate_stream.json")])
+ stream = resolve(self._EMBED, _REFERER)
+
+ assert stream.url == "https://cdn.example/hls/testfilecode/master.txt?t=testtoken&e=1700000000"
+ assert stream.ext == "m3u8"
+ assert stream.headers["Referer"] == self._EMBED
+ assert stream.headers["Origin"] == "https://playmate.example"
+
+ call = session.post.call_args
+ assert call.args == ("https://playmate.example/api/s",)
+ assert call.kwargs["json"] == {"c": "testfilecode", "d": "web"}
+ headers = call.kwargs["headers"]
+ assert headers["Origin"] == "https://playmate.example"
+ assert headers["Referer"] == self._EMBED
+ assert headers["Sec-Fetch-Site"] == "same-origin"
diff --git a/tests/extract/test_streamtape.py b/tests/extract/test_streamtape.py
index 973724d..b61426e 100644
--- a/tests/extract/test_streamtape.py
+++ b/tests/extract/test_streamtape.py
@@ -5,6 +5,7 @@
import pytest
+from alt_ani_cli.extract import streamtape
from alt_ani_cli.extract.streamtape import resolve
_EMBED = "https://streamtape.com/e/abc123"
@@ -78,3 +79,20 @@ def test_no_match_raises_value_error(self):
with pytest.raises(ValueError, match="streamtape") as exc_info:
resolve(_EMBED, _REFERER)
assert exc_info.value.category == "parser_drift"
+
+
+@pytest.mark.unit
+class TestResolveStreamtapeFixture:
+ def test_robotlink_concatenation_fixture_ignores_decoy_div(self, fake_session, load_fixture):
+ session = fake_session(streamtape, get=[load_fixture("streamtape_embed.html")])
+ stream = resolve(_EMBED, _REFERER)
+
+ assert stream.url == (
+ "https://streamtape.example/get_video?id=testfilecode&expires=1700000000&ip=TESTIP0000&token=testtoken0000"
+ )
+ assert stream.ext == "mp4"
+ assert stream.headers["Referer"] == _EMBED
+
+ call = session.get.call_args
+ assert call.args == (_EMBED,)
+ assert call.kwargs["headers"]["Referer"] == _REFERER
diff --git a/tests/extract/test_uqload.py b/tests/extract/test_uqload.py
index bec13e1..c257914 100644
--- a/tests/extract/test_uqload.py
+++ b/tests/extract/test_uqload.py
@@ -5,6 +5,7 @@
import pytest
+from alt_ani_cli.extract import uqload
from alt_ani_cli.extract.uqload import resolve
_EMBED = "https://uqload.is/e/mhq7xbccftp5"
@@ -88,3 +89,21 @@ def test_m3u8_url_gets_m3u8_ext(self):
with _make_session_patch(html):
stream = resolve(_EMBED, _REFERER)
assert stream.ext == "m3u8"
+
+
+@pytest.mark.unit
+class TestResolveUqloadFixture:
+ def test_packed_dl_fixture_resolves_after_unpacking(self, fake_session, load_fixture):
+ body = load_fixture("uqload_dl_packed.html")
+ assert "v.mp4" not in body
+ session = fake_session(uqload, post=[body])
+ stream = resolve(_EMBED, _REFERER)
+
+ assert stream.url == "https://cdn.example/3rfktestfilecode/v.mp4"
+ assert stream.ext == "mp4"
+ assert stream.headers["Referer"] == _EMBED
+
+ call = session.post.call_args
+ assert call.args == ("https://uqload.is/dl",)
+ assert call.kwargs["data"] == {"op": "embed", "file_code": "mhq7xbccftp5", "auto": "0", "referer": _EMBED}
+ assert call.kwargs["headers"]["Referer"] == _EMBED
diff --git a/tests/extract/test_vidara.py b/tests/extract/test_vidara.py
index b862dc0..67c7114 100644
--- a/tests/extract/test_vidara.py
+++ b/tests/extract/test_vidara.py
@@ -5,6 +5,7 @@
import pytest
+from alt_ani_cli.extract import vidara
from alt_ani_cli.extract.vidara import resolve
_EMBED = "https://vidara.to/e/gGnwW3ekLDWQX"
@@ -90,3 +91,23 @@ def test_works_with_alternate_host_viewdara(self):
call = session.post.call_args
assert call.args == ("https://viewdara.com/api/stream",)
assert call.kwargs["headers"]["Origin"] == "https://viewdara.com"
+
+
+@pytest.mark.unit
+class TestResolveVidaraFixture:
+ @pytest.mark.parametrize("host", ["vidara.to", "iosbgaigo.com"])
+ def test_stream_api_fixture_is_host_agnostic(self, fake_session, load_fixture, host):
+ embed = f"https://{host}/e/testfilecode"
+ session = fake_session(vidara, post=[load_fixture("vidara_stream.json")])
+ stream = resolve(embed, _REFERER)
+
+ assert stream.url == "https://cdn.example/hls/testfilecode/master.m3u8?t=testtoken&e=1700000000"
+ assert stream.ext == "m3u8"
+ assert stream.headers["Referer"] == embed
+ assert stream.headers["Origin"] == f"https://{host}"
+
+ call = session.post.call_args
+ assert call.args == (f"https://{host}/api/stream",)
+ assert call.kwargs["json"] == {"filecode": "testfilecode", "device": "web"}
+ assert call.kwargs["headers"]["Origin"] == f"https://{host}"
+ assert call.kwargs["headers"]["Referer"] == embed
diff --git a/tests/fixtures/extract/dood_embed.html b/tests/fixtures/extract/dood_embed.html
new file mode 100644
index 0000000..a7570f7
--- /dev/null
+++ b/tests/fixtures/extract/dood_embed.html
@@ -0,0 +1,23 @@
+
+
+
+
+
+testfilecode.mp4 - DoodStream
+
+
+
+
+
+
+
+
+
+
+
diff --git a/tests/fixtures/extract/dood_pass_md5.txt b/tests/fixtures/extract/dood_pass_md5.txt
new file mode 100644
index 0000000..b0bc9c8
--- /dev/null
+++ b/tests/fixtures/extract/dood_pass_md5.txt
@@ -0,0 +1 @@
+https://cdn.example/stream/testfilecode~
diff --git a/tests/fixtures/extract/flyf_assign.json b/tests/fixtures/extract/flyf_assign.json
new file mode 100644
index 0000000..5e650e6
--- /dev/null
+++ b/tests/fixtures/extract/flyf_assign.json
@@ -0,0 +1,4 @@
+{
+ "url": "https://cdn.example",
+ "token": "testtoken0000"
+}
diff --git a/tests/fixtures/extract/jwplayer_morencius_packed.html b/tests/fixtures/extract/jwplayer_morencius_packed.html
new file mode 100644
index 0000000..e6f7f71
--- /dev/null
+++ b/tests/fixtures/extract/jwplayer_morencius_packed.html
@@ -0,0 +1,17 @@
+
+
+
+
+
+Watch testfilecode
+
+
+
+
+
+
+
+
+
diff --git a/tests/fixtures/extract/lycoris_embed.json b/tests/fixtures/extract/lycoris_embed.json
new file mode 100644
index 0000000..4b0e900
--- /dev/null
+++ b/tests/fixtures/extract/lycoris_embed.json
@@ -0,0 +1,11 @@
+{
+ "episodeInfo": {
+ "primarySource": {
+ "FHD": "https://cdn.example/d/testfilecode/episode-1-1080p.mp4",
+ "HD": "https://cdn.example/d/testfilecode/episode-1-720p.mp4",
+ "SD": "https://cdn.example/d/testfilecode/episode-1-480p.mp4",
+ "preview": "https://cdn.example/d/testfilecode/preview.mp4",
+ "SourceMKV": "https://cdn.example/d/testfilecode/episode-1-1080p.mkv"
+ }
+ }
+}
diff --git a/tests/fixtures/extract/mp4upload_embed.html b/tests/fixtures/extract/mp4upload_embed.html
new file mode 100644
index 0000000..e403dfd
--- /dev/null
+++ b/tests/fixtures/extract/mp4upload_embed.html
@@ -0,0 +1,25 @@
+
+
+
+
+testfilecode.mp4
+
+
+
+
+
+
+
+
diff --git a/tests/fixtures/extract/playmate_stream.json b/tests/fixtures/extract/playmate_stream.json
new file mode 100644
index 0000000..f92bcfb
--- /dev/null
+++ b/tests/fixtures/extract/playmate_stream.json
@@ -0,0 +1,3 @@
+{
+ "sx": "https://cdn.example/hls/testfilecode/master.txt?t=testtoken&e=1700000000"
+}
diff --git a/tests/fixtures/extract/streamtape_embed.html b/tests/fixtures/extract/streamtape_embed.html
new file mode 100644
index 0000000..816a555
--- /dev/null
+++ b/tests/fixtures/extract/streamtape_embed.html
@@ -0,0 +1,18 @@
+
+
+
+
+testfilecode.mp4
+
+
+
+
+
+
+
+/streamtape.example/get_video?id=testfilecode&expires=1600000000&ip=DECOYIP&token=decoytoken
+
+
+
diff --git a/tests/fixtures/extract/uqload_dl_packed.html b/tests/fixtures/extract/uqload_dl_packed.html
new file mode 100644
index 0000000..13a335a
--- /dev/null
+++ b/tests/fixtures/extract/uqload_dl_packed.html
@@ -0,0 +1,16 @@
+
+
+
+
+testfilecode.mp4
+
+
+
+
+
+
+
+
+
diff --git a/tests/fixtures/extract/vidara_stream.json b/tests/fixtures/extract/vidara_stream.json
new file mode 100644
index 0000000..418a643
--- /dev/null
+++ b/tests/fixtures/extract/vidara_stream.json
@@ -0,0 +1,7 @@
+{
+ "default_sub_lang": "Polish",
+ "filecode": "testfilecode",
+ "streaming_url": "https://cdn.example/hls/testfilecode/master.m3u8?t=testtoken&e=1700000000",
+ "subtitles": null,
+ "title": ""
+}
