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 import mk8_guard mk8_guard._GR_CACHE['data'] = None 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) # post-restore convergence parks the factory-enabled dummy STA parked = [r for r in self.runs if r[0][:3] == ['uci', 'set', 'wireless.dummy_radio0.disabled=1']] self.assertEqual(len(parked), 1) downs = [r for r in self.runs if r[0][:4] == ['ip', 'link', 'set', 'wlan0']] self.assertTrue(downs) raises = [r for r in self.runs if r[0][:4] == ['ip', 'link', 'set', 'wlan0mon']] self.assertTrue(raises) 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()