37 lines
1.5 KiB
Python
37 lines
1.5 KiB
Python
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
from audio_policy import AudioPolicy
|
|
|
|
class Worker:
|
|
def __init__(self):
|
|
import threading
|
|
self.lock=threading.RLock();self.policy='auto';self.job=None;self.process=None;self.warming=False;self.unloaded=0
|
|
def unload_idle(self):self.unloaded+=1
|
|
|
|
class Profiles:
|
|
def status(self):return {'profiles':[{'id':'tts-1','kind':'audio','runnable':True},{'id':'stt-1','kind':'stt','runnable':True}]}
|
|
|
|
class Endpoint:
|
|
state='stopped'
|
|
|
|
class Scheduler:
|
|
def __init__(self):
|
|
import threading
|
|
self.cv=threading.Condition();self.gpu_mode='llm';self.active=0;self.transition=False
|
|
|
|
class PolicyTests(unittest.TestCase):
|
|
def test_persist_validate_and_unload(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
path=Path(directory)/'audio-policy.json';tts=Worker();stt=Worker()
|
|
manager=AudioPolicy(path,Profiles(),tts,stt,Scheduler(),Endpoint())
|
|
with self.assertRaises(ValueError):manager.configure({'kind':'tts','mode':'warm','profile_id':'missing'})
|
|
self.assertFalse(path.exists())
|
|
manager.configure({'kind':'tts','mode':'warm','profile_id':'tts-1'})
|
|
self.assertEqual(tts.policy,'warm')
|
|
manager.configure({'kind':'stt','mode':'per_request','profile_id':None})
|
|
self.assertEqual(stt.unloaded,1)
|
|
restored=AudioPolicy(path,Profiles(),Worker(),Worker(),Scheduler(),Endpoint())
|
|
self.assertEqual(restored.status()['settings']['tts']['mode'],'warm')
|
|
self.assertEqual(restored.status()['settings']['stt']['mode'],'per_request')
|