"""Idle HLS sessions are reaped on a timer, not only when the next session is requested.

plex/stream.py promises "a periodic reaper kills idle ones", but `reap_idle` ran only inside
POST /session: after the last playback, an abandoned session (its ffmpeg still working through the
film, its directory a whole segmented copy of it) lived until somebody pressed play again.

And it must not take a PAUSED viewer's session: a remux finishes long before the film does, and a
paused player with a full buffer fetches nothing — so "finished and unfetched for 30 s" (the rule a
new session request applies) is not "abandoned".
"""

import asyncio
import time

import pytest

import app.main as app_main
from app.main import app
from app.plex import stream


class _Configured:
    def __enter__(self):
        return None

    def __exit__(self, *exc):
        return False


def test_idle_sessions_are_reaped_while_nobody_starts_a_new_one(monkeypatch):
    async def _no_sweep():
        return None

    reaped: list[int] = []
    monkeypatch.setattr(app_main, "SessionLocal", _Configured)
    monkeypatch.setattr(app_main.state, "is_configured", lambda db: True)
    monkeypatch.setattr(app_main, "_sweep_hls_orphans", _no_sweep)
    monkeypatch.setattr(app_main, "start_scheduler", lambda: None)
    monkeypatch.setattr(app_main, "shutdown_scheduler", lambda: None)
    monkeypatch.setattr(app_main.plex_subtitles, "shutdown", lambda: None)
    monkeypatch.setattr(app_main, "_HLS_REAP_EVERY_S", 0.01)
    monkeypatch.setattr(app_main.plex_stream, "reap_abandoned", lambda: reaped.append(1) or 0)

    async def _cycle():
        async with app_main.lifespan(app):
            await asyncio.sleep(0.2)

    asyncio.run(_cycle())
    assert len(reaped) >= 2, "no periodic reap ran while the app was up"


class _Proc:
    def __init__(self, finished: bool):
        self.finished = finished
        self.killed = False

    def poll(self):
        return 0 if self.finished else None

    def terminate(self):
        self.killed = True

    def wait(self, timeout=None):
        return 0


class _Session:
    def __init__(self, tmp_path, name: str, finished: bool, idle_s: float):
        self.dir = tmp_path / name
        self.dir.mkdir()
        self.proc = _Proc(finished)
        self.last_access = time.time() - idle_s


@pytest.fixture
def sessions(monkeypatch):
    live: dict = {}
    monkeypatch.setattr(stream, "_sessions", live)
    return live


def test_the_periodic_pass_keeps_a_paused_viewers_finished_session(tmp_path, sessions):
    # A remux that finished a minute ago, its viewer paused with a full buffer since.
    s = _Session(tmp_path, "paused", finished=True, idle_s=90)
    sessions["12345"] = s
    assert stream.reap_abandoned() == 0
    assert "12345" in sessions and s.dir.exists() and not s.proc.killed


def test_the_periodic_pass_still_reaps_what_nobody_has_touched_for_the_idle_limit(
    tmp_path, sessions
):
    sessions["1"] = _Session(
        tmp_path, "gone_done", finished=True, idle_s=stream._SESSION_IDLE_S + 5
    )
    sessions["2"] = _Session(
        tmp_path, "gone_live", finished=False, idle_s=stream._SESSION_IDLE_S + 5
    )
    assert stream.reap_abandoned() == 2
    assert sessions == {}


def test_a_new_session_request_still_reaps_finished_sessions_quickly(tmp_path, sessions):
    sessions["12345"] = _Session(tmp_path, "done", finished=True, idle_s=45)
    assert stream.reap_idle() == 1
    assert sessions == {}