Replace Deck video generation with selected external service lifecycle

test_video.py

@@ -1,65 +1,36 @@
 import tempfile
-import time
 import unittest
-from pathlib import Path
-from unittest.mock import Mock,patch
-from video import Video
+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 make(self,d):
-  scheduler=Scheduler(Mock());profiles=Mock();profiles.status.return_value={'profiles':[dict(id='v',name='test',kind='video')]}
-  return Video(Path(d),profiles,Mock(),scheduler,Mock())
- def wait(self,v):
-  v.thread.join(3);self.assertFalse(v.thread.is_alive())
- def test_exclusive_mode_stops_owned_work_and_restores_llm(self):
+ def test_switch_and_persistent_provider_selection(self):
   with tempfile.TemporaryDirectory() as d:
-   v=self.make(d);v.select('v');v.blockers=Mock(return_value=[]);v.load=Mock()
-   v.switch('video');self.wait(v)
-   self.assertEqual(v.mode,'video');v.stop_owned.assert_called_once();v.load.assert_called_once()
+   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 v.scheduler.lease(('chat',)):pass
-   with self.assertRaises(ValueError):v.select('v')
-   v.switch('llm');self.wait(v);self.assertEqual(v.scheduler.gpu_mode,'llm')
-   with v.scheduler.lease(('chat',)):pass
- def test_failed_load_does_not_leave_gpu_gate_locked(self):
+    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:
-   v=self.make(d);v.select('v');v.blockers=Mock(return_value=[]);v.load=Mock(side_effect=ValueError('busy'))
-   v.switch('video');self.wait(v)
-   self.assertEqual(v.status()['error'],'busy');self.assertEqual(v.scheduler.gpu_mode,'llm')
- def test_clear_video_selection_persists(self):
+   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:
-   v=self.make(d);v.select('v');v.select(None)
-   self.assertIsNone(v.selected);self.assertIsNone(self.make(d).selected)
-   with self.assertRaises(ValueError):v.switch('video')
- def test_foreign_process_prevents_spawn(self):
-  with tempfile.TemporaryDirectory() as d,patch('video.probe',return_value=[{'processes':1}]),patch('video.subprocess.Popen') as spawn:
-   v=self.make(d)
-   with self.assertRaises(ValueError):v.load({})
-   spawn.assert_not_called()
- def test_request_parameters_override_defaults_and_validate(self):
+   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:
-   v=self.make(d);v.mode='video';v.state='ready';v.profile={'id':'v','name':'test','parameters':dict(width=512,height=320,fps=24,frames=9,seed=42)}
-   with patch('video.threading.Thread') as thread,patch('video.shutil.disk_usage') as disk:
-    disk.return_value.free=100*1024**3
-    result=v.generate({'prompt':'synthetic','width':640})
-    self.assertEqual(result['parameters']['width'],640);self.assertEqual(result['parameters']['fps'],24)
-    self.assertNotIn('prompt',str(v.status()))
-    with self.assertRaises(ValueError):v.generate({'prompt':'synthetic'})
-    v.job=None
-    for extra in [{'frames':10},{'fps':float('nan')},{'width':True},{'model':'wrong'},{'steps':100}]:
-     with self.assertRaises(ValueError):v.generate(dict(prompt='synthetic',**extra))
- def test_multiple_worker_messages_in_one_read(self):
-  import subprocess,sys
-  with tempfile.TemporaryDirectory() as d:
-   v=self.make(d);v.job={'id':'test'}
-   p=subprocess.Popen([sys.executable,'-c','import sys,time;sys.stdout.write(\'{"state":"progress","phase":"test"}\\n{"state":"complete"}\\n\');sys.stdout.flush();time.sleep(3)'],stdout=subprocess.PIPE,bufsize=0)
-   v.process=p
-   try:
-    self.assertEqual(v.read(p,1)['state'],'complete');self.assertEqual(v.job['phase'],'test')
-   finally:p.terminate();p.wait();p.stdout.close()
- def test_missing_components_leave_current_mode_untouched(self):
-  with tempfile.TemporaryDirectory() as d:
-   v=self.make(d);v.select('v');v.blockers=Mock(return_value=['missing'])
-   with self.assertRaises(ValueError):v.switch('video')
-   self.assertEqual(v.scheduler.gpu_mode,'llm');v.stop_owned.assert_not_called()
+   h=Helper();v=Video(Scheduler(Mock()),Mock(),h,d);v.select('ltx');h.call=Mock(side_effect=OSError())
+   self.assertEqual(v.status()['mode'],'video')