feat(reliability): boot-time UCI reconciler

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()}

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()