jozamudi
834ca80a8c
- providers/{anidb,anikoto,hianime}.py: per-source modules behind the
ANI_SOURCE fallback dispatch (upstream.py _call/_providers)
- hianime.py: pystardust/ani-cli#1897 flow — search/episode-list/
servers API, HD-1 hash -> base64 embed url, window.__P blob
(base64 + XOR 'otaku-embed-v1') -> master m3u8; registers CDN hosts
and the required Referer via stream_meta()
- upstream.py: get_stream/fetch_parallel accept extra headers;
default source order hianime,anidb,anikoto (anidb.app is 503)
- app.py: /stream allowlist accepts provider-registered hosts and
sends their Referer; status probe follows the primary provider
(hianime variant URIs are relative to the variant path)
- app.js: error-time position save floors at furthest point reached
(hls.js resets currentTime to 0 before playerFail on fatal errors)
- tests: default-order + hianime import checks; live browser suite green
77 lines
2.3 KiB
Python
77 lines
2.3 KiB
Python
"""test_sources.py — ANI_SOURCE dispatch + fallback, offline (fake providers).
|
|
|
|
Run: .venv/bin/pytest tests/test_sources.py -q
|
|
"""
|
|
import os
|
|
import sys
|
|
import types
|
|
from pathlib import Path
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0, str(ROOT))
|
|
|
|
FAKE = {"id": 1, "slug_id": "fake-1", "name": "Fake Anime", "poster": None}
|
|
|
|
|
|
def _fake(name, boom=False):
|
|
m = types.ModuleType(f"server.providers.{name}")
|
|
|
|
def search(q, boom=boom):
|
|
if boom:
|
|
raise RuntimeError(f"{name} is down (test)")
|
|
return [FAKE]
|
|
|
|
m.search = search
|
|
m.details = lambda s: {"id": 1, "name": "Fake Anime"}
|
|
m.episodes = lambda i: [{"id": 1, "number": 1, "filler": False}]
|
|
m.languages = lambda e: [{"code": "jpn", "name": "Sub", "embed_url": "x"}]
|
|
m.master_m3u8 = lambda e, l="eng": "https://cdn.example/master.m3u8"
|
|
sys.modules[m.__name__] = m
|
|
return m
|
|
|
|
|
|
def test_fallback_skips_dead_provider():
|
|
_fake("good")
|
|
_fake("dead", boom=True)
|
|
os.environ["ANI_SOURCE"] = "dead,good"
|
|
from server import upstream
|
|
r = upstream.search("anything")
|
|
assert r == [FAKE], r
|
|
|
|
|
|
def test_all_dead_raises_with_all_errors():
|
|
_fake("dead", boom=True)
|
|
os.environ["ANI_SOURCE"] = "dead"
|
|
from server import upstream
|
|
try:
|
|
upstream.search("x")
|
|
except RuntimeError as e:
|
|
assert "dead" in str(e) and "all sources failed" in str(e), e
|
|
else:
|
|
raise AssertionError("expected all-sources-failed")
|
|
|
|
|
|
def test_anidb_provider_imports():
|
|
import server.providers.anidb as a # noqa: F401
|
|
for fn in ("search", "details", "episodes", "languages", "master_m3u8"):
|
|
assert callable(getattr(a, fn)), fn
|
|
|
|
|
|
def test_default_is_hianime_first():
|
|
os.environ.pop("ANI_SOURCE", None)
|
|
from server import upstream
|
|
assert upstream._providers()[:3] == ["hianime", "anidb", "anikoto"]
|
|
|
|
|
|
def test_hianime_provider_imports():
|
|
import server.providers.hianime as a # noqa: F401
|
|
for fn in ("search", "details", "episodes", "languages", "master_m3u8"):
|
|
assert callable(getattr(a, fn)), fn
|
|
assert callable(a.stream_meta)
|
|
|
|
|
|
def test_anikoto_provider_imports():
|
|
import server.providers.anikoto as a # noqa: F401
|
|
for fn in ("search", "details", "episodes", "languages", "master_m3u8"):
|
|
assert callable(getattr(a, fn)), fn
|