"""
ffprobe over a member's file is bounded, and a bounded wait stops the process.

`probe_video` runs before every stream and subtitle request and in the
enrichment pool. A file that keeps ffprobe busy must not hold any of them, and
giving up on the wait is not enough: an asyncio subprocess whose wait was
cancelled keeps running. These run a stand-in ffprobe that never answers and
check both — the call returns, and the process is gone.
"""

import asyncio
import os
import sys
import time

import pytest
from meshbay_node import media_probe, platform

pytestmark = pytest.mark.skipif(sys.platform == "win32", reason="a POSIX shell stand-in")


@pytest.fixture
def hanging_ffprobe(tmp_path, monkeypatch):
    pid_file = tmp_path / "pid"
    tool = tmp_path / "ffprobe"
    tool.write_text(f"#!/bin/sh\necho $$ > {pid_file}\nexec sleep 600\n")
    tool.chmod(0o755)
    monkeypatch.setattr(platform, "_ffprobe_path", str(tool))
    return pid_file


def _gone(pid: int) -> bool:
    try:
        os.kill(pid, 0)
    except ProcessLookupError:
        return True
    # A zombie still answers kill(0); its state says it has exited.
    try:
        with open(f"/proc/{pid}/stat") as f:
            return f.read().split()[2] == "Z"
    except OSError:
        return True


async def test_a_probe_that_never_answers_times_out_and_is_stopped(hanging_ffprobe, monkeypatch):
    monkeypatch.setattr(media_probe, "FFPROBE_TIMEOUT_SECS", 0.5)
    started = time.monotonic()
    with pytest.raises(RuntimeError, match="timed out"):
        await media_probe.probe_video("/nonexistent/file.mkv")
    assert time.monotonic() - started < 5
    assert _gone(int(hanging_ffprobe.read_text()))


async def test_a_caller_giving_up_stops_it_too(hanging_ffprobe):
    with pytest.raises(TimeoutError):
        await asyncio.wait_for(media_probe.probe_video("/nonexistent/file.mkv"), 0.5)
    for _ in range(50):
        if hanging_ffprobe.exists():
            break
        await asyncio.sleep(0.05)
    assert _gone(int(hanging_ffprobe.read_text()))