247 lines
10 KiB
Python
247 lines
10 KiB
Python
import os
|
|
import sys
|
|
import unittest
|
|
from unittest import mock
|
|
|
|
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_reparks_resurrected_dummy_sta(self):
|
|
server._health['ticks'] = server.HEALTH_STA_PARK_INTERVAL - 1
|
|
with mock.patch.object(
|
|
server, '_sta_uplink_enabled', return_value=True), \
|
|
mock.patch.object(
|
|
server, '_park_dummy_sta') as park, \
|
|
mock.patch.object(
|
|
server, '_raise_monitors', return_value=[]), \
|
|
mock.patch('mk8_guard.reconcile',
|
|
return_value={'changed': [], 'pool_cleared': False,
|
|
'monitors_raised': []}) as rec:
|
|
h = server.health_check()
|
|
park.assert_called_once()
|
|
rec.assert_called_once()
|
|
self.assertEqual(server._health['ticks'],
|
|
server.HEALTH_STA_PARK_INTERVAL)
|
|
# not on interval ticks: no re-park, no reconcile
|
|
server._health['ticks'] = 1
|
|
with mock.patch.object(
|
|
server, '_sta_uplink_enabled', return_value=True), \
|
|
mock.patch.object(
|
|
server, '_park_dummy_sta') as park, \
|
|
mock.patch('mk8_guard.reconcile') as rec:
|
|
server.health_check()
|
|
park.assert_not_called()
|
|
rec.assert_not_called()
|
|
|
|
def test_health_reconcile_journals_changed_keys(self):
|
|
import mk8_events
|
|
events = []
|
|
old_log = mk8_events.log_event
|
|
old_run = server.device_run
|
|
old_iface = server._iface_up
|
|
|
|
def fake_run(args, timeout=20, input_data=None):
|
|
a = list(args)
|
|
if a[:2] == ['pidof', 'pineapd']:
|
|
return (0, '12345\n', '')
|
|
if a[:2] == ['ip', 'link', 'show']:
|
|
return (0, '4: wlan0mon: <UP> state unknown', '')
|
|
return (0, '', '')
|
|
|
|
mk8_events.log_event = lambda kind, **kw: events.append((kind, kw))
|
|
server.device_run = fake_run
|
|
server._iface_up = lambda name: True
|
|
old_ticks = server._health.get('ticks')
|
|
try:
|
|
server._health['ticks'] = server.HEALTH_STA_PARK_INTERVAL - 1
|
|
with mock.patch.object(server, '_sta_uplink_enabled',
|
|
return_value=False), \
|
|
mock.patch.object(server, '_raise_monitors',
|
|
return_value=[]), \
|
|
mock.patch('mk8_guard.reconcile',
|
|
return_value={'changed':
|
|
['pineapd.@pineapd[0].autossidpool'],
|
|
'pool_cleared': False,
|
|
'monitors_raised': []}):
|
|
server.health_check()
|
|
self.assertTrue(any(k == 'guard_fix' and 'autossidpool' in kw.get('msg', '')
|
|
for k, kw in events))
|
|
finally:
|
|
mk8_events.log_event = old_log
|
|
server.device_run = old_run
|
|
server._iface_up = old_iface
|
|
if old_ticks is None:
|
|
server._health.pop('ticks', None)
|
|
else:
|
|
server._health['ticks'] = old_ticks
|
|
|
|
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()
|