Offer isolated VoxCPM runtime installation from model setup
This commit is contained in:
1 parent
1f0fc42344
commit
aee31359b6
18 files changed
+168
-21
No files matched your search
@@ -0,0 +1,28 @@
|
||||
import json,tempfile,unittest
|
||||
from pathlib import Path
|
||||
from unittest.mock import Mock,patch
|
||||
from voxcpm_runtime import VoxCPMRuntime,VERSION
|
||||
class VoxCPMRuntimeTests(unittest.TestCase):
|
||||
def test_requires_verified_marker_and_valid_owned_environment(self):
|
||||
with tempfile.TemporaryDirectory() as d:
|
||||
r=VoxCPMRuntime(d,None);self.assertFalse(r.status()['installed'])
|
||||
base=Path(d)/('a'*32);(base/'python/bin').mkdir(parents=True);(base/'python/bin/python').touch();(Path(d)/'active.json').write_text(json.dumps(dict(id=base.name)))
|
||||
(base/'ready').write_text('wrong');self.assertFalse(r.status()['installed'])
|
||||
(base/'ready').write_text(VERSION);self.assertTrue(r.status()['installed']);self.assertFalse(r.status()['worker_available'])
|
||||
def test_success_and_failure_do_not_touch_shared_runtime_or_load_weights(self):
|
||||
for fail in (False,True):
|
||||
with tempfile.TemporaryDirectory() as d:
|
||||
shared=Mock();shared.paths.return_value=(Path('/does/not/exist'),Path('/does/not/exist'))
|
||||
r=VoxCPMRuntime(d,shared);r.job=dict(id='b'*32,state='running',phase='test')
|
||||
calls=[]
|
||||
def run(args,env=None):
|
||||
calls.append([str(a) for a in args])
|
||||
if fail:raise RuntimeError('synthetic install failure')
|
||||
r._command=run;r._run();self.assertEqual(r.job['state'],'failed' if fail else 'complete')
|
||||
if fail:self.assertFalse((Path(d)/'active.json').exists());self.assertFalse((Path(d)/('b'*32)).exists())
|
||||
else:
|
||||
flat=' '.join(' '.join(x) for x in calls);self.assertIn('voxcpm=='+VERSION,flat);self.assertIn('pip check',flat);self.assertNotIn('snapshot_download',flat);self.assertNotIn('model.generate(',flat)
|
||||
def test_cancel_does_not_activate_partial_environment(self):
|
||||
with tempfile.TemporaryDirectory() as d:
|
||||
r=VoxCPMRuntime(d,None);r.job=dict(id='c'*32,state='running',phase='test');r._command=Mock(side_effect=InterruptedError());r._run()
|
||||
self.assertEqual(r.job['state'],'cancelled');self.assertFalse((Path(d)/'active.json').exists())
|
||||
Reference in new issue
Block a user