Spaces:
Running on Zero
Running on Zero
File size: 2,364 Bytes
1b2da16 | 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 | import json
import math
import pytest
import gradio as gr
from alignment import attach_timings
from controller import Controller
from lifecycle import check_cancelled
from test_lifecycle import FakeEngine, invoke
def test_preserves_exact_unicode_and_punctuation():
text=" L’été, c’est déjà 2026 ! 你好。"
tokens=["L'été", "c'est", "déjà", "2026", "你", "好"]
words=[dict(text=t,start_time=i*.2,end_time=(i+1)*.2) for i,t in enumerate(tokens)]
result=attach_timings(text,words,1.2)
assert ''.join(w['text'] for w in result)==text
assert result[-1]['end']==1.2
@pytest.mark.parametrize('words',[
[dict(text='Other',start_time=0,end_time=.1)],
[dict(text='Hello',start_time=math.nan,end_time=.1)],
[dict(text='Hello',start_time=0,end_time=2)],
[dict(text='Hello',start_time=.1,end_time=0)],
[dict(text='Hello',start_time=0,end_time=0)],
])
def test_rejects_unreliable_timing(words):
with pytest.raises(ValueError):attach_timings('Hello',words,.5)
class FakeAligner:
def __init__(self,fail=False):self.fail=fail;self.calls=0
def prepare(self,pcm,text):return text
def run(self,text,marker,progress):
self.calls+=1
check_cancelled(marker)
if self.fail:raise RuntimeError('alignment failure')
return [dict(text='Hello',start_time=0,end_time=.08)]
def test_second_stage_failure_retains_new_audio(tmp_path):
c=Controller(FakeEngine(),tmp_path,['Savannah'],FakeAligner(fail=True))
result=list(invoke(c,text='Hello'))[-1]
payload=json.loads(result[0])
assert payload['alignment']=='unavailable' and payload['words']==[]
assert 'unavailable' in result[1] and len(list(c.outputs.glob('*.wav')))==1
assert c.jobs.active()==0
def test_cancel_between_calls_cleans_export_and_retries(tmp_path):
a=FakeAligner();c=Controller(FakeEngine(),tmp_path,['Savannah'],a)
iterator=invoke(c,text='Hello')
next(iterator);next(iterator);state=next(iterator)
assert 'Synchronizing' in state[1]
c.cancel(gr.Request(session_hash='session-a'))
result=list(iterator)[-1]
assert 'Cancelled' in result[1] and result[0]==gr.skip() and a.calls==0
assert not list(c.outputs.glob('*.wav')) and c.jobs.active()==0
result=list(invoke(c,text='Hello'))[-1]
assert json.loads(result[0])['alignment']=='ready' and a.calls==1
|