Roman
9148b5fb6a
- Détection de version via l'API Gitea, application via Watchtower - Router système, UI admin (intégrations), bandeau de mise à jour - Dockerfile multi-étapes, docker-compose, scripts/release.sh - Version centralisée dans app/version.py (pyproject ou OHM_VERSION au build)
189 lines
8.1 KiB
Python
189 lines
8.1 KiB
Python
"""Tests d'intégration API (auth, downloads, favoris, streaming, admin)."""
|
|
|
|
from app.config import get_settings
|
|
from app.db import db
|
|
from app.scrapers.base import ScrapeError, SearchResult
|
|
|
|
|
|
async def test_health(client):
|
|
r = await client.get("/health")
|
|
assert r.status_code == 200 and r.json() == {"status": "ok"}
|
|
|
|
|
|
async def test_pages_require_auth(client):
|
|
r = await client.get("/", follow_redirects=False)
|
|
assert r.status_code == 303 and r.headers["location"] == "/login"
|
|
r = await client.get("/api/downloads", follow_redirects=False)
|
|
assert r.status_code == 303
|
|
|
|
|
|
async def test_register_login_me(client, admin_cookies):
|
|
r = await client.get("/auth/me", cookies=admin_cookies)
|
|
data = r.json()
|
|
assert data["username"] == "admin" and data["is_admin"] is True
|
|
|
|
|
|
async def test_duplicate_username(client):
|
|
await client.post("/auth/register", data={"username": "toto", "password": "secret123"})
|
|
r = await client.post("/auth/register", data={"username": "toto", "password": "secret123"})
|
|
assert r.status_code == 409
|
|
|
|
|
|
async def test_enqueue_bad_internal_url(client, admin_cookies):
|
|
r = await client.post(
|
|
"/api/downloads", json={"internal_url": "invalide"}, cookies=admin_cookies
|
|
)
|
|
assert r.status_code == 422
|
|
|
|
|
|
async def test_favorites_flow(client, admin_cookies):
|
|
fav = {"source": "vostfree", "source_id": "1-x", "title": "Frieren"}
|
|
r = await client.post("/api/favorites", json=fav, cookies=admin_cookies)
|
|
assert r.status_code == 201
|
|
r = await client.post("/api/favorites", json=fav, cookies=admin_cookies)
|
|
assert r.status_code == 409
|
|
r = await client.get("/api/favorites", cookies=admin_cookies)
|
|
assert r.json()["total"] == 1
|
|
fav_id = r.json()["items"][0]["id"]
|
|
r = await client.delete(f"/api/favorites/{fav_id}", cookies=admin_cookies)
|
|
assert r.status_code == 200
|
|
|
|
|
|
async def test_stream_range(client, admin_cookies):
|
|
payload = b"video-bytes" * 1000
|
|
path = get_settings().download_dir / "ep1.mp4"
|
|
path.write_bytes(payload)
|
|
await db.execute(
|
|
"INSERT INTO downloads (source_key, video_url, page_url, title, file_path, status, "
|
|
"total_bytes, downloaded_bytes) VALUES ('k', 'u', 'p', 'Ep 1', 'ep1.mp4', 'done', ?, ?)",
|
|
(len(payload), len(payload)),
|
|
)
|
|
r = await client.get("/api/stream/1", cookies=admin_cookies)
|
|
assert r.status_code == 200 and r.content == payload
|
|
r = await client.get("/api/stream/1", headers={"Range": "bytes=10-19"}, cookies=admin_cookies)
|
|
assert r.status_code == 206
|
|
assert r.content == payload[10:20]
|
|
assert r.headers["content-range"] == f"bytes 10-19/{len(payload)}"
|
|
r = await client.get(
|
|
"/api/stream/1", headers={"Range": "bytes=999999999-"}, cookies=admin_cookies
|
|
)
|
|
assert r.status_code == 416
|
|
|
|
|
|
async def test_admin_required(client):
|
|
await client.post("/auth/register", data={"username": "admin2", "password": "secret123"})
|
|
r = await client.post("/auth/register", data={"username": "user", "password": "secret123"})
|
|
user_cookies = r.cookies
|
|
r = await client.get("/api/admin/users", cookies=user_cookies)
|
|
assert r.status_code == 403
|
|
|
|
|
|
async def test_admin_user_management(client, admin_cookies):
|
|
await client.post("/auth/register", data={"username": "user", "password": "secret123"})
|
|
r = await client.get("/api/admin/users", cookies=admin_cookies)
|
|
users = {u["username"]: u["id"] for u in r.json()}
|
|
assert set(users) == {"admin", "user"}
|
|
admin_id, user_id = users["admin"], users["user"]
|
|
r = await client.post(f"/api/admin/users/{user_id}/toggle-active", cookies=admin_cookies)
|
|
assert r.json() == {"is_active": False}
|
|
r = await client.post(f"/api/admin/users/{user_id}/toggle-admin", cookies=admin_cookies)
|
|
assert r.json() == {"is_admin": True}
|
|
# pas d'auto-modification
|
|
r = await client.post(f"/api/admin/users/{admin_id}/toggle-admin", cookies=admin_cookies)
|
|
assert r.status_code == 400
|
|
r = await client.delete(f"/api/admin/users/{user_id}", cookies=admin_cookies)
|
|
assert r.status_code == 200
|
|
|
|
|
|
async def test_source_toggle(client, admin_cookies):
|
|
r = await client.post(
|
|
"/api/admin/sources/vostfree/toggle", json={"enabled": False}, cookies=admin_cookies
|
|
)
|
|
assert r.json()["enabled"] is False
|
|
r = await client.get("/api/sources", cookies=admin_cookies)
|
|
states = {s["name"]: s["enabled"] for s in r.json()}
|
|
assert states["vostfree"] is False
|
|
assert states["french_manga"] is True
|
|
|
|
|
|
async def test_source_health(client, admin_cookies, monkeypatch):
|
|
# source inconnue → 404 (pas de 500)
|
|
r = await client.post("/api/admin/sources/inconnue/health", cookies=admin_cookies)
|
|
assert r.status_code == 404
|
|
|
|
# état initial : aucune santé connue
|
|
r = await client.get("/api/admin/sources", cookies=admin_cookies)
|
|
assert all(s["health"] is None for s in r.json())
|
|
|
|
async def ok_search(self, query: str) -> list[SearchResult]:
|
|
return [SearchResult(source=self.name, source_id="x", title="X", url="http://x")]
|
|
|
|
monkeypatch.setattr("app.scrapers.sources.vostfree.VostfreeScraper.search", ok_search)
|
|
r = await client.post("/api/admin/sources/vostfree/health", cookies=admin_cookies)
|
|
data = r.json()
|
|
assert data["healthy"] is True and data["detail"] == "1 résultats" and data["checked_at"]
|
|
|
|
# résultat persisté et exposé par GET /sources
|
|
r = await client.get("/api/admin/sources", cookies=admin_cookies)
|
|
health = {s["name"]: s["health"] for s in r.json()}
|
|
assert health["vostfree"]["healthy"] is True
|
|
assert health["french_manga"] is None
|
|
|
|
async def failing_search(self, query: str) -> list[SearchResult]:
|
|
raise ScrapeError("site indisponible")
|
|
|
|
monkeypatch.setattr("app.scrapers.sources.vostfree.VostfreeScraper.search", failing_search)
|
|
r = await client.post("/api/admin/sources/vostfree/health", cookies=admin_cookies)
|
|
data = r.json()
|
|
assert r.status_code == 200 and data["healthy"] is False and data["detail"] == "site indisponible"
|
|
|
|
async def broken_search(self, query: str) -> list[SearchResult]:
|
|
raise RuntimeError("boom")
|
|
|
|
monkeypatch.setattr("app.scrapers.sources.vostfree.VostfreeScraper.search", broken_search)
|
|
r = await client.post("/api/admin/sources/vostfree/health", cookies=admin_cookies)
|
|
data = r.json()
|
|
assert r.status_code == 200 and data["healthy"] is False and "boom" in data["detail"]
|
|
|
|
|
|
async def test_source_toggle_unknown(client, admin_cookies):
|
|
r = await client.post(
|
|
"/api/admin/sources/inconnue/toggle", json={"enabled": True}, cookies=admin_cookies
|
|
)
|
|
assert r.status_code == 404
|
|
|
|
|
|
async def test_source_url_override(client, admin_cookies):
|
|
from app.scrapers.base import get_source
|
|
|
|
r = await client.get("/api/admin/sources", cookies=admin_cookies)
|
|
src = next(s for s in r.json() if s["name"] == "vostfree")
|
|
default = src["default_base_url"]
|
|
assert src["base_url"] == default and src["overridden"] is False
|
|
|
|
# source inconnue → 404
|
|
r = await client.put("/api/admin/sources/inconnue/url", json={"url": "https://x.org"}, cookies=admin_cookies)
|
|
assert r.status_code == 404
|
|
|
|
# URL invalide → 422
|
|
r = await client.put("/api/admin/sources/vostfree/url", json={"url": "pas-une-url"}, cookies=admin_cookies)
|
|
assert r.status_code == 422
|
|
|
|
# changement de domaine (slash final toléré) → appliqué à l'instance + persisté
|
|
r = await client.put("/api/admin/sources/vostfree/url", json={"url": "https://exemple.org/"},
|
|
cookies=admin_cookies)
|
|
assert r.status_code == 200
|
|
data = r.json()
|
|
assert data["base_url"] == "https://exemple.org" and data["overridden"] is True
|
|
assert get_source("vostfree").base_url == "https://exemple.org"
|
|
|
|
r = await client.get("/api/admin/sources", cookies=admin_cookies)
|
|
src = next(s for s in r.json() if s["name"] == "vostfree")
|
|
assert src["base_url"] == "https://exemple.org" and src["overridden"] is True
|
|
|
|
# URL vide → retour au défaut du code
|
|
r = await client.put("/api/admin/sources/vostfree/url", json={"url": ""}, cookies=admin_cookies)
|
|
data = r.json()
|
|
assert data["base_url"] == default and data["overridden"] is False
|
|
assert get_source("vostfree").base_url == default
|