234 lines
9.0 KiB
Python
234 lines
9.0 KiB
Python
import os
|
|
import sys
|
|
import unittest
|
|
|
|
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', 'payload',
|
|
'user', 'remote_access', 'pager-webui'))
|
|
import server
|
|
|
|
|
|
def setUpModule():
|
|
__import__('importlib').reload(server)
|
|
|
|
|
|
class _Ctx(object):
|
|
def __init__(self, body=None):
|
|
self.body = body
|
|
|
|
|
|
class ReliabilityApiTest(unittest.TestCase):
|
|
def setUp(self):
|
|
self.runs = []
|
|
self.old_device_run = server.device_run
|
|
server._health.update({
|
|
'sigsegv_last': None, 'last_fix': 0.0, 'fixes': 0,
|
|
'last_action': None, 'pineap_up': False, 'monitor_fixes': 0})
|
|
|
|
def fake_run(args, timeout=20, input_data=None):
|
|
self.runs.append((list(args), timeout))
|
|
return (0, '', '')
|
|
|
|
server.device_run = fake_run
|
|
|
|
def tearDown(self):
|
|
server.device_run = self.old_device_run
|
|
|
|
def test_h_health_exposes_reliability_feed(self):
|
|
status, h = server.h_health(None)
|
|
self.assertEqual(status, 200)
|
|
for key in ('reliability', 'events', 'guard'):
|
|
self.assertIn(key, h)
|
|
for counter in ('boots', 'unexpected_boots', 'rollbacks',
|
|
'restarts', 'guard_fixes'):
|
|
self.assertIn(counter, h['reliability'])
|
|
self.assertIsInstance(h['events'], list)
|
|
self.assertIn('in_sync', h['guard'])
|
|
self.assertIn('pool_size', h['guard'])
|
|
|
|
def test_check_boot_marker_uses_module_marker_path(self):
|
|
import tempfile
|
|
marker = tempfile.mktemp()
|
|
old = server.BOOT_MARKER
|
|
server.BOOT_MARKER = marker
|
|
try:
|
|
if os.path.exists(marker):
|
|
os.unlink(marker)
|
|
self.assertFalse(server.check_boot_marker())
|
|
self.assertTrue(os.path.exists(marker),
|
|
'check_boot_marker must use server.BOOT_MARKER')
|
|
open(marker, 'w').write('0')
|
|
self.assertTrue(server.check_boot_marker())
|
|
finally:
|
|
if os.path.exists(marker):
|
|
os.unlink(marker)
|
|
server.BOOT_MARKER = old
|
|
|
|
def test_clean_cycle_boot_shutdown_next_boot_not_unexpected(self):
|
|
import tempfile
|
|
import threading
|
|
import mk8_events as events_mod
|
|
marker = tempfile.mktemp()
|
|
old = (server.BOOT_MARKER, server.LIVE_STOP, server.HEALTH_STOP,
|
|
server._recon_hopper_stop, events_mod.mark_boot)
|
|
stops = (threading.Event(), threading.Event(), threading.Event())
|
|
server.BOOT_MARKER = marker
|
|
server.LIVE_STOP, server.HEALTH_STOP, \
|
|
server._recon_hopper_stop = stops
|
|
booted = []
|
|
events_mod.mark_boot = \
|
|
lambda unexpected=False: booted.append(unexpected)
|
|
try:
|
|
# Simulate a previous run's marker left behind: boot is unexpected.
|
|
open(marker, 'w').write('0')
|
|
self.assertTrue(server.check_boot_marker())
|
|
self.assertEqual(booted, [True])
|
|
# Clean shutdown clears the marker...
|
|
self.assertTrue(os.path.exists(marker))
|
|
server._request_shutdown()
|
|
self.assertFalse(os.path.exists(marker),
|
|
'graceful shutdown must clear the boot marker')
|
|
for ev in stops:
|
|
self.assertTrue(ev.is_set())
|
|
# ...so the next boot is clean and re-arms the marker.
|
|
self.assertFalse(server.check_boot_marker())
|
|
self.assertEqual(booted[-1], False)
|
|
self.assertTrue(os.path.exists(marker))
|
|
# _clear_boot_marker is best-effort on missing/None markers.
|
|
server._clear_boot_marker()
|
|
self.assertFalse(os.path.exists(marker))
|
|
server.BOOT_MARKER = None
|
|
server._clear_boot_marker()
|
|
finally:
|
|
(server.BOOT_MARKER, server.LIVE_STOP, server.HEALTH_STOP,
|
|
server._recon_hopper_stop, events_mod.mark_boot) = old
|
|
if os.path.exists(marker):
|
|
os.unlink(marker)
|
|
|
|
def test_profile_routes_registered(self):
|
|
handler, _ = server.ROUTER.dispatch('GET',
|
|
'/api/reliability/profiles')
|
|
self.assertEqual(handler, server.h_profiles_get)
|
|
handler, _ = server.ROUTER.dispatch('POST', '/api/reliability/profile')
|
|
self.assertEqual(handler, server.h_profile_save)
|
|
handler, _ = server.ROUTER.dispatch('POST', '/api/reliability/restore')
|
|
self.assertEqual(handler, server.h_profile_restore)
|
|
handler, _ = server.ROUTER.dispatch('GET', '/api/reliability/nope')
|
|
self.assertIsNone(handler)
|
|
|
|
def test_h_profiles_get_lists_profiles(self):
|
|
import mk8_profiles
|
|
old = mk8_profiles.list_profiles
|
|
mk8_profiles.list_profiles = lambda: ['a', 'b']
|
|
try:
|
|
status, data = server.h_profiles_get(None)
|
|
finally:
|
|
mk8_profiles.list_profiles = old
|
|
self.assertEqual(status, 200)
|
|
self.assertEqual(data, {'profiles': ['a', 'b']})
|
|
|
|
def test_h_profile_save_validates_saves_and_journals(self):
|
|
import mk8_events
|
|
import mk8_profiles
|
|
calls = {'snapshots': []}
|
|
events = []
|
|
old_snapshot, old_log = mk8_profiles.snapshot, mk8_events.log_event
|
|
|
|
def fake_snapshot(name):
|
|
if not all(c.isalnum() or c in '._-' for c in name) \
|
|
or name in ('.', '..') or len(name) > 64:
|
|
raise ValueError('invalid profile name')
|
|
calls['snapshots'].append(name)
|
|
return True
|
|
|
|
def fake_log(kind, **kw):
|
|
events.append((kind, kw))
|
|
|
|
mk8_profiles.snapshot = fake_snapshot
|
|
mk8_events.log_event = fake_log
|
|
try:
|
|
status, data = server.h_profile_save(_Ctx({'name': ' pre-x-1 '}))
|
|
self.assertEqual(status, 200)
|
|
self.assertEqual(data, {'ok': True})
|
|
self.assertEqual(calls['snapshots'], ['pre-x-1'])
|
|
self.assertEqual(events[-1][0], 'profile_save')
|
|
|
|
status, data = server.h_profile_save(_Ctx({'name': ' '}))
|
|
self.assertEqual(status, 400)
|
|
self.assertIn('error', data)
|
|
|
|
status, data = server.h_profile_save(_Ctx({'name': '../evil'}))
|
|
self.assertEqual(status, 400)
|
|
self.assertIn('error', data)
|
|
self.assertEqual(calls['snapshots'], ['pre-x-1'])
|
|
|
|
status, data = server.h_profile_save(_Ctx({}))
|
|
self.assertEqual(status, 400)
|
|
|
|
def failed_snapshot(name):
|
|
calls['snapshots'].append(name)
|
|
return False
|
|
mk8_profiles.snapshot = failed_snapshot
|
|
status, data = server.h_profile_save(_Ctx({'name': 'p2'}))
|
|
self.assertEqual(status, 502)
|
|
self.assertFalse(data['ok'])
|
|
self.assertEqual(events[-1][0], 'profile_save')
|
|
self.assertEqual(events[-1][1].get('sev'), 'warn')
|
|
finally:
|
|
mk8_profiles.snapshot = old_snapshot
|
|
mk8_events.log_event = old_log
|
|
|
|
def test_h_profile_restore_gated_reload_journal(self):
|
|
import mk8_gate
|
|
import mk8_events
|
|
import mk8_profiles
|
|
calls = {'gate': [], 'events': []}
|
|
olds = (mk8_gate.enter, mk8_profiles.restore, mk8_events.log_event)
|
|
|
|
def fake_enter(op):
|
|
calls['gate'].append(op)
|
|
return 'snap-1'
|
|
|
|
def fake_restore(name):
|
|
calls['restored'] = name
|
|
return {'ok': True, 'restored': ['wireless']}
|
|
|
|
def fake_log(kind, **kw):
|
|
calls['events'].append((kind, kw))
|
|
|
|
mk8_gate.enter = fake_enter
|
|
mk8_profiles.restore = fake_restore
|
|
mk8_events.log_event = fake_log
|
|
try:
|
|
status, result = server.h_profile_restore(_Ctx({'name': 'p1'}))
|
|
self.assertEqual(status, 200)
|
|
self.assertEqual(result, {'ok': True, 'restored': ['wireless']})
|
|
self.assertEqual(calls['gate'], ['restore_profile'])
|
|
self.assertEqual(calls['restored'], 'p1')
|
|
reloads = [r for r in self.runs if r[0][:2] == ['wifi', 'reload']]
|
|
self.assertEqual(len(reloads), 1)
|
|
self.assertEqual(calls['events'][-1][0], 'profile_restore')
|
|
|
|
status, result = server.h_profile_restore(
|
|
_Ctx({'name': 'missing'}))
|
|
|
|
def missing_restore(name):
|
|
calls['restored'] = name
|
|
return {'ok': False, 'restored': [], 'error': 'not found'}
|
|
mk8_profiles.restore = missing_restore
|
|
status, result = server.h_profile_restore(
|
|
_Ctx({'name': 'missing'}))
|
|
self.assertEqual(status, 502)
|
|
self.assertFalse(result['ok'])
|
|
self.assertEqual(calls['events'][-1][0], 'profile_restore')
|
|
self.assertEqual(calls['events'][-1][1].get('sev'), 'warn')
|
|
|
|
status, result = server.h_profile_restore(_Ctx({'name': ''}))
|
|
self.assertEqual(status, 400)
|
|
finally:
|
|
(mk8_gate.enter, mk8_profiles.restore,
|
|
mk8_events.log_event) = olds
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|