37 lines
2.3 KiB
Python
37 lines
2.3 KiB
Python
import tempfile
|
|
import unittest
|
|
from unittest.mock import Mock
|
|
from inference import Scheduler,InferenceError
|
|
from video import Video
|
|
|
|
class Helper:
|
|
def __init__(self):self.services=[dict(id='ltx',name='LTX',running=False,ready=False),dict(id='other',name='Other',running=False,ready=False)];self.actions=[];self.fail_start=False
|
|
def call(self,action,ident=None):
|
|
self.actions.append((action,ident))
|
|
if action=='video-status':return {'services':[dict(s) for s in self.services]}
|
|
if action=='video-start' and self.fail_start:raise ValueError('GPU belegt')
|
|
row=next(s for s in self.services if s['id']==ident);row['running']=row['ready']=action=='video-start';return dict(row)
|
|
class VideoTests(unittest.TestCase):
|
|
def test_switch_and_persistent_provider_selection(self):
|
|
with tempfile.TemporaryDirectory() as d:
|
|
h=Helper();stop=Mock();scheduler=Scheduler(Mock());v=Video(scheduler,stop,h,d)
|
|
v.select('ltx');self.assertEqual(Video(Scheduler(Mock()),Mock(),h,d).selected,'ltx')
|
|
v.switch('video');v.thread.join(3);self.assertEqual(v.status()['mode'],'video');stop.assert_called_once()
|
|
with self.assertRaises(InferenceError):
|
|
with scheduler.lease(('chat',)):pass
|
|
with self.assertRaises(ValueError):v.select('other')
|
|
v.switch('llm');v.thread.join(3);self.assertEqual(v.status()['mode'],'llm')
|
|
v.select('other');v.switch('video');v.thread.join(3);self.assertIn(('video-start','other'),h.actions)
|
|
def test_restart_adopts_running_service_without_stopping_it(self):
|
|
with tempfile.TemporaryDirectory() as d:
|
|
h=Helper();h.services[0].update(running=True,ready=True);v=Video(Scheduler(Mock()),Mock(),h,d)
|
|
self.assertEqual(v.status()['mode'],'video');v.close();self.assertFalse(any(a=='video-stop' for a,_ in h.actions))
|
|
def test_foreign_gpu_failure_reopens_llm_without_starting_service(self):
|
|
with tempfile.TemporaryDirectory() as d:
|
|
h=Helper();h.fail_start=True;v=Video(Scheduler(Mock()),Mock(),h,d);v.select('ltx');v.switch('video');v.thread.join(3)
|
|
self.assertEqual(v.status()['mode'],'llm');self.assertEqual(v.error,'GPU belegt')
|
|
def test_helper_loss_fails_closed_for_configured_service(self):
|
|
with tempfile.TemporaryDirectory() as d:
|
|
h=Helper();v=Video(Scheduler(Mock()),Mock(),h,d);v.select('ltx');h.call=Mock(side_effect=OSError())
|
|
self.assertEqual(v.status()['mode'],'video')
|