feat(reliability): UCI profile snapshot store

payload/user/remote_access/pager-webui/mk8_profiles.py

@@ -0,0 +1,80 @@
+"""UCI profile snapshots under /mmc/mk8/profiles/<name>/{pineapd,wireless,network}"""
+import os, time
+
+PROFILES_DIR = '/mmc/mk8/profiles'
+CONFIGS = ('pineapd', 'wireless', 'network')
+
+
+def run_cmd(args, timeout=20, input_data=None):
+    """Lazy import avoids a circular import with server.py; tests monkeypatch."""
+    from server import device_run
+    return device_run(args, timeout=timeout, input_data=input_data)
+
+
+def _path(name):
+    return os.path.join(PROFILES_DIR, name)
+
+
+def snapshot(name):
+    dest = _path(name)
+    try:
+        os.makedirs(dest, exist_ok=True)
+        wrote = False
+        for cfg in CONFIGS:
+            rc, out, err = run_cmd(['uci', 'export', cfg])
+            if rc != 0 or not (out or '').strip():
+                continue
+            with open(os.path.join(dest, cfg + '.uci'), 'w') as f:
+                f.write(out)
+            wrote = True
+        return wrote
+    except OSError:
+        return False
+
+
+def auto_name(op):
+    return 'pre-%s-%d' % (op, int(time.time()))
+
+
+def list_profiles():
+    try:
+        return sorted(d for d in os.listdir(PROFILES_DIR)
+                      if os.path.isdir(_path(d)))
+    except OSError:
+        return []
+
+
+def delete(name):
+    import shutil
+    shutil.rmtree(_path(name), ignore_errors=True)
+
+
+def restore(name):
+    """Restore configs then commit once per config. Caller runs wifi reload
+    / service restart as appropriate for the operation."""
+    src = _path(name)
+    restored = []
+    if not os.path.isdir(src):
+        return {'ok': False, 'restored': [], 'error': 'profile not found'}
+    for cfg in CONFIGS:
+        fpath = os.path.join(src, cfg + '.uci')
+        if not os.path.isfile(fpath):
+            continue
+        with open(fpath) as f:
+            text = f.read()
+        rc, _, err = run_cmd(['uci', 'import', cfg], input_data=text)
+        if rc != 0:
+            return {'ok': False, 'restored': restored,
+                    'error': 'import failed'}
+        run_cmd(['uci', 'commit', cfg])
+        restored.append(cfg)
+    return {'ok': True, 'restored': restored}
+
+
+LASTKNOWN_GOOD = 'lastknown-good'
+
+
+def promote_lastknown_good():
+    """Replace the lastknown-good profile with the live config."""
+    delete(LASTKNOWN_GOOD)
+    return snapshot(LASTKNOWN_GOOD)

tests/test_mk8_profiles.py

@@ -0,0 +1,61 @@
+import os, sys, tempfile, unittest
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', 'payload', 'user', 'remote_access', 'pager-webui'))
+import mk8_profiles
+
+CONFIGS = ('pineapd', 'wireless', 'network')
+
+class ProfilesTest(unittest.TestCase):
+    def setUp(self):
+        self.dir = tempfile.mkdtemp()
+        mk8_profiles.PROFILES_DIR = os.path.join(self.dir, 'profiles')
+        self.state = {'pineapd': 'config pineapd\n\toption x y\n',
+                      'wireless': 'config wireless\n',
+                      'network': 'config network\n'}
+        def fake_run(args, timeout=20, input_data=None):
+            a = list(args)
+            if a[:2] == ['uci', 'import']:
+                self.imports = getattr(self, 'imports', [])
+                self.imports.append((a[2], input_data))
+                return (0, '', '')
+            if a[:2] == ['uci', 'export']:
+                return (0, self.state.get(a[2], ''), '')
+            if a[:2] == ['uci', 'commit']:
+                return (0, '', '')
+            return (0, '', '')
+        self.runs = []
+        mk8_profiles.run_cmd = lambda args, timeout=20, input_data=None: (
+            self.runs.append(list(args)) or fake_run(args, timeout, input_data))
+
+    def test_snapshot_and_list(self):
+        self.assertTrue(mk8_profiles.snapshot('testprof'))
+        self.assertIn('testprof', mk8_profiles.list_profiles())
+
+    def test_restore_issues_import_per_config(self):
+        mk8_profiles.snapshot('p1')
+        ok = mk8_profiles.restore('p1')
+        self.assertTrue(ok['ok'])
+        imported = [r for r in self.runs if r[:2] == ['uci', 'import']]
+        self.assertEqual(len(imported), len(CONFIGS))
+        commits = [r for r in self.runs if r[:2] == ['uci', 'commit']]
+        self.assertGreaterEqual(len(commits), 1)
+        self.assertIn(('pineapd', 'config pineapd\n\toption x y\n'),
+                      self.imports)
+        self.assertEqual(ok['restored'], list(CONFIGS))
+
+    def test_auto_name_format(self):
+        name = mk8_profiles.auto_name('client_connect')
+        self.assertTrue(name.startswith('pre-client_connect-'))
+
+    def test_promote_lastknown_good(self):
+        self.assertTrue(mk8_profiles.promote_lastknown_good())
+        self.assertIn(mk8_profiles.LASTKNOWN_GOOD,
+                      mk8_profiles.list_profiles())
+        again = mk8_profiles.promote_lastknown_good()
+        self.assertTrue(again)
+
+    def test_snapshot_false_when_nothing_written(self):
+        self.state = {}
+        self.assertFalse(mk8_profiles.snapshot('empty'))
+
+if __name__ == '__main__':
+    unittest.main()