Mark-VIII/tests/test_health.py

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_with_growing_sigsegv_disables_pool(self):
        self.ping_ok = False
        self.sigsegs = 5
        result = server.health_check()
        self.assertIn('pool broadcast disabled', result['last_action'])
        self.assertEqual(self.uci_state['pineapd.@ssidpool[0].disable'], '1')
        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_stabilizes_known_crash_sources(self):
        self.ping_ok = False
        self.uci_state['pineapd.@ssidpool[0].disable'] = '1'
        result = server.health_check()
        self.assertEqual(self.uci_state['pineapd.wlan2mon.disable'], '1')
        self.assertEqual(self.uci_state['pineapd.wlan1mon.bands'], '5')
        self.assertIn('stabilized', result['last_action'])

    def test_down_clears_refilled_pool_list(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.assertNotIn('pineapd.@ssidpool[0].ssid', self.uci_state)
        self.assertIn('pool-list cleared', result['last_action'])

    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_disables_pool_regardless_of_sigsegv_history(self):
        self.ping_ok = False
        server._health['sigsegv_last'] = 4
        result = server.health_check()
        self.assertIn('SSID pool broadcast disabled', result['last_action'])
        self.assertEqual(self.uci_state['pineapd.@ssidpool[0].disable'], '1')

    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)


if __name__ == '__main__':
    unittest.main()