20 lines
1.5 KiB
Python
20 lines
1.5 KiB
Python
import importlib.util,pathlib,unittest
|
|
from unittest.mock import patch
|
|
from types import SimpleNamespace
|
|
s=importlib.util.spec_from_file_location('manager',pathlib.Path(__file__).with_name('gpu1-mode.py'));m=importlib.util.module_from_spec(s);s.loader.exec_module(m)
|
|
class Safety(unittest.TestCase):
|
|
def test_absent_container_case_insensitive(self):
|
|
with patch.object(m.subprocess,'run',return_value=SimpleNamespace(returncode=1,stderr='error: no such object: followup-audio-asr',stdout='')):self.assertEqual(m.container_state()['Status'],'absent')
|
|
def test_daemon_error_not_absent(self):
|
|
with patch.object(m.subprocess,'run',return_value=SimpleNamespace(returncode=1,stderr='permission denied',stdout='')):
|
|
with self.assertRaisesRegex(RuntimeError,'STATE_UNKNOWN'):m.container_state()
|
|
def test_foreign_process_refused(self):
|
|
with self.assertRaisesRegex(RuntimeError,'OTHER_WORKLOAD'):m.ensure_exclusive({1,2},{1})
|
|
def test_owned_allowed(self):m.ensure_exclusive({1},{1})
|
|
def test_empty_allowed(self):m.ensure_exclusive(set(),set())
|
|
def test_idle(self):self.assertEqual(m.idle_values('vllm:num_requests_running{engine="0"} 0\nvllm:num_requests_waiting{engine="0"} 0'),[0,0])
|
|
def test_busy(self):self.assertEqual(m.idle_values('vllm:num_requests_running{engine="0"} 1\nvllm:num_requests_waiting{engine="0"} 2'),[1,2])
|
|
def test_missing_metrics_refused(self):
|
|
with self.assertRaisesRegex(RuntimeError,'METRICS_MISSING'):m.idle_values('')
|
|
if __name__=='__main__':unittest.main()
|