Spaces:
Running on Zero
Running on Zero
| 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 | |