72 lines
4.7 KiB
Python
72 lines
4.7 KiB
Python
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest.mock import Mock
|
|
from video_comfy import VideoComfy
|
|
from profiles import LTX_REPO,LTX_FILE,LTX_COMPONENTS,Profiles
|
|
|
|
class VideoComfyTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.tmp=tempfile.TemporaryDirectory();self.root=Path(self.tmp.name)
|
|
self.model=dict(id='a'*64,repo=LTX_REPO,file=LTX_FILE,kind='video',revision='1'*40)
|
|
self.entries=[self.model]+[dict(id=str(i)*64,repo=LTX_REPO,file=info['files'][0],kind='video',revision='1'*40) for i,info in enumerate(LTX_COMPONENTS.values(),2)]
|
|
self.catalog=Mock();self.catalog.status.side_effect=lambda:dict(entries=self.entries)
|
|
def entry(ident):
|
|
row=next((x for x in self.entries if x['id']==ident),None)
|
|
if not row:raise ValueError('missing')
|
|
return row
|
|
self.catalog.entry.side_effect=entry;self.profiles=Profiles(self.root/'profiles.json',self.catalog)
|
|
self.runtime=Mock();python=self.root/'python';python.touch();comfy=self.root/'comfy';(comfy/'comfy/ldm/lightricks').mkdir(parents=True)
|
|
self.runtime.paths.return_value=(python,comfy);self.runtime.status.return_value=dict(installed=True)
|
|
self.scheduler=Mock();self.scheduler.gpu_mode='llm'
|
|
helper=Mock();helper.call.return_value={'services':[]}
|
|
self.video=VideoComfy(self.scheduler,Mock(),helper,self.root/'video',self.catalog,self.profiles,self.runtime)
|
|
def tearDown(self):self.tmp.cleanup()
|
|
def test_ready_reuses_runtime_without_install(self):
|
|
row=self.video.models()[0];self.assertTrue(row['runnable']);self.assertEqual(len(row['components']),4)
|
|
self.video.select(self.model['id']);self.assertEqual(self.video.status()['selected'],self.model['id']);self.runtime.install.assert_not_called()
|
|
def test_missing_component_is_not_ready(self):
|
|
self.entries.pop();row=self.video.models()[0];self.assertFalse(row['runnable']);self.assertIn('Spatial',row['blockers'][0])
|
|
def test_wrong_revision_is_not_silently_used(self):
|
|
self.entries[1]['revision']='2'*40;self.assertFalse(self.video.models()[0]['runnable'])
|
|
def test_missing_runtime_and_model_cannot_start(self):
|
|
self.runtime.paths.return_value=(self.root/'missing',self.root/'missing-comfy')
|
|
self.video.select(self.model['id'])
|
|
with self.assertRaises(ValueError):self.video.switch('video')
|
|
def test_unknown_models_not_advertised(self):
|
|
self.model['repo']='unknown/model';self.assertEqual(self.video.models(),[])
|
|
with self.assertRaises(ValueError):self.video.select(self.model['id'])
|
|
|
|
def test_service_token_only_authorizes_video(self):
|
|
self.video.root.mkdir();(self.video.root/'comfy-client-token').write_text('synthetic-service-secret')
|
|
self.assertFalse(self.video.service_authenticated('Bearer synthetic-service-secret'))
|
|
self.scheduler.gpu_mode='video'
|
|
self.assertTrue(self.video.service_authenticated('Bearer synthetic-service-secret'))
|
|
self.assertFalse(self.video.service_authenticated('Bearer wrong'))
|
|
self.assertFalse(self.video.service_authenticated('Bearer ü'))
|
|
|
|
def test_music_switch_is_exclusive_and_returns_to_llm(self):
|
|
import threading
|
|
from types import SimpleNamespace
|
|
scheduler=SimpleNamespace(gpu_mode='llm',cv=threading.Condition(),active=0,tts_active=0,transition=False,key=('chat',))
|
|
self.video.scheduler=scheduler;self.video._stop=Mock();self.video._foreign_services=lambda:[]
|
|
self.video.music_profiles=lambda:[dict(id='music',name='YuE2')]
|
|
self.video.switch('music');self.video.thread.join(2)
|
|
self.assertEqual(scheduler.gpu_mode,'music');self.assertEqual(self.video.state,'ready');self.video.stop_owned.assert_called_once();self.video._stop.assert_called_once()
|
|
self.video.switch('llm');self.video.thread.join(2);self.assertEqual(scheduler.gpu_mode,'llm')
|
|
def test_music_requires_enabled_ready_profile(self):
|
|
with self.assertRaisesRegex(ValueError,'Musikprofil'):self.video.switch('music')
|
|
|
|
def test_separator_switch_drains_owned_workers_without_starting_video(self):
|
|
import threading
|
|
from types import SimpleNamespace
|
|
self.video.scheduler=SimpleNamespace(gpu_mode='llm',cv=threading.Condition(),active=0,tts_active=0,transition=False,key=('chat',))
|
|
self.video._stop=Mock();self.video._start=Mock();self.video._foreign_services=lambda:[]
|
|
self.video.separator_status=lambda:dict(installed=True,models=[dict(installed=True)])
|
|
self.video.switch('separator');self.video.thread.join(2)
|
|
self.assertEqual(self.video.scheduler.gpu_mode,'separator');self.video._start.assert_not_called();self.video.stop_owned.assert_called_once()
|
|
self.video.switch('llm');self.video.thread.join(2);self.assertEqual(self.video.scheduler.gpu_mode,'llm')
|
|
def test_separator_requires_ready_model(self):
|
|
self.video.separator_status=lambda:dict(installed=False,models=[])
|
|
with self.assertRaisesRegex(ValueError,'Audio-Separator'):self.video.switch('separator')
|