ani-web/tests/test_animekai.py

"""test_animekai.py — animekai provider + MegaPlay decrypt + /stream Range path.

All offline (monkeypatched network) so the suite never touches animekai.be,
MegaPlay, or any CDN. Live smoke tests live in test_animekai_live.py and
are skipped unless ANI_LIVE=1.

Run:  .venv/bin/pytest tests/test_animekai.py -q
"""
import base64
import os
import sys
import types
from pathlib import Path

import pytest

ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT))

# Live 1anime-style host the /stream allowlist tests will register
HOST = "my.1anime.site"
VIDEO = f"https://{HOST}/stream/69f9d2aef757c305714fdc246e06d6bc"


def test_animekai_provider_imports():
    import server.providers.animekai as a  # noqa: F401
    for fn in ("search", "details", "episodes", "languages", "master_m3u8"):
        assert callable(getattr(a, fn)), fn
    assert callable(a.stream_meta)
    assert callable(a.subtitles)


def test_animekai_stream_meta_allowlist():
    import server.providers.animekai as a
    assert a.stream_meta("https://other.example/x.m3u8") is None
    a.STREAM_HOSTS[HOST] = {"Referer": "https://" + HOST + "/"}
    assert a.stream_meta(VIDEO) == {"Referer": "https://" + HOST + "/"}
    a.STREAM_HOSTS.clear()


# --- MegaPlay payload: old plaintext shape, new AES "enc" shape ------------
# enc sample: real getSourcesNew response captured 2026-09-29 (deterministic
# per data-id; key/IV static in the player's public JS — see upstream._MP_KEY)
ENC = ("wdeBruh3qqn_i5wUNnyaPcXqidp1UWP84FfPHzGyKXA2hDZBfMCmZ4FLvs7_pQuH549Ept"
       "ax8UOjAJyRIZfrRhUGUKy9OBeGh2yB_-m_JLAlLnWTLzYZC3_C5I4ltveYoiaU66Do9Rg"
       "I9bCetmk_o87-sd66brXnWV1MbjhLnjw")


def test_megaplay_sources_old_shape_passthrough():
    from server.upstream import megaplay_sources
    d = {"sources": {"file": "https://cdn.x/master.m3u8"}, "tracks": []}
    out = megaplay_sources(d)
    assert out["sources"]["file"] == "https://cdn.x/master.m3u8"


def test_megaplay_sources_new_shape_decrypts():
    from server.upstream import megaplay_sources
    out = megaplay_sources({"enc": ENC, "tracks": [{"file": "a.vtt"}]})
    f = out["sources"]["file"]
    assert f.startswith("https://") and f.endswith("master.m3u8"), f
    assert out["tracks"] == [{"file": "a.vtt"}]  # outer fields untouched


def test_megaplay_sources_no_sources_unchanged():
    from server.upstream import megaplay_sources
    assert megaplay_sources({}) == {}


def test_megaplay_sources_garbage_enc_raises():
    from server.upstream import megaplay_sources
    # corrupt payload must raise (provider -> fallback chain), not return None
    with pytest.raises(Exception):
        megaplay_sources({"enc": "AAAA" * 10})


# --- /stream Range pass-through (progressive MP4), offline -----------------

class _FakeStream:
    """Minimal stand-in for a streaming curl_cffi response (206 + chunk)."""
    status_code = 206
    headers = {"content-type": "video/mp4",
               "content-range": f"bytes 0-1023/{1024 * 1024}",
               "accept-ranges": "bytes"}

    def iter_content(self, chunk_size=1):
        yield b"x" * 1024

    def close(self):
        pass


@pytest.fixture()
def app_mod(monkeypatch):
    monkeypatch.setenv("ANI_SOURCE", "animekai")
    import server.upstream as up
    monkeypatch.setattr(up, "get_stream",
                        lambda url, timeout=12, headers=None: _FakeStream())
    import server.app as m
    return m


def _drain(it):
    import asyncio
    async def _go():
        return b"".join([c async for c in it])
    return asyncio.run(_go())


def _call(app_mod, token, range_hdr):
    """Invoke the /stream handler in-process (no TestClient: it wants httpx).
    Returns (status, headers_dict, body_bytes)."""
    req = types.SimpleNamespace(headers={"range": range_hdr} if range_hdr else {})
    resp = app_mod.stream(token, request=req)
    body = _drain(resp.body_iterator) if getattr(resp, "body_iterator", None) is not None else resp.body
    return resp.status_code, dict(resp.headers), body


def test_stream_range_forwards_206_and_chunk(app_mod):
    import server.providers.animekai as a
    a.STREAM_HOSTS[HOST] = {"Referer": "https://" + HOST + "/"}
    token = base64.urlsafe_b64encode(VIDEO.encode()).decode()
    status, hdrs, body = _call(app_mod, token, "bytes=0-1023")
    assert status == 206, status
    assert hdrs.get("content-range") == f"bytes 0-1023/{1024 * 1024}"
    assert len(body) == 1024


def test_stream_range_not_cached(app_mod):
    import server.providers.animekai as a
    a.STREAM_HOSTS[HOST] = {"Referer": "https://" + HOST + "/"}
    token = base64.urlsafe_b64encode(VIDEO.encode()).decode()
    _call(app_mod, token, "bytes=0-1023")
    assert VIDEO not in app_mod._STREAM_CACHE, "ranged chunk poisoned the LRU"