64 lines
3.6 KiB
Python
64 lines
3.6 KiB
Python
import importlib.machinery,importlib.util,json,subprocess,tempfile,unittest
|
|
from pathlib import Path
|
|
|
|
class QuiesceTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.tmp=tempfile.TemporaryDirectory();self.addCleanup(self.tmp.cleanup)
|
|
self.directory=Path(self.tmp.name)
|
|
# the installed hook has no .py suffix (50-svos), so give the loader explicitly; path is relative to this file
|
|
loader=importlib.machinery.SourceFileLoader('svos_quiesce',str(Path(__file__).with_name('50-svos')))
|
|
spec=importlib.util.spec_from_loader('svos_quiesce',loader)
|
|
self.module=importlib.util.module_from_spec(spec);spec.loader.exec_module(self.module)
|
|
self.states={'svos-kernel.service':('active','success'),'svos-interface.service':('inactive','success'),
|
|
'svos-terminal.service':('inactive','success'),'svos-knowledge.service':('inactive','success'),'svos-postoffice.service':('inactive','success'),
|
|
'svos-voice.service':('inactive','success')}
|
|
self.calls=[];self.stop_fail=False;self.start_fail=False;self.dirty_stop=False
|
|
def run(args,**kwargs):
|
|
self.calls.append(args)
|
|
if args[0]=='show':
|
|
state,result=self.states[args[1]]
|
|
return subprocess.CompletedProcess(args,0,f'ActiveState={state}\nResult={result}\n','')
|
|
units=[x for x in args[1:] if not x.startswith('--')]
|
|
if args[0]=='stop':
|
|
self.states[units[0]]=('inactive','timeout' if self.dirty_stop else 'success')
|
|
if self.stop_fail:raise subprocess.TimeoutExpired(args,35)
|
|
for unit in units:self.states[unit]=('inactive','timeout' if self.dirty_stop else 'success')
|
|
elif args[0]=='start':
|
|
if self.start_fail:raise subprocess.CalledProcessError(1,args)
|
|
for unit in units:self.states[unit]=('active','success')
|
|
return subprocess.CompletedProcess(args,0,'','')
|
|
self.q=self.module.Quiesce(self.directory,run=run)
|
|
def test_restore_only_previously_active_services(self):
|
|
self.q.freeze();self.assertEqual(self.states['svos-kernel.service'][0],'inactive')
|
|
self.assertEqual(self.states['svos-interface.service'][0],'inactive')
|
|
self.q.thaw();self.assertEqual(self.states['svos-kernel.service'][0],'active')
|
|
self.assertEqual(self.states['svos-interface.service'][0],'inactive')
|
|
self.assertFalse(self.q.state.exists());self.q.thaw()
|
|
def test_stop_timeout_refuses_freeze_and_schedules_recovery(self):
|
|
self.stop_fail=True
|
|
with self.assertRaises(subprocess.TimeoutExpired):self.q.freeze()
|
|
self.assertEqual(self.states['svos-kernel.service'][0],'active')
|
|
self.assertTrue(self.q.state.exists())
|
|
self.q.thaw();self.assertFalse(self.q.state.exists())
|
|
def test_unclean_stop_is_not_reported_as_consistent(self):
|
|
self.dirty_stop=True
|
|
with self.assertRaises(RuntimeError):self.q.freeze()
|
|
self.assertEqual(self.states['svos-kernel.service'][0],'active')
|
|
def test_failed_thaw_preserves_recovery_record(self):
|
|
self.q.freeze();self.start_fail=True
|
|
with self.assertRaises(subprocess.CalledProcessError):self.q.thaw()
|
|
self.assertTrue(self.q.state.exists());self.start_fail=False
|
|
self.q.thaw();self.assertFalse(self.q.state.exists())
|
|
def test_repeated_freeze_cannot_overwrite_original_running_set(self):
|
|
self.q.freeze();before=self.q.state.read_bytes()
|
|
with self.assertRaises(RuntimeError):self.q.freeze()
|
|
self.assertEqual(self.q.state.read_bytes(),before)
|
|
self.q.thaw();self.assertEqual(self.states['svos-kernel.service'][0],'active')
|
|
def test_transition_refuses_before_any_stop(self):
|
|
self.states['svos-interface.service']=('deactivating','success')
|
|
with self.assertRaises(RuntimeError):self.q.freeze()
|
|
self.assertFalse(self.q.state.exists())
|
|
self.assertFalse(any(x[0]=='stop' for x in self.calls))
|
|
|
|
if __name__=='__main__':unittest.main()
|