ohm_streaming/tests/test_core.py

"""Tests unitaires : format interne, sanitisation, auth, registres scrapers."""

import httpx
import pytest

from app import auth
from app.scrapers import http as http_module
from app.scrapers.base import (
    ScrapeError,
    decode_internal_url,
    encode_internal_url,
    get_source,
    import_all_scrapers,
    resolve_hoster,
)
from app.services.downloads import plex_filename, sanitize_filename

# ------------------------------------------------------------ format interne


def test_internal_url_roundtrip():
    value = encode_internal_url("https://cdn/x.mp4", "https://site/ep1", "Mon Titre")
    assert decode_internal_url(value) == ("https://cdn/x.mp4", "https://site/ep1", "Mon Titre")


def test_internal_url_rejects_separator():
    with pytest.raises(ValueError):
        encode_internal_url("https://a|b", "p", "t")


def test_internal_url_decode_invalid():
    with pytest.raises(ValueError):
        decode_internal_url("un|deux")


# ------------------------------------------------------------ sanitisation


@pytest.mark.parametrize(
    ("raw", "expected"),
    [
        ("Naruto: Ép 01 <VOSTFR> / test", "Naruto Ép 01 VOSTFR test"),
        ("..", "video"),
        ("", "video"),
        ("titre/norm\\al", "titre norm al"),
        ("  espaces  multiples  ", "espaces multiples"),
    ],
)
def test_sanitize_filename(raw, expected):
    assert sanitize_filename(raw) == expected


def test_sanitize_no_traversal():
    assert "/" not in sanitize_filename("../../etc/passwd")
    assert sanitize_filename("../../etc/passwd") == "etc passwd"


# ------------------------------------------------------------ nommage Plex


@pytest.mark.parametrize(
    ("title", "expected"),
    [
        (
            "The Eminence in Shadow - Saison 1 - E9 (VF)",
            "The Eminence in Shadow - S01E09 (VF).mp4",
        ),
        (
            "The Eminence in Shadow - Saison 1 - E10 (VF)",
            "The Eminence in Shadow - S01E10 (VF).mp4",
        ),
        ("Anime X - E1 (VOSTFR)", "Anime X - S01E01 (VOSTFR).mp4"),
        ("One Piece - E1122", "One Piece - S01E1122.mp4"),
        ("Série - saison 2 - e3 (vf)", "Série - S02E03 (vf).mp4"),
        ("Episode Test 1", None),
        ("Mon Titre", None),
        ("Series S01E05 VOSTFR WEB-DL", None),
    ],
)
def test_plex_filename(title, expected):
    assert plex_filename(title, ".mp4") == expected


# ------------------------------------------------------------ auth


async def test_first_user_is_admin():
    user = await auth.create_user("alice", "password1")
    assert user.is_admin is True
    second = await auth.create_user("bob", "password1")
    assert second.is_admin is False


async def test_authenticate():
    await auth.create_user("carol", "password1")
    assert await auth.authenticate("carol", "password1") is not None
    assert await auth.authenticate("carol", "wrong") is None
    assert await auth.authenticate("nobody", "password1") is None


async def test_access_token_roundtrip():
    user = await auth.create_user("dave", "password1")
    token = auth.create_access_token(user)
    payload = auth.decode_access_token(token)
    assert payload["username"] == "dave"
    assert auth.decode_access_token("garbage") is None


async def test_refresh_token_rotation():
    user = await auth.create_user("erin", "password1")
    token = await auth.create_refresh_token(user.id)
    assert (await auth.use_refresh_token(token)).username == "erin"
    # Un refresh token est à usage unique (rotation)
    assert await auth.use_refresh_token(token) is None


async def test_disabled_account_cannot_login():
    user = await auth.create_user("frank", "password1")
    await auth.db.execute("UPDATE users SET is_active = 0 WHERE id = ?", (user.id,))
    assert await auth.authenticate("frank", "password1") is None


# ------------------------------------------------------------ registres scrapers


def test_sources_registered():
    import_all_scrapers()
    assert get_source("vostfree").name == "vostfree"
    assert get_source("french_manga").name == "french_manga"


def test_hoster_resolution():
    import_all_scrapers()
    assert resolve_hoster("https://video.sibnet.ru/shell.php?videoid=1").name == "sibnet"
    assert resolve_hoster("https://uqload.com/embed-x.html").name == "uqload"
    assert resolve_hoster("https://vidmoly.to/embed-x.html").name == "vidmoly"
    assert resolve_hoster("https://sendvid.com/embed/x").name == "sendvid"
    assert resolve_hoster("https://inconnu.example.com/v/1") is None


# ------------------------------------------------------------ client HTTP


class _StatusClient:
    """Fake client HTTP : sert une série de codes statut puis un corps 200."""

    def __init__(self, statuses: list[int]):
        self._statuses = statuses
        self.calls = 0

    async def get(self, url, headers=None):
        self.calls += 1
        request = httpx.Request("GET", url)
        status = self._statuses.pop(0) if self._statuses else 200
        response = httpx.Response(status, request=request, text="ok")
        response.raise_for_status()
        return response


async def test_fetch_fails_fast_on_definitive_4xx(monkeypatch):
    client = _StatusClient([422])
    monkeypatch.setattr("app.scrapers.http.get_client", lambda: client)
    with pytest.raises(ScrapeError, match="422"):
        await http_module.fetch("https://site.example/lecteur/prepare/1?content=2&episode=3")
    assert client.calls == 1  # aucun retry sur erreur définitive


@pytest.mark.parametrize("status", [429, 500, 503])
async def test_fetch_retries_transitory_errors(monkeypatch, status):
    client = _StatusClient([status, 200])
    monkeypatch.setattr("app.scrapers.http.get_client", lambda: client)

    async def no_sleep(_delay):
        return None

    monkeypatch.setattr("app.scrapers.http.asyncio.sleep", no_sleep)
    assert await http_module.fetch("https://site.example/page", retries=2) == "ok"
    assert client.calls == 2