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()