43 lines
2.6 KiB
Python
43 lines
2.6 KiB
Python
import json
|
|
from pathlib import Path
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import patch
|
|
from image_runtime import ImageRuntime
|
|
|
|
class ImageRuntimeTests(unittest.TestCase):
|
|
def test_installed_bundle_blocks_reinstall(self):
|
|
with tempfile.TemporaryDirectory() as d:
|
|
root=Path(d);python=root/'python';python.touch();comfy=root/'comfy';comfy.mkdir();(comfy/'main.py').touch()
|
|
with patch('image_runtime.BUNDLED',(python,comfy)):
|
|
manager=ImageRuntime(root/'state');self.assertTrue(manager.status()['installed'])
|
|
with self.assertRaises(ValueError):manager.start()
|
|
def test_failed_install_preserves_active_and_removes_partial(self):
|
|
with tempfile.TemporaryDirectory() as d:
|
|
root=Path(d);manager=ImageRuntime(root);manager.job=dict(id='a'*32,state='running')
|
|
with patch.object(manager,'_command',side_effect=RuntimeError('test failure')):manager._run()
|
|
self.assertEqual(manager.status()['job']['state'],'failed');self.assertFalse((root/'active.json').exists());self.assertFalse((root/('a'*32)).exists())
|
|
def test_success_activates_only_after_validation(self):
|
|
with tempfile.TemporaryDirectory() as d:
|
|
root=Path(d);manager=ImageRuntime(root);manager.job=dict(id='b'*32,state='running');commands=[]
|
|
def command(args):
|
|
self.assertFalse((root/'active.json').exists());commands.append([str(x) for x in args])
|
|
with patch.object(manager,'_command',side_effect=command):manager._run()
|
|
self.assertEqual(manager.status()['job']['state'],'complete');self.assertTrue((root/('b'*32)/'ready').exists())
|
|
self.assertTrue(any(x[-2:]==['pip','check'] for x in commands))
|
|
self.assertEqual(manager.paths()[1],root/('b'*32)/'comfy')
|
|
def test_cancel_never_activates(self):
|
|
with tempfile.TemporaryDirectory() as d:
|
|
manager=ImageRuntime(d);manager.job=dict(id='c'*32,state='running');manager.cancel.set();manager._run()
|
|
self.assertEqual(manager.status()['job']['state'],'cancelled');self.assertFalse((Path(d)/'active.json').exists())
|
|
def test_install_tempfiles_use_owned_disk_not_container_tmpfs(self):
|
|
import sys
|
|
with tempfile.TemporaryDirectory() as d:
|
|
manager=ImageRuntime(d);manager.job=dict(id='d'*32,state='running');base=Path(d)/manager.job['id'];base.mkdir()
|
|
manager._command([sys.executable,'-c',"import os; from pathlib import Path; p=Path(os.environ['TMPDIR']); assert p.name=='tmp'; (p/'test').write_text('ok')"])
|
|
self.assertEqual((base/'tmp/test').read_text(),'ok')
|
|
def test_restart_interrupted_install(self):
|
|
with tempfile.TemporaryDirectory() as d:
|
|
(Path(d)/'installation.json').write_text(json.dumps(dict(state='running')))
|
|
self.assertEqual(ImageRuntime(d).status()['job']['state'],'interrupted')
|