Spaces:
Running on Zero
Running on Zero
File size: 5,793 Bytes
6d1acb0 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 | 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
|