Skip to content

Commit e339c3b

Browse files
test(extract): add fixture-based regression tests for custom extractors (#27)
1 parent 5cfce38 commit e339c3b

21 files changed

Lines changed: 404 additions & 0 deletions

‎tests/extract/conftest.py‎

Lines changed: 54 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,54 @@
1+
import json
2+
import socket
3+
from pathlib import Path
4+
from unittest.mock import MagicMock
5+
6+
import pytest
7+
from curl_cffi.curl import Curl
8+
9+
FIXTURES_DIR = Path(__file__).parent.parent / "fixtures" / "extract"
10+
11+
12+
@pytest.fixture(autouse=True)
13+
def _no_outbound_network(monkeypatch):
14+
"""Fail fast if a mis-targeted Session patch would let a test reach libcurl or a raw socket."""
15+
16+
def _blocked(*args, **kwargs):
17+
raise AssertionError("outbound network attempted in tests/extract")
18+
19+
monkeypatch.setattr(Curl, "perform", _blocked)
20+
monkeypatch.setattr(socket.socket, "connect", _blocked)
21+
22+
23+
@pytest.fixture
24+
def load_fixture():
25+
def _load(name: str):
26+
text = (FIXTURES_DIR / name).read_text(encoding="utf-8")
27+
return json.loads(text) if name.endswith(".json") else text
28+
29+
return _load
30+
31+
32+
def _fake_response(body):
33+
resp = MagicMock(status_code=200, **{"raise_for_status.return_value": None})
34+
if isinstance(body, str):
35+
resp.text = body
36+
else:
37+
resp.json.return_value = body
38+
return resp
39+
40+
41+
@pytest.fixture
42+
def fake_session(monkeypatch):
43+
"""Serve bodies in call order: ``str`` as ``.text``, anything else as ``.json()``."""
44+
45+
def _install(module, *, get=(), post=()):
46+
session = MagicMock()
47+
session.get.side_effect = [_fake_response(b) for b in get]
48+
session.post.side_effect = [_fake_response(b) for b in post]
49+
session.__enter__.return_value = session
50+
session.__exit__.return_value = False
51+
monkeypatch.setattr(module.cffi_requests, "Session", MagicMock(return_value=session))
52+
return session
53+
54+
return _install

‎tests/extract/test_dood.py‎

Lines changed: 23 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -5,6 +5,7 @@
55

66
import pytest
77

8+
from alt_ani_cli.extract import dood
89
from alt_ani_cli.extract.dood import resolve
910

1011
_EMBED = "https://dood.la/e/xyz"
@@ -97,3 +98,25 @@ def test_bad_base_url_raises_before_http(self):
9798
resolve("not-a-url", _REFERER)
9899
session.get.assert_not_called()
99100
assert exc_info.value.category == "parser_drift"
101+
102+
103+
@pytest.mark.unit
104+
class TestResolveDoodFixture:
105+
_EMBED = "https://dood.example/e/testfilecode"
106+
107+
def test_embed_and_pass_md5_fixtures_resolve_to_tokenized_url(self, fake_session, load_fixture):
108+
session = fake_session(dood, get=[load_fixture("dood_embed.html"), load_fixture("dood_pass_md5.txt")])
109+
with patch("alt_ani_cli.extract.dood.time.time", return_value=1700000000.0):
110+
stream = resolve(self._EMBED, _REFERER)
111+
112+
assert stream.url == "https://cdn.example/stream/testfilecode~?token=testtokenAbc123&expiry=1700000000000"
113+
assert stream.ext == "mp4"
114+
assert stream.headers["Referer"] == "https://dood.example/"
115+
assert "User-Agent" in stream.headers
116+
117+
embed_call, pass_call = session.get.call_args_list
118+
assert embed_call.args == (self._EMBED,)
119+
assert embed_call.kwargs["headers"]["Referer"] == _REFERER
120+
assert pass_call.args == ("https://dood.example/pass_md5/42-1700000000-testhash/testfilecode",)
121+
assert pass_call.kwargs["headers"]["Referer"] == self._EMBED
122+
assert pass_call.kwargs["headers"]["X-Requested-With"] == "XMLHttpRequest"
Lines changed: 34 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,34 @@
1+
"""Invariants for the host response fixtures in tests/fixtures/extract/."""
2+
3+
import json
4+
import re
5+
from pathlib import Path
6+
7+
import pytest
8+
9+
FIXTURES_DIR = Path(__file__).parent.parent / "fixtures" / "extract"
10+
11+
# Scheme-ful URLs (JSON-escaped slashes included) and protocol-relative string literals.
12+
_URL_HOST_RE = re.compile(r"https?:\\?/\\?/([^/\\\"'\s:?]+)|[\"']//([^/\"'\s:?]+)")
13+
14+
15+
def _fixture_files():
16+
return sorted(p for p in FIXTURES_DIR.iterdir() if p.is_file())
17+
18+
19+
@pytest.mark.unit
20+
class TestExtractFixtureFiles:
21+
def test_directory_exists_and_is_not_empty(self):
22+
assert FIXTURES_DIR.is_dir()
23+
assert _fixture_files()
24+
25+
@pytest.mark.parametrize("path", _fixture_files(), ids=lambda p: p.name)
26+
def test_fixture_is_utf8_and_json_fixtures_parse(self, path):
27+
text = path.read_bytes().decode("utf-8")
28+
if path.suffix == ".json":
29+
json.loads(text)
30+
31+
@pytest.mark.parametrize("path", _fixture_files(), ids=lambda p: p.name)
32+
def test_fixture_urls_point_only_at_reserved_example_hosts(self, path):
33+
hosts = {a or b for a, b in _URL_HOST_RE.findall(path.read_text(encoding="utf-8"))}
34+
assert all(h.endswith(".example") for h in hosts), hosts

‎tests/extract/test_flyf.py‎

Lines changed: 23 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -5,6 +5,7 @@
55

66
import pytest
77

8+
from alt_ani_cli.extract import flyf
89
from alt_ani_cli.extract.flyf import resolve
910

1011
_EMBED = "https://flyf.lat/embed/Gdbv8BEkZhOFZ2O"
@@ -83,3 +84,25 @@ def test_bad_embed_url_raises_before_http(self):
8384
resolve("not-a-url", _REFERER)
8485
session.get.assert_not_called()
8586
assert exc_info.value.category == "parser_drift"
87+
88+
89+
@pytest.mark.unit
90+
class TestResolveFlyfFixture:
91+
_EMBED = "https://flyf.example/embed/testfilecode"
92+
93+
def test_assign_fixture_resolves_to_raw_file(self, fake_session, load_fixture):
94+
session = fake_session(flyf, get=[load_fixture("flyf_assign.json")])
95+
stream = resolve(self._EMBED, _REFERER)
96+
97+
assert stream.url == "https://cdn.example/raw/testtoken0000"
98+
assert stream.ext == "mp4"
99+
assert stream.headers["Referer"] == self._EMBED
100+
assert "User-Agent" in stream.headers
101+
102+
call = session.get.call_args
103+
assert call.args == ("https://api.flyfile.app/api/streaming/assign/testfilecode",)
104+
headers = call.kwargs["headers"]
105+
assert headers["Referer"] == self._EMBED
106+
assert headers["X-FlyFile-View"] == "embed"
107+
assert headers["X-Embed-Referrer"] == self._EMBED
108+
assert headers["X-Adblock-Detected"] == "0"

‎tests/extract/test_jwplayer.py‎

Lines changed: 21 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -5,6 +5,7 @@
55

66
import pytest
77

8+
from alt_ani_cli.extract import jwplayer
89
from alt_ani_cli.extract.jwplayer import _best_hls_url, _decode_base_n, _normalize_stream_url, resolve, unpack_packer
910

1011
_SIMPLE_PACKED = (
@@ -171,3 +172,23 @@ def test_no_video_url_raises_value_error(self):
171172
with pytest.raises(ValueError, match="jwplayer") as exc_info:
172173
resolve(_EMBED, _REFERER)
173174
assert exc_info.value.category == "parser_drift"
175+
176+
177+
@pytest.mark.unit
178+
class TestResolveJwplayerFixture:
179+
def test_morencius_packed_page_resolves_hls4_relative_path(self, fake_session, load_fixture):
180+
html = load_fixture("jwplayer_morencius_packed.html")
181+
# Words live only in the packer key table, so the stream URL can only come out of the real unpacker.
182+
assert '"hls4"' not in html and "master.m3u8" not in html
183+
session = fake_session(jwplayer, get=[html])
184+
185+
stream = resolve(_EMBED, _REFERER)
186+
187+
assert stream.url == "https://morencius.com/stream/testfilecode/testtoken/1700000000/master.m3u8"
188+
assert stream.ext == "m3u8"
189+
assert stream.headers["Referer"] == _EMBED
190+
assert "User-Agent" in stream.headers
191+
192+
call = session.get.call_args
193+
assert call.args == (_EMBED,)
194+
assert call.kwargs["headers"]["Referer"] == _REFERER

‎tests/extract/test_lycoris.py‎

Lines changed: 26 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -5,6 +5,7 @@
55

66
import pytest
77

8+
from alt_ani_cli.extract import lycoris
89
from alt_ani_cli.extract.lycoris import resolve
910

1011
_EMBED = "https://www.lycoris.cafe/embed?id=180136&episode=1"
@@ -98,3 +99,28 @@ def test_bad_embed_url_raises_before_http(self):
9899
resolve("not-a-url", _REFERER)
99100
session.get.assert_not_called()
100101
assert exc_info.value.category == "parser_drift"
102+
103+
104+
@pytest.mark.unit
105+
class TestResolveLycorisFixture:
106+
_EMBED = "https://lycoris.example/embed?id=4242&episode=1"
107+
108+
def test_embed_api_fixture_resolves_best_url_and_qualities(self, fake_session, load_fixture):
109+
session = fake_session(lycoris, get=[load_fixture("lycoris_embed.json")])
110+
stream = resolve(self._EMBED, _REFERER)
111+
112+
base = "https://cdn.example/d/testfilecode"
113+
assert stream.url == f"{base}/episode-1-1080p.mp4"
114+
assert stream.ext == "mp4"
115+
assert stream.qualities == {
116+
"1080p": f"{base}/episode-1-1080p.mp4",
117+
"720p": f"{base}/episode-1-720p.mp4",
118+
"480p": f"{base}/episode-1-480p.mp4",
119+
"source-mkv": f"{base}/episode-1-1080p.mkv",
120+
}
121+
assert stream.headers["Referer"] == self._EMBED
122+
123+
call = session.get.call_args
124+
assert call.args == ("https://lycoris.example/api/embed",)
125+
assert call.kwargs["params"] == {"id": "4242", "episode": "1"}
126+
assert call.kwargs["headers"]["Referer"] == self._EMBED

‎tests/extract/test_mp4upload.py‎

Lines changed: 17 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,7 @@
66

77
import pytest
88

9+
from alt_ani_cli.extract import mp4upload
910
from alt_ani_cli.extract.mp4upload import resolve
1011

1112
_EMBED = "https://www.mp4upload.com/embed-abc123.html"
@@ -67,3 +68,19 @@ def test_referer_in_stream_headers(self):
6768
with _make_session_patch("<script>player.src('https://cdn.mp4upload.com/v/video.mp4');</script>"):
6869
stream = resolve(_EMBED, _REFERER)
6970
assert stream.headers.get("Referer") == _EMBED
71+
72+
73+
@pytest.mark.unit
74+
class TestResolveMp4uploadFixture:
75+
def test_videojs_object_src_fixture_resolves(self, fake_session, load_fixture):
76+
session = fake_session(mp4upload, get=[load_fixture("mp4upload_embed.html")])
77+
stream = resolve(_EMBED, _REFERER)
78+
79+
assert stream.url == "https://cdn.example:183/d/testfilecodetesttoken/video.mp4"
80+
assert stream.ext == "mp4"
81+
assert stream.headers["Referer"] == _EMBED
82+
assert "User-Agent" in stream.headers
83+
84+
call = session.get.call_args
85+
assert call.args == (_EMBED,)
86+
assert call.kwargs["headers"]["Referer"] == _REFERER

‎tests/extract/test_playmate.py‎

Lines changed: 23 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -5,6 +5,7 @@
55

66
import pytest
77

8+
from alt_ani_cli.extract import playmate
89
from alt_ani_cli.extract.playmate import resolve
910

1011
_EMBED = "https://playmate.to/embed/MFyuwmxvBGiUE"
@@ -76,3 +77,25 @@ def test_mp4_sx_gets_mp4_ext(self):
7677
with _make_session_patch({"sx": "https://cdn.example.com/video.mp4"}):
7778
stream = resolve(_EMBED, _REFERER)
7879
assert stream.ext == "mp4"
80+
81+
82+
@pytest.mark.unit
83+
class TestResolvePlaymateFixture:
84+
_EMBED = "https://playmate.example/embed/testfilecode"
85+
86+
def test_api_fixture_master_txt_is_recognised_as_hls(self, fake_session, load_fixture):
87+
session = fake_session(playmate, post=[load_fixture("playmate_stream.json")])
88+
stream = resolve(self._EMBED, _REFERER)
89+
90+
assert stream.url == "https://cdn.example/hls/testfilecode/master.txt?t=testtoken&e=1700000000"
91+
assert stream.ext == "m3u8"
92+
assert stream.headers["Referer"] == self._EMBED
93+
assert stream.headers["Origin"] == "https://playmate.example"
94+
95+
call = session.post.call_args
96+
assert call.args == ("https://playmate.example/api/s",)
97+
assert call.kwargs["json"] == {"c": "testfilecode", "d": "web"}
98+
headers = call.kwargs["headers"]
99+
assert headers["Origin"] == "https://playmate.example"
100+
assert headers["Referer"] == self._EMBED
101+
assert headers["Sec-Fetch-Site"] == "same-origin"

‎tests/extract/test_streamtape.py‎

Lines changed: 18 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -5,6 +5,7 @@
55

66
import pytest
77

8+
from alt_ani_cli.extract import streamtape
89
from alt_ani_cli.extract.streamtape import resolve
910

1011
_EMBED = "https://streamtape.com/e/abc123"
@@ -78,3 +79,20 @@ def test_no_match_raises_value_error(self):
7879
with pytest.raises(ValueError, match="streamtape") as exc_info:
7980
resolve(_EMBED, _REFERER)
8081
assert exc_info.value.category == "parser_drift"
82+
83+
84+
@pytest.mark.unit
85+
class TestResolveStreamtapeFixture:
86+
def test_robotlink_concatenation_fixture_ignores_decoy_div(self, fake_session, load_fixture):
87+
session = fake_session(streamtape, get=[load_fixture("streamtape_embed.html")])
88+
stream = resolve(_EMBED, _REFERER)
89+
90+
assert stream.url == (
91+
"https://streamtape.example/get_video?id=testfilecode&expires=1700000000&ip=TESTIP0000&token=testtoken0000"
92+
)
93+
assert stream.ext == "mp4"
94+
assert stream.headers["Referer"] == _EMBED
95+
96+
call = session.get.call_args
97+
assert call.args == (_EMBED,)
98+
assert call.kwargs["headers"]["Referer"] == _REFERER

‎tests/extract/test_uqload.py‎

Lines changed: 19 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -5,6 +5,7 @@
55

66
import pytest
77

8+
from alt_ani_cli.extract import uqload
89
from alt_ani_cli.extract.uqload import resolve
910

1011
_EMBED = "https://uqload.is/e/mhq7xbccftp5"
@@ -88,3 +89,21 @@ def test_m3u8_url_gets_m3u8_ext(self):
8889
with _make_session_patch(html):
8990
stream = resolve(_EMBED, _REFERER)
9091
assert stream.ext == "m3u8"
92+
93+
94+
@pytest.mark.unit
95+
class TestResolveUqloadFixture:
96+
def test_packed_dl_fixture_resolves_after_unpacking(self, fake_session, load_fixture):
97+
body = load_fixture("uqload_dl_packed.html")
98+
assert "v.mp4" not in body
99+
session = fake_session(uqload, post=[body])
100+
stream = resolve(_EMBED, _REFERER)
101+
102+
assert stream.url == "https://cdn.example/3rfktestfilecode/v.mp4"
103+
assert stream.ext == "mp4"
104+
assert stream.headers["Referer"] == _EMBED
105+
106+
call = session.post.call_args
107+
assert call.args == ("https://uqload.is/dl",)
108+
assert call.kwargs["data"] == {"op": "embed", "file_code": "mhq7xbccftp5", "auto": "0", "referer": _EMBED}
109+
assert call.kwargs["headers"]["Referer"] == _EMBED

0 commit comments

Comments
 (0)