Add selectable native LTX Original runtime and DeskWEB API integration

test_ltx_original.py

@@ -0,0 +1,85 @@
+import json
+import tempfile
+import unittest
+from pathlib import Path
+from unittest.mock import Mock,patch
+from ltx_original_runtime import LTXOriginalRuntime,DESKTOP_REV
+from video_original import VideoRuntimes
+from ltx_original_entry import install_device_policy
+from test_video_comfy import VideoComfyTests
+
+class OriginalSelectionTests(VideoComfyTests):
+ def setUp(self):
+  super().setUp();self.original=Mock();self.original.status.return_value={'installed':True}
+  self.video=VideoRuntimes(self.scheduler,Mock(),self.video.helper,self.root/'video',self.catalog,self.profiles,self.runtime,original_runtime=self.original)
+ def test_missing_comfy_does_not_block_installed_original(self):
+  self.runtime.paths.return_value=(self.root/'missing',self.root/'missing-comfy')
+  self.video.select_runtime('original');self.video.select(self.model['id'])
+  self.assertTrue(self.video.models()[0]['runnable']);self.assertEqual(self.video.status()['runtime_kind'],'original')
+  saved=json.loads(self.video.path.read_text());self.assertEqual(saved['runtime'],'original');self.assertEqual(saved['model_id'],self.model['id'])
+ def test_missing_original_is_explicit(self):
+  self.original.status.return_value={'installed':False};self.video.select_runtime('original')
+  self.assertFalse(self.video.models()[0]['runnable']);self.assertIn('LTX Original fehlt',self.video.models()[0]['blockers'][-1])
+ def test_no_runtime_switch_while_video_running(self):
+  self.scheduler.gpu_mode='video'
+  with self.assertRaises(ValueError):self.video.select_runtime('original')
+  self.assertEqual(self.video.runtime_kind(),'comfy')
+ def test_reject_unknown_runtime(self):
+  with self.assertRaises(ValueError):self.video.select_runtime('../backend')
+
+class InstallerTests(unittest.TestCase):
+ def test_no_implicit_install_and_no_external_active_path(self):
+  with tempfile.TemporaryDirectory() as directory:
+   root=Path(directory);image=Mock();image.paths.return_value=(root/'missing',root/'comfy')
+   runtime=LTXOriginalRuntime(root,image);self.assertFalse(runtime.status()['installed'])
+   (root/'active.json').write_text(json.dumps({'id':'../../external'}));self.assertFalse(runtime.status()['installed'])
+   self.assertEqual(runtime.status()['required_disk_gib'],25)
+ def test_existing_ready_install_is_reused(self):
+  with tempfile.TemporaryDirectory() as directory:
+   root=Path(directory);base=root/('a'*32);(base/'python/bin').mkdir(parents=True);(base/'desktop/backend').mkdir(parents=True)
+   (base/'python/bin/python').touch();(base/'desktop/backend/ltx2_server.py').touch();(base/'ready').write_text(DESKTOP_REV)
+   (root/'active.json').write_text(json.dumps({'id':base.name}));image=Mock();image.paths.return_value=(root/'missing',root/'comfy')
+   runtime=LTXOriginalRuntime(root,image);self.assertTrue(runtime.status()['installed'])
+   with self.assertRaises(ValueError):runtime.start()
+
+class DevicePolicyTests(unittest.TestCase):
+ def test_encoder_gpu_and_conditioning_transfer_preserve_native_contract(self):
+  from collections import namedtuple
+  Row=namedtuple('Row','video audio mask')
+  class Tensor:
+   def __init__(self,device):self.device=device
+   def to(self,device):return Tensor(device)
+  class Encoder:
+   def __init__(self,model_paths,dtype,device,offload_mode='none'):self.device=device;self.offload=offload_mode
+   def __call__(self,prompts,**kwargs):return [Row(Tensor(self.device),None,Tensor(self.device))]
+  torch=Mock();torch.device.side_effect=lambda x:x
+  modes=Mock();modes.DISK='disk'
+  install_device_policy(torch,Encoder,modes)
+  encoder=Encoder('paths','bf16','cuda:0');self.assertEqual(encoder.device,'cuda:1');self.assertEqual(encoder.offload,'disk')
+  result=encoder(['synthetic prompt'])[0];self.assertEqual(result.video.device,'cuda:0');self.assertIsNone(result.audio);self.assertEqual(result.mask.device,'cuda:0')
+
+class ProxyContractTests(unittest.TestCase):
+ def test_native_paths_and_errors_are_forwarded_without_schema_adapter(self):
+  import http.client
+  import threading
+  from http.server import BaseHTTPRequestHandler,ThreadingHTTPServer
+  from video_comfy_proxy import relay
+  class Native(BaseHTTPRequestHandler):
+   def log_message(self,*args):pass
+   def do_POST(self):
+    raw=self.rfile.read(int(self.headers.get('Content-Length',0)))
+    self.server.received=(self.path,raw)
+    self.send_response(422);self.send_header('Content-Type','application/json');self.end_headers();self.wfile.write(b'{"detail":"native validation"}')
+  native=ThreadingHTTPServer(('127.0.0.1',0),Native)
+  class Forward(BaseHTTPRequestHandler):
+   def log_message(self,*args):pass
+   def do_POST(self):relay(self,native.server_port)
+  forward=ThreadingHTTPServer(('127.0.0.1',0),Forward)
+  for server in (native,forward):threading.Thread(target=server.serve_forever,daemon=True).start()
+  try:
+   client=http.client.HTTPConnection('127.0.0.1',forward.server_port);body=b'{"prompt":"synthetic","imagePath":"/synthetic/input.jpg"}'
+   client.request('POST','/api/generate?contract=1',body,{'Content-Type':'application/json'})
+   response=client.getresponse();self.assertEqual(response.status,422);self.assertEqual(response.read(),b'{"detail":"native validation"}');client.close()
+   self.assertEqual(native.received,('/api/generate?contract=1',body))
+  finally:
+   for server in (forward,native):server.shutdown();server.server_close()