from concurrent.futures import ThreadPoolExecutor
import json
import threading
import time
from types import SimpleNamespace

import gradio as gr
import numpy as np
import pytest
import soundfile as sf

from controller import Controller
from lifecycle import AlreadyRunning, Cancelled, Jobs, check_cancelled


class FakeEngine:
    """Lifecycle fixture only. This is not an inference or audio-quality test."""
    def __init__(self, fail=False):
        self.calls=0; self.active=0; self.peak=0; self.lock=threading.Lock(); self.fail=fail

    def prepare(self, request, reference):
        return request

    def run(self, request, marker, progress):
        with self.lock:
            self.calls+=1; self.active+=1; self.peak=max(self.peak,self.active)
        try:
            time.sleep(.02)
            check_cancelled(marker)
            if self.fail: raise RuntimeError("private user text must not be exposed")
            return SimpleNamespace(pcm=np.zeros(4410,dtype=np.float32),seconds=.02,cancelled=False)
        finally:
            with self.lock:self.active-=1


def invoke(controller, session="session-a", **changes):
    args=dict(text="A short test.",picker=json.dumps({"voice":"Savannah","language":"en_us"}),
              source="Preset",clip=None,consent=False,normalize=True,seed=42,randomize=False,
              temperature=.75,top_p=.69,top_k=80,repetition_penalty=2.4,rate=0,
              request=gr.Request(session_hash=session), progress=lambda *a,**k:None)
    args.update(changes)
    return controller.generate(**args)


def test_double_submit_and_owner_isolation(tmp_path):
    jobs=Jobs(tmp_path)
    a=jobs.begin("a"); b=jobs.begin("b")
    with pytest.raises(AlreadyRunning):jobs.begin("a")
    assert jobs.cancel("a")
    with pytest.raises(Cancelled):check_cancelled(a.marker)
    check_cancelled(b.marker)
    jobs.finish(a);jobs.finish(b)
    assert jobs.active()==0 and not list(tmp_path.glob("*.cancel"))


def test_success_cleanup_precedes_final_yield(tmp_path):
    c=Controller(FakeEngine(),tmp_path,["Savannah"])
    stream=invoke(c)
    next(stream)  # request received
    next(stream)  # GPU returned; finalizing
    result=next(stream)
    assert c.jobs.active()==0
    assert c.gate.acquire(blocking=False)
    c.gate.release()
    payload=json.loads(result[0]); assert payload["seed"]==42
    assert len(payload["wave"])==640
    wav=list(c.outputs.glob("*.wav")); assert len(wav)==1
    assert sf.info(wav[0]).samplerate==44100
    stream.close()  # Consumer never needs to resume to trigger cleanup.


def test_cancel_before_gpu_and_retry(tmp_path):
    engine=FakeEngine();c=Controller(engine,tmp_path,["Savannah"])
    stream=invoke(c);next(stream)
    c.cancel(gr.Request(session_hash="session-a"))
    values=list(stream)
    assert "Cancelled" in values[-1][1] and values[-1][0]==gr.skip()
    assert engine.calls==0 and c.jobs.active()==0
    assert not list(c.outputs.glob("*.wav"))
    assert "Ready" in list(invoke(c))[-1][1]


def test_generator_closed_early_cleans_job(tmp_path):
    c=Controller(FakeEngine(),tmp_path,["Savannah"])
    stream=invoke(c);next(stream);stream.close()
    assert c.jobs.active()==0


def test_engine_failure_keeps_previous_audio_and_releases_slot(tmp_path):
    e=FakeEngine();c=Controller(e,tmp_path,["Savannah"])
    first=list(invoke(c))[-1][0]
    files_before=set(c.outputs.glob("*.wav"));e.fail=True
    stream=invoke(c);seen=[]
    with pytest.raises(gr.Error):
        for value in stream:seen.append(value)
    assert seen[-1][0]==gr.skip() and seen[-1][2]["interactive"] is True
    assert "private user text" not in seen[-1][1]
    assert c.jobs.active()==0 and e.active==0
    assert set(c.outputs.glob("*.wav"))==files_before
    assert json.loads(first)["url"]


def test_cancel_during_gpu_callback(tmp_path):
    entered=threading.Event();resume=threading.Event()
    class Running(FakeEngine):
        def run(self, request, marker, progress):
            entered.set();assert resume.wait(2);check_cancelled(marker)
    c=Controller(Running(),tmp_path,["Savannah"])
    with ThreadPoolExecutor(1) as pool:
        future=pool.submit(lambda:list(invoke(c)))
        assert entered.wait(2)
        c.cancel(gr.Request(session_hash="session-a"));resume.set()
        assert "Cancelled" in future.result(timeout=2)[-1][1]
    assert c.jobs.active()==0 and not list(c.outputs.glob("*.wav"))


def test_two_sessions_share_one_inference_slot_and_separate_files(tmp_path):
    e=FakeEngine();c=Controller(e,tmp_path,["Savannah"])
    with ThreadPoolExecutor(2) as pool:
        results=list(pool.map(lambda session:list(invoke(c,session))[-1], ["a","b"]))
    assert e.peak==1 and e.calls==2 and c.jobs.active()==0
    assert json.loads(results[0][0])["url"]!=json.loads(results[1][0])["url"]


def test_waiting_for_slot_is_cancellable(tmp_path):
    c=Controller(FakeEngine(),tmp_path,["Savannah"]);c.gate.acquire()
    stream=invoke(c);next(stream)
    c.cancel(gr.Request(session_hash="session-a"))
    assert "Cancelled" in list(stream)[-1][1]
    c.gate.release();assert c.jobs.active()==0


def test_validation_error_is_an_api_error_without_gpu_call(tmp_path):
    e=FakeEngine();c=Controller(e,tmp_path,["Savannah"])
    stream=invoke(c,text="")
    first=next(stream)
    assert first[0]==gr.skip()
    with pytest.raises(gr.Error):next(stream)
    assert e.calls==0 and c.jobs.active()==0


def test_duplicate_api_request_cannot_reset_active_controls(tmp_path):
    c=Controller(FakeEngine(),tmp_path,["Savannah"])
    running=invoke(c);next(running)
    duplicate=invoke(c);reply=next(duplicate)
    assert reply[2]==gr.skip() and reply[3]==gr.skip()
    with pytest.raises(gr.Error):next(duplicate)
    assert c.jobs.active()==1
    running.close();assert c.jobs.active()==0