Files
ohm_streaming/tests/test_core.py
T
Roman 41566ab5fb v0.1.0 — Réécriture complète de OhmStreaming
Nouvelle version réécrite de zéro : recherche multi-sources (Vostfree,
French-Manga), extraction 2 niveaux, proxy vidéo intégré, streaming/téléchargement
HLS, métadonnées Kitsu, bibliothèque locale, comptes JWT + administration,
découverte fusionnée, indexeur Torznab (Sonarr/Prowlarr).
2026-09-22 10:05:47 +00:00

110 lines
3.5 KiB
Python

"""Tests unitaires : format interne, sanitisation, auth, registres scrapers."""
import pytest
from app import auth
from app.scrapers.base import (
decode_internal_url,
encode_internal_url,
get_source,
import_all_scrapers,
resolve_hoster,
)
from app.services.downloads import 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"
# ------------------------------------------------------------ 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