test(extract): add fixture-based regression tests for custom extractors

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

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"

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

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"

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

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

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("<script>player.src('https://cdn.mp4upload.com/v/video.mp4');</script>"):
             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

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"

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

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

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