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 HealthCheckTest(unittest.TestCase): def setUp(self): self.runs = [] self.ping_ok = True self.ip_link_ok = True self.sigsegvs = 0 self.iface_up = {'wlan0mon': True, 'wlan1mon': True} self.uci_state = {} server._health.update({ 'sigsegv_last': None, 'last_fix': 0.0, 'fixes': 0, 'last_action': None, 'pineap_up': False, 'monitor_fixes': 0}) self.old_iface_up = server._iface_up server._iface_up = lambda name: self.iface_up.get(name, True) def fake_run(args, timeout=20, input_data=None): self.runs.append((list(args), timeout)) a = list(args) if a[0] == 'pidof' and a[1] == 'pineapd': if self.ping_ok: return (0, '12345\n', '') return (1, '', '') if a[0] == 'logread': return (0, 'SIGSEGV\n' * self.sigsegs if hasattr(self, 'sigsegs') else '', '') if a[:3] == ['ip', 'link', 'set']: if self.ip_link_ok: self.iface_up[a[3]] = True return (0, '', '') return (1, '', 'interface unavailable') if a[:2] == ['uci', 'set']: k, _, v = a[2].partition('=') self.uci_state[k] = v if a[:2] == ['uci', 'delete']: for k in list(self.uci_state): if k == a[2] or k.startswith(a[2] + '.'): del self.uci_state[k] if a[:2] == ['uci', 'get']: return (0, self.uci_state.get(a[2], '') + '\n', '') if a[0] == 'uci' and a[1] == 'show': sec = a[2] return (0, ''.join("%s=%s\n" % (k, v) for k, v in self.uci_state.items() if k == sec or k.startswith(sec + '.')), '') return (0, '', '') server.device_run = fake_run def tearDown(self): server._iface_up = self.old_iface_up def test_pineap_up_reports_no_action(self): result = server.health_check() self.assertTrue(result['pineap_up']) self.assertIsNone(result['last_action']) self.assertEqual([r[0] for r in self.runs], [['pidof', 'pineapd']], 'health check must not write to the pineapd socket') def test_pineap_up_repairs_dropped_monitors(self): self.iface_up = {'wlan0mon': False, 'wlan1mon': True} result = server.health_check() self.assertTrue(result['pineap_up']) self.assertEqual(result['last_action'], 'monitor interfaces brought up') self.assertIn(['ip', 'link', 'set', 'wlan0mon', 'up'], [r[0] for r in self.runs]) self.assertEqual(result['monitor_fixes'], 1) def test_down_restarts_pineapd_without_rewriting_uci(self): self.ping_ok = False self.sigsegs = 5 self.uci_state['pineapd.@ssidpool[0].ssid'] = 'QmVlcg==' result = server.health_check() self.assertEqual(result['last_action'], 'pineapd restart') self.assertNotIn('pineapd.@ssidpool[0].disable', self.uci_state) self.assertEqual(self.uci_state['pineapd.@ssidpool[0].ssid'], 'QmVlcg==') self.assertFalse(any(r[0][:2] == ['uci', 'set'] for r in self.runs)) self.assertIn(['/etc/init.d/pineapd', 'restart'], [r[0] for r in self.runs]) self.assertEqual(result['fixes'], 1) def test_down_with_pool_already_disabled_restarts_pineapd(self): self.ping_ok = False self.uci_state['pineapd.@ssidpool[0].disable'] = '1' self.uci_state['pineapd.wlan2mon.disable'] = '1' self.uci_state['pineapd.wlan2mon.hop'] = '0' self.uci_state['pineapd.wlan1mon.bands'] = '5' self.uci_state['pineapd.wlan0mon.bands'] = '2' self.uci_state['pineapd.wlan1mon.hop'] = '0' result = server.health_check() self.assertEqual(result['last_action'], 'pineapd restart') self.assertIn(['/etc/init.d/pineapd', 'restart'], [r[0] for r in self.runs]) def test_down_does_not_stabilize_or_clear_pool(self): self.ping_ok = False self.uci_state['pineapd.@ssidpool[0].disable'] = '1' self.uci_state['pineapd.@ssidpool[0].ssid'] = 'QmVlcg==' result = server.health_check() self.assertEqual(result['last_action'], 'pineapd restart') self.assertEqual(self.uci_state['pineapd.@ssidpool[0].ssid'], 'QmVlcg==') self.assertNotIn('pineapd.wlan2mon.disable', self.uci_state) self.assertNotIn('pineapd.wlan1mon.bands', self.uci_state) def test_down_without_crash_brings_monitors_up(self): self.ping_ok = False self.uci_state['pineapd.@ssidpool[0].disable'] = '1' self.iface_up = {'wlan0mon': True, 'wlan1mon': False} result = server.health_check() self.assertEqual(result['last_action'], 'monitor interfaces brought up') self.assertIn(['ip', 'link', 'set', 'wlan1mon', 'up'], [r[0] for r in self.runs]) def test_down_does_not_disable_pool_regardless_of_sigsegv_history(self): self.ping_ok = False server._health['sigsegv_last'] = 4 result = server.health_check() self.assertEqual(result['last_action'], 'pineapd restart') self.assertNotIn('pineapd.@ssidpool[0].disable', self.uci_state) def test_fix_cooldown_prevents_thrash(self): self.ping_ok = False server._health['last_fix'] = server.time.time() - 30 server.health_check() server.health_check() fixes = [r for r in self.runs if r[0][0] == '/etc/init.d/pineapd'] self.assertEqual(len(fixes), 1, 'cooldown must allow only one restart') def test_health_endpoint_shape(self): server._health['sigsegv_last'] = 7 status, payload = server.h_health(type('C', (), {'query': {}})()) self.assertEqual(status, 200) self.assertEqual(payload['sigsegv_count'], 7) self.assertIn('wlan1mon_up', payload) self.assertIn('pool_disabled', payload) class SupervisorExtrasTest(unittest.TestCase): def runTestWith(self): # helper: reuse existing setUp fake_run pass def test_mem_percent_math(self): import tempfile content = 'MemTotal: 250000 kB\nMemAvailable: 100000 kB\n' path = tempfile.mktemp() open(path, 'w').write(content) self.assertEqual(server._mem_percent(path), 60) def test_health_reports_events_and_counters(self): import mk8_events mk8_events.log_event('restart', msg='x') status, h = server.h_health(None) self.assertEqual(status, 200) self.assertIn('events', h) self.assertIn('boots', h['reliability']) def test_boot_marker_detects_unexpected(self): import mk8_events, tempfile, os marker = tempfile.mktemp() old = server.BOOT_MARKER server.BOOT_MARKER = marker try: open(marker, 'w').write('0') self.assertTrue(server.check_boot_marker()) os.unlink(marker) self.assertFalse(server.check_boot_marker()) finally: server.BOOT_MARKER = old if __name__ == '__main__': unittest.main()