"""Bounded single-worker scheduling and cancellation through real process cleanup."""
import os
import sys
import threading
import time

import pytest

from producer import gui_jobs, setup
from producer.gui import rooted


def terminal(jobs, job_id):
    deadline = time.monotonic() + 10
    while time.monotonic() < deadline:
        job = jobs.get(job_id)
        if job["status"] in ("succeeded", "failed", "cancelled"):
            return job
        time.sleep(0.01)
    pytest.fail("Job failed to finish")


def test_single_worker_bounded_queue_and_running_process_cancellation(wav, tmp_path, monkeypatch):
    source = wav()
    started = threading.Event()
    pid_file = tmp_path / "child.pid"
    calls = []
    def slow_scan(path, progress):
        calls.append(path)
        if len(calls) == 1:
            started.set()
            setup.run_process([sys.executable, "-c",
                f"import os,time;open({str(pid_file)!r},'w').write(str(os.getpid()));time.sleep(60)"],
                "Waiting for worker", progress)
        return {"bins": 2048}
    monkeypatch.setattr(gui_jobs, "waveform", slow_scan)
    jobs = gui_jobs.Jobs(tmp_path, lambda value: rooted(tmp_path, value))
    snapshot = {"path": source.name, "input_id": gui_jobs.identity(source)}
    try:
        first = jobs.submit("scan", snapshot)
        assert started.wait(5)
        # Public snapshots cannot mutate the immutable in-flight job.
        first["snapshot"]["path"] = "missing.wav"
        queued = [jobs.submit("scan", snapshot) for _ in range(3)]
        with pytest.raises(gui_jobs.Busy):
            jobs.submit("scan", snapshot)
        assert len(calls) == 1
        assert jobs.cancel(queued[0]["id"])["status"] == "cancelled"
        deadline = time.monotonic() + 5
        while not pid_file.exists() and time.monotonic() < deadline:
            time.sleep(0.01)
        pid = int(pid_file.read_text())
        jobs.cancel(first["id"])
        assert terminal(jobs, first["id"])["status"] == "cancelled"
        with pytest.raises(ProcessLookupError):
            os.kill(pid, 0)
        assert terminal(jobs, queued[1]["id"])["status"] == "succeeded"
        assert terminal(jobs, queued[2]["id"])["cached"]
        assert len(calls) == 2
    finally:
        jobs.close()
    assert not jobs.work.exists()


def test_history_and_result_cache_are_bounded_and_evicted_media_expires(wav, tmp_path, monkeypatch):
    source = wav()
    monkeypatch.setattr(gui_jobs, "MAX_HISTORY", 5)
    monkeypatch.setattr(gui_jobs, "CACHE_ENTRIES", 2)
    def preview(path, work, snapshot, progress):
        (work / "original.wav").write_bytes(b"preview")
        (work / "processed.wav").write_bytes(b"preview")
        return {}
    monkeypatch.setattr(gui_jobs, "render_preview", preview)
    jobs = gui_jobs.Jobs(tmp_path, lambda value: rooted(tmp_path, value))
    snapshot = {"path": source.name, "input_id": gui_jobs.identity(source)}
    try:
        results = []
        for index in range(10):
            job = jobs.submit("preview", snapshot | {"start": index})
            result = terminal(jobs, job["id"])
            assert result["status"] == "succeeded", result
            results.append(result["result"]["result_id"])
        assert len(jobs.list()) == 5
        assert len(jobs.cache) == 2
        assert len(list(jobs.work.iterdir())) == 2
        with pytest.raises(KeyError):
            jobs.media(results[0], "original")
        assert jobs.media(results[-1], "original") == b"preview"
        with pytest.raises(KeyError):
            jobs.media(results[-1], "../original")
        # Byte budget independently evicts even below the entry-count limit.
        monkeypatch.setattr(gui_jobs, "CACHE_BYTES", 1)
        with jobs.condition:
            jobs._evict()
        assert not jobs.cache
        assert not list(jobs.work.iterdir())
    finally:
        jobs.close()


def test_queued_input_replacement_is_rejected_and_shutdown_cancels_scan(wav, tmp_path, monkeypatch):
    source = wav()
    started = threading.Event()
    def scan(path, progress):
        started.set()
        while True:
            progress("Scanning", 1, 10)
            time.sleep(0.01)
    monkeypatch.setattr(gui_jobs, "waveform", scan)
    jobs = gui_jobs.Jobs(tmp_path, lambda value: rooted(tmp_path, value))
    snapshot = {"path": source.name, "input_id": gui_jobs.identity(source)}
    running = jobs.submit("scan", snapshot)
    assert started.wait(5)
    queued = jobs.submit("scan", snapshot)
    with source.open("ab") as handle:
        handle.write(b"replacement")
    jobs.cancel(running["id"])
    assert terminal(jobs, queued["id"])["status"] == "failed"
    assert "changed" in jobs.get(queued["id"])["error"]
    # Submit with updated identity and exercise app shutdown cleanup.
    last = jobs.submit("scan", snapshot | {"input_id": gui_jobs.identity(source)})
    jobs.close()
    assert jobs.get(last["id"])["status"] == "cancelled"
    assert not jobs.work.exists()