ohm_streaming/tests/test_downloads.py

"""Tests du gestionnaire de téléchargements (serveur HTTP local avec Range)."""

import asyncio

import pytest
from aiohttp import web

from app.config import get_settings
from app.db import db
from app.services.downloads import download_manager as dm

PAYLOAD = b"x" * 500_000


async def _start_file_server() -> tuple[web.AppRunner, int]:
    """Mini serveur HTTP supportant les requêtes Range."""

    async def handle(request: web.Request) -> web.StreamResponse:
        range_header = request.headers.get("Range")
        if range_header:
            start = int(range_header.removeprefix("bytes=").split("-")[0])
            return web.Response(
                body=PAYLOAD[start:],
                status=206,
                headers={"Content-Range": f"bytes {start}-{len(PAYLOAD) - 1}/{len(PAYLOAD)}"},
            )
        return web.Response(body=PAYLOAD, headers={"Content-Length": str(len(PAYLOAD))})

    app = web.Application()
    app.router.add_get("/video.mp4", handle)
    runner = web.AppRunner(app)
    await runner.setup()
    site = web.TCPSite(runner, "127.0.0.1", 0)
    await site.start()
    port = site._server.sockets[0].getsockname()[1]
    return runner, port


@pytest.fixture
async def file_server():
    runner, port = await _start_file_server()
    yield f"http://127.0.0.1:{port}"
    await runner.cleanup()


@pytest.fixture
async def manager():
    await dm.start()
    yield dm
    await dm.stop()


async def _wait_status(download_id: int, wanted: set[str], timeout: float = 10) -> dict:
    for _ in range(int(timeout * 10)):
        data = await dm.get(download_id)
        if data["status"] in wanted:
            return data
        await asyncio.sleep(0.1)
    raise AssertionError(f"statut {wanted} non atteint (actuel: {data['status']})")


async def test_full_download(file_server, manager):
    d = await dm.enqueue(f"{file_server}/video.mp4", "http://page", "Episode Test 1")
    data = await _wait_status(d["id"], {"done", "failed"})
    assert data["status"] == "done", data.get("error")
    assert data["percent"] == 100.0
    path = get_settings().download_dir / data["file_path"]
    assert path.read_bytes() == PAYLOAD


async def test_anti_duplicate(file_server, manager):
    d1 = await dm.enqueue(f"{file_server}/video.mp4", "http://page", "Ep")
    d2 = await dm.enqueue(f"{file_server}/video.mp4", "http://page", "Ep")
    assert d1["id"] == d2["id"]
    assert d2["duplicate"] is True
    await _wait_status(d1["id"], {"done", "failed"})


async def test_pause_resume(file_server, manager):
    d = await dm.enqueue(f"{file_server}/video.mp4", "http://page", "Ep Pause")
    await asyncio.sleep(0.2)
    await dm.pause(d["id"])
    data = await _wait_status(d["id"], {"paused", "done"})
    if data["status"] == "paused":  # assez rapide pour être pausé
        assert await dm.resume(d["id"]) is True
    data = await _wait_status(d["id"], {"done", "failed"})
    assert data["status"] == "done", data.get("error")
    path = get_settings().download_dir / data["file_path"]
    assert path.read_bytes() == PAYLOAD


async def test_failed_download_and_retry(manager):
    d = await dm.enqueue("http://127.0.0.1:1/nope.mp4", "http://page", "Ep KO")
    data = await _wait_status(d["id"], {"failed"})
    assert data["error"]
    assert await dm.retry(d["id"]) is True
    data = await _wait_status(d["id"], {"failed"})
    assert await dm.cancel(d["id"]) is True


async def test_cancel(file_server, manager):
    d = await dm.enqueue(f"{file_server}/video.mp4", "http://page", "Ep Cancel")
    await dm.cancel(d["id"])
    data = await dm.get(d["id"])
    assert data["status"] in ("cancelled", "done")  # course possible si déjà fini

async def test_delete_keeps_file(file_server, manager):
    d = await dm.enqueue(f"{file_server}/video.mp4", "http://page", "Anime X - E1 (VOSTFR)")
    data = await _wait_status(d["id"], {"done", "failed"})
    assert data["status"] == "done", data.get("error")
    path = get_settings().download_dir / data["file_path"]
    assert path.is_file()
    assert await dm.delete(d["id"]) is True
    assert await dm.get(d["id"]) is None
    assert path.is_file()  # le fichier vidéo est conservé


async def test_delete_with_file(file_server, manager):
    d = await dm.enqueue(f"{file_server}/video.mp4", "http://page", "Anime X - E2 (VOSTFR)")
    data = await _wait_status(d["id"], {"done", "failed"})
    path = get_settings().download_dir / data["file_path"]
    assert await dm.delete(d["id"], delete_file=True) is True
    assert await dm.get(d["id"]) is None
    assert not path.exists()
    assert await dm.delete(d["id"]) is False  # déjà supprimé → introuvable


async def test_delete_active(file_server, manager):
    d = await dm.enqueue(f"{file_server}/video.mp4", "http://page", "Anime X - E3")
    filename = d["file_path"]
    assert await dm.delete(d["id"]) is True
    assert await dm.get(d["id"]) is None
    download_dir = get_settings().download_dir
    assert not (download_dir / filename).exists()
    assert not (download_dir / (filename + ".part")).exists()

async def test_series_subfolder_and_scan(file_server, manager):
    """Les épisodes sont rangés dans un sous-dossier par animé, restaurés au scan."""
    d = await dm.enqueue(f"{file_server}/video.mp4", "http://page", "Anime Z - E7 (VOSTFR)")
    data = await _wait_status(d["id"], {"done", "failed"})
    assert data["status"] == "done", data.get("error")
    assert data["file_path"] == "Anime Z/Anime Z - S01E07 (VOSTFR).mp4"
    assert (get_settings().download_dir / data["file_path"]).read_bytes() == PAYLOAD

    # DB vidée + redémarrage → le fichier en sous-dossier est restauré tel quel
    await db.execute("DELETE FROM downloads")
    await dm.stop()
    await dm.start()
    items = await dm.list_all()
    paths = [i["file_path"] for i in items]  # d'autres fichiers de tests peuvent trainer
    assert "Anime Z/Anime Z - S01E07 (VOSTFR).mp4" in paths