diff --git a/payload/user/remote_access/pager-webui/mk8_guard.py b/payload/user/remote_access/pager-webui/mk8_guard.py new file mode 100644 index 0000000..d8728f7 --- /dev/null +++ b/payload/user/remote_access/pager-webui/mk8_guard.py @@ -0,0 +1,35 @@ +"""Boot-time reconciliation of crash-prone PineAP settings.""" +from server import (_apply_uci_wanted, _monitor_down, _raise_monitors, + PINEAPD_SAFE_UCI, device_run) + +WANTED_EXTRA = {'pineapd.@pineapd[0].autossidpool': '0'} +POOL_CLEAR_MAX = 20 +MONITORS = ('wlan0mon', 'wlan1mon') + + +def _pool_size(): + rc, out, err = device_run( + ['uci', 'get', 'pineapd.@ssidpool[0].ssid']) + if rc != 0 or not (out or '').strip(): + return 0 + return len([s for s in out.strip().split('\n') if s]) + + +def reconcile(clear_pool=True): + changed = _apply_uci_wanted(dict(PINEAPD_SAFE_UCI, **WANTED_EXTRA)) + pool_cleared = False + if clear_pool and _pool_size() > POOL_CLEAR_MAX: + device_run(['uci', 'delete', 'pineapd.@ssidpool[0].ssid']) + pool_cleared = True + if changed or pool_cleared: + device_run(['uci', 'commit', 'pineapd']) + raised = _raise_monitors() if any(_monitor_down(m) for m in MONITORS) else [] + return {'changed': changed, 'pool_cleared': pool_cleared, + 'monitors_raised': raised} + + +def guard_report(): + from server import _pending_uci + pending = _pending_uci(dict(PINEAPD_SAFE_UCI, **WANTED_EXTRA)) + return {'in_sync': not pending, 'pending': pending, + 'pool_size': _pool_size()} diff --git a/tests/test_mk8_guard.py b/tests/test_mk8_guard.py new file mode 100644 index 0000000..569b1c9 --- /dev/null +++ b/tests/test_mk8_guard.py @@ -0,0 +1,59 @@ +import os, sys, unittest +sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', 'payload', 'user', 'remote_access', 'pager-webui')) +import server +import mk8_guard + +class GuardTest(unittest.TestCase): + def setUp(self): + self.calls = [] + self.uci = {} + self.pool = 0 + self.mon_up = {'wlan0mon': True, 'wlan1mon': True} + self.old_iface_up = server._iface_up + self.old_server_device_run = server.device_run + self.old_guard_device_run = mk8_guard.device_run + server._iface_up = lambda name: self.mon_up.get(name, True) + + def fake_run(args, timeout=20, input_data=None): + a = list(args) + self.calls.append(a) + if a[:2] == ['uci', 'get']: + key = a[2] + if key == 'pineapd.@ssidpool[0].ssid': + return (0, ''.join('s%d\n' % i for i in range(self.pool)), '') + return (0, self.uci.get(key, '') + '\n', '') + if a[:2] == ['uci', 'set']: + k, _, v = a[2].partition('=') + self.uci[k] = v + if a[:2] == ['uci', 'delete']: + self.pool = 0 + return (0, '', '') + mk8_guard.device_run = fake_run + server.device_run = fake_run + + def tearDown(self): + server._iface_up = self.old_iface_up + server.device_run = self.old_server_device_run + mk8_guard.device_run = self.old_guard_device_run + + def test_applies_all_wanted_when_missing(self): + self.pool = 0 + result = mk8_guard.reconcile(clear_pool=False) + sets = [c[2] for c in self.calls if c[:2] == ['uci', 'set']] + self.assertEqual(len(sets), + len(server.PINEAPD_SAFE_UCI) + len(mk8_guard.WANTED_EXTRA)) + self.assertTrue(result['changed']) + + def test_clears_large_pool_only(self): + self.pool = 25 + result = mk8_guard.reconcile(clear_pool=True) + self.assertTrue(result['pool_cleared']) + self.assertIn(['uci', 'delete', 'pineapd.@ssidpool[0].ssid'], self.calls) + + def test_small_pool_untouched(self): + self.pool = 5 + result = mk8_guard.reconcile(clear_pool=True) + self.assertFalse(result['pool_cleared']) + +if __name__ == '__main__': + unittest.main()