sbb-remote 0.10.0: config-get/config-set; GUI watches its config

Keystone's settings pages edit both configs. config-get returns the VM
config text, the defaults and the path; config-set validates new text
with the same check() as load_config, keeps .bak and replaces the file
atomically. Both run before load_config, so a broken config can still be
read and fixed. The GUI reloads its config when the file changes,
re-adding the watch after an atomic write replaces the inode.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

2 files changed, 70 insertions, 0 deletions

diff --git a/tests/test_gui.py b/tests/test_gui.py
index 52e4ea7..c89dc86 100644
--- a/tests/test_gui.py
+++ b/tests/test_gui.py
@@ -628,6 +628,25 @@ class SmokeTest(unittest.TestCase):
         w.dashboard.jobs.clicked.emit()
         self.assertIs(w.tabs.currentWidget(), w.jobs_tab)
 
+    def test_config_file_watched(self):
+        tmp = tempfile.mkdtemp()
+        self.addCleanup(shutil.rmtree, tmp)
+        old = os.environ.get("XDG_CONFIG_HOME")
+        os.environ["XDG_CONFIG_HOME"] = tmp
+        self.addCleanup(lambda: os.environ.__setitem__("XDG_CONFIG_HOME", old) if old
+                        else os.environ.pop("XDG_CONFIG_HOME"))
+        os.makedirs(os.path.dirname(config.path()))
+
+        def save(jobs):  # an atomic write, as Keystone's settings page does: a new inode
+            with open(config.path() + ".tmp", "w") as f:
+                f.write(f"[refresh]\njobs = {jobs}\n")
+            os.replace(config.path() + ".tmp", config.path())
+        save(5)
+        w, _remote = self.window()
+        for jobs in (7, 9):  # the second one proves the path was watched again
+            save(jobs)
+            self.wait(lambda: w.cfg["refresh"]["jobs"] == jobs)
+
     def wait(self, cond, secs=5):
         end = time.time() + secs
         while not cond() and time.time() < end:
diff --git a/tests/test_remote.py b/tests/test_remote.py
index 7288828..3ac584a 100644
--- a/tests/test_remote.py
+++ b/tests/test_remote.py
@@ -570,6 +570,57 @@ class PhantomTest(Base):
             sbb.KINDS["fix-current"](self.cfg, ["x"])
 
 
+class ConfigCommandsTest(Base):
+    def setUp(self):
+        super().setUp()
+        self.path = os.environ["SBB_CONFIG"]
+        with open(self.path) as f:
+            self.old = f.read()
+
+    def set(self, text):
+        stdin = io.TextIOWrapper(io.BytesIO(text.encode()))
+        with mock.patch.object(sbb.sys, "stdin", stdin):
+            return self.run_main("config-set")
+
+    def text(self):
+        with open(self.path) as f:
+            return f.read()
+
+    def test_set_accepted(self):
+        new = self.old + 'overlays = ["a"]\n'
+        self.assertEqual(self.set(new), (0, {"ok": True}))
+        self.assertEqual(self.text(), new)
+        with open(self.path + ".bak") as f:
+            self.assertEqual(f.read(), self.old)
+        self.assertEqual(sbb.load_config()["overlays"], ["a"])
+
+    def test_set_refused(self):
+        for text, why in (("state_dir = \n", "config"), ("bogus = 1\n", "bogus: unknown key"),
+                          ("state_dir = 1\n", "state_dir: expected str"),
+                          ("[manifests]\nx = 1\n", "manifests: expected strings")):
+            rc, data = self.set(text)
+            self.assertEqual(rc, 1, text)
+            self.assertIn(why, data["error"])
+            self.assertEqual(self.text(), self.old)
+            self.assertFalse(os.path.exists(self.path + ".bak"))
+
+    def test_get(self):
+        self.assertEqual(self.run_main("config-get"),
+                         (0, {"text": self.old, "defaults": sbb.DEFAULTS, "path": self.path}))
+
+    def test_get_missing(self):
+        os.remove(self.path)
+        self.assertEqual(self.run_main("config-get")[1]["text"], "")
+
+    def test_broken_config_can_be_read_and_fixed(self):
+        with open(self.path, "w") as f:
+            f.write("bogus = 1\n")
+        self.assertEqual(self.run_main("version")[0], 1)
+        self.assertEqual(self.run_main("config-get")[1]["text"], "bogus = 1\n")
+        self.assertEqual(self.set(self.old), (0, {"ok": True}))
+        self.assertEqual(self.run_main("version")[0], 0)
+
+
 class MainTest(Base):
     def test_unknown_command(self):
         rc, data = self.run_main("nope")