99 lines
4.5 KiB
Python
99 lines
4.5 KiB
Python
"""Maintenance handoff: durable choices, queued jobs, natural drain and recovery."""
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import tempfile
|
|
import types
|
|
import unittest
|
|
|
|
SCRIPTS = Path(__file__).resolve().parents[1]
|
|
|
|
|
|
class MaintenanceDrain(unittest.TestCase):
|
|
def setUp(self):
|
|
tmp = tempfile.TemporaryDirectory()
|
|
self.addCleanup(tmp.cleanup)
|
|
self.root = Path(tmp.name)
|
|
self.tx = self.root / 'tx'
|
|
self.tx.mkdir()
|
|
self.ns = {'__name__': 'maintenance_fixture'}
|
|
for part in ('prepare', 'runtime', 'worker'):
|
|
exec(compile((SCRIPTS / f'public_update_{part}.py').read_text(), part, 'exec'), self.ns)
|
|
self.ns.update(TRANSACTIONS=self.root, MAINTENANCE_UNITS=('on.timer', 'off.timer', 'helper.service'))
|
|
self.ns['trusted_open'] = lambda p, directory=False: os.open(p, os.O_RDONLY | os.O_NOFOLLOW | (os.O_DIRECTORY if directory else 0))
|
|
self.clock = 0
|
|
self.sleeps = 0
|
|
self.events = []
|
|
self.helper = [('activating', '17'), ('inactive', '0')]
|
|
self.ns['time'] = types.SimpleNamespace(monotonic=lambda: self.clock, time=lambda: 123, sleep=self.sleep)
|
|
self.ns['command'] = self.command
|
|
self.state = {'id': 'a' * 32, 'runtime_startup_guard': True}
|
|
|
|
def sleep(self, seconds):
|
|
self.clock += seconds
|
|
self.sleeps += 1
|
|
|
|
def command(self, _, action, unit, *args, **kwargs):
|
|
self.events.append((action, unit))
|
|
if action == 'show':
|
|
enabled = 'static' if unit.endswith('.service') else 'disabled'
|
|
if '--value' in args:
|
|
return enabled
|
|
if unit == 'helper.service':
|
|
active, job = self.helper[0]
|
|
if len(self.helper) > 1:
|
|
self.helper.pop(0)
|
|
else:
|
|
active, job = ('active' if unit == 'on.timer' else 'inactive'), '0'
|
|
return f'LoadState=loaded\nActiveState={active}\nUnitFileState={enabled}\nJob={job}\n'
|
|
if action == 'stop':
|
|
self.assertEqual(unit, 'on.timer', 'never kill a running helper or alter an inactive timer')
|
|
saved = json.loads((self.tx / 'state.json').read_text())
|
|
self.assertEqual(saved['maintenance_units']['on.timer']['ActiveState'], 'active')
|
|
self.assertEqual((self.root / 'maintenance-paused').read_text(), self.state['id'])
|
|
return ''
|
|
|
|
def test_active_oneshot_drains_and_resume_preserves_timer_decisions(self):
|
|
state = self.ns['runtime_pause_units'](self.tx, self.state)
|
|
self.assertEqual(self.sleeps, 1)
|
|
self.ns['runtime_resume_units'](state)
|
|
self.assertEqual([e for e in self.events if e[0] == 'start'], [('start', 'on.timer')])
|
|
self.assertFalse((self.root / 'maintenance-paused').exists())
|
|
|
|
def test_inactive_with_queued_job_is_not_drained(self):
|
|
self.helper = [('inactive', '18'), ('activating', '18'), ('inactive', '0')]
|
|
self.ns['runtime_pause_units'](self.tx, self.state)
|
|
self.assertEqual(self.sleeps, 2)
|
|
|
|
def test_timeout_keeps_durable_timer_choices_for_recovery(self):
|
|
self.helper = [('activating', '19')]
|
|
def advance(_):
|
|
self.clock += 180
|
|
self.ns['time'].sleep = advance
|
|
with self.assertRaisesRegex(ValueError, 'did not drain'):
|
|
self.ns['runtime_pause_units'](self.tx, self.state)
|
|
saved = json.loads((self.tx / 'state.json').read_text())
|
|
self.assertEqual(set(saved['maintenance_units']), {'on.timer', 'off.timer'})
|
|
self.ns['runtime_resume_units'](dict(saved, phase='restored-pending-start'))
|
|
self.assertEqual([e for e in self.events if e[0] == 'start'], [('start', 'on.timer')])
|
|
self.assertFalse((self.root / 'maintenance-paused').exists())
|
|
|
|
def test_timer_stop_failure_retains_recoverable_original_choices(self):
|
|
original = self.ns['command']
|
|
def fail(*args, **kwargs):
|
|
result = original(*args, **kwargs)
|
|
if args[1] == 'stop':
|
|
raise RuntimeError('fixture stop failure')
|
|
return result
|
|
self.ns['command'] = fail
|
|
with self.assertRaisesRegex(RuntimeError, 'fixture stop failure'):
|
|
self.ns['runtime_pause_units'](self.tx, self.state)
|
|
saved = json.loads((self.tx / 'state.json').read_text())
|
|
self.ns['command'] = original
|
|
self.ns['runtime_resume_units'](saved)
|
|
self.assertEqual([e for e in self.events if e[0] == 'start'], [('start', 'on.timer')])
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|