fix(configurator): write printer.cfg atomically at boot

BlocksScreen/helper_methods.py

@@ -12,7 +12,9 @@
 import math
 import os
 import pathlib
+import shutil
 import struct
+import tempfile
 
 logger = logging.getLogger(__name__)
 
@@ -357,6 +359,32 @@ def check_file_on_path(
     return (pathlib.Path(path) / filename).exists()
 
 
+def atomic_write_text(path: pathlib.Path, text: str) -> None:
+    """Write via temp file + rename: a power cut never truncates it; symlinks kept."""
+    target = path.resolve()
+    fd, tmp = tempfile.mkstemp(dir=target.parent, prefix=f".{target.name}.")
+    try:
+        with os.fdopen(fd, "w", encoding="utf-8") as f:
+            f.write(text)
+            f.flush()
+            os.fsync(f.fileno())
+        if target.exists():
+            shutil.copymode(target, tmp)
+        os.replace(tmp, target)
+    except BaseException:
+        try:
+            os.unlink(tmp)
+        except OSError:
+            pass
+        raise
+    # The rename survives a power cut only once the directory entry is on disk.
+    dir_fd = os.open(target.parent, os.O_RDONLY)
+    try:
+        os.fsync(dir_fd)
+    finally:
+        os.close(dir_fd)
+
+
 def get_file_name(filename: str | None) -> str:
     """Extract the basename from a file path (handles ``/`` and ``\\``).
 

BlocksScreen/tools/configuration_manager/configurator.py

@@ -12,6 +12,7 @@
 from typing import Literal
 
 from devices.discovery.serial_devices import SerialScanner
+from helper_methods import atomic_write_text
 
 HOME = pathlib.Path.home()
 CONFIG_REPO: pathlib.Path = HOME / "RF50-Klipper"
@@ -149,9 +150,15 @@ def copy_file_simple(orig: pathlib.Path | str, dest: pathlib.Path | str):
         dest = pathlib.Path(dest)
     if not orig.is_file() or not orig.exists():
         raise FileNotFoundError("File %s not found" % orig.as_posix())
-    res = shutil.copy2(orig, dest, follow_symlinks=False)
+    if dest.is_dir():
+        dest = dest / orig.name
+    existed = dest.exists()
+    # Overwritten in place, a power cut mid-copy leaves Klipper a truncated config.
+    atomic_write_text(dest, orig.read_text(encoding="utf-8"))
+    if not existed:
+        shutil.copymode(orig, dest)
     _logger.info("Copied file %s to destination %s" % (orig.name, dest))
-    return res
+    return str(dest)
 
 
 class ConfigManager:
@@ -609,7 +616,7 @@ def merge_cfg(
 
             merged = self._fix_beacon_serial(merged)
             if _tfl != merged:
-                target_file.write_text(merged, encoding="utf-8")
+                atomic_write_text(target_file, merged)
             return True
         except Exception as e:
             _logger.error("Caught exception while merging: %s" % e)

BlocksScreen/tools/configuration_manager/device_configurator.py

@@ -20,13 +20,11 @@
 
 import json
 import logging
-import os
 import pathlib
-import shutil
-import tempfile
 from collections.abc import Iterable, Sequence
 
 from devices.discovery.serial_devices import Device, SerialScanner
+from helper_methods import atomic_write_text
 
 from .configurator import ConfigManager
 from .device_profiles import (
@@ -47,32 +45,6 @@
 CONSENT_PATH = pathlib.Path.home() / ".config" / "blockscreen" / "device_consent.json"
 
 
-def atomic_write_text(path: pathlib.Path, text: str) -> None:
-    """Replace path's content in one step: readers see the old or the new file.
-
-    Klipper may read printer.cfg at any moment (its own start, a restart), and
-    a power cut mid-write must not leave a truncated file - the next boot's
-    merge would then rebuild it from the repo and lose the SAVE_CONFIG block.
-    Symlinks are followed so the link itself is kept.
-    """
-    target = path.resolve()
-    fd, tmp = tempfile.mkstemp(dir=target.parent, prefix=f".{target.name}.")
-    try:
-        with os.fdopen(fd, "w", encoding="utf-8") as f:
-            f.write(text)
-            f.flush()
-            os.fsync(f.fileno())
-        if target.exists():
-            shutil.copymode(target, tmp)
-        os.replace(tmp, target)
-    except BaseException:
-        try:
-            os.unlink(tmp)
-        except OSError:
-            pass
-        raise
-
-
 class DeviceConsent:
     """Profile names the user accepted, kept across reboots."""
 

tests/util/test_configurator_unit.py

@@ -261,6 +261,26 @@ def test_copies_to_file_path(self, tmp_path):
         copy_file_simple(src, dest / "renamed.cfg")
         assert (dest / "renamed.cfg").is_file()
 
+    def test_new_file_keeps_source_mode(self, tmp_path):
+        src = _write(tmp_path / "src.cfg", "data")
+        os.chmod(src, 0o644)
+        dest = tmp_path / "new.cfg"
+        copy_file_simple(src, dest)
+        assert dest.stat().st_mode & 0o777 == 0o644
+
+    def test_failed_overwrite_keeps_old_file(self, tmp_path, monkeypatch):
+        src = _write(tmp_path / "src.cfg", "new")
+        dest = _write(tmp_path / "dest.cfg", "old")
+
+        def power_cut(*_):
+            raise OSError("power cut")
+
+        monkeypatch.setattr(os, "replace", power_cut)
+        with pytest.raises(OSError):
+            copy_file_simple(src, dest)
+        assert dest.read_text() == "old"
+        assert not list(tmp_path.glob(".dest.cfg.*"))
+
 
 # =========================================================================
 # _build_file_index
@@ -585,6 +605,20 @@ def test_target_has_marker_preserved(self, tmp_path):
         assert SV_CONFIG_MARKER in merged
         assert "#*# saved_value: 123" in merged
 
+    def test_failed_write_keeps_saved_config(self, tmp_path, monkeypatch):
+        mgr = _make_manager(tmp_path)
+        src = _write(tmp_path / "src_printer.cfg", "[a]\nx: 1\n[b]\ny: 2\n")
+        old = "[a]\nx: 1\n" + SV_CONFIG_MARKER + "\n#*# saved_value: 123\n"
+        tgt = _write(tmp_path / "tgt_printer.cfg", old)
+
+        def power_cut(*_):
+            raise OSError("power cut")
+
+        monkeypatch.setattr(os, "replace", power_cut)
+        assert mgr.merge_cfg(src, target_file=tgt, marker=SV_CONFIG_MARKER) is False
+        assert tgt.read_text() == old
+        assert not list(tmp_path.glob(".tgt_printer.cfg.*"))
+
     def test_target_no_marker_copies_source(self, tmp_path):
         mgr = _make_manager(tmp_path)
         src = _write(

tests/util/test_device_configurator_unit.py

@@ -289,7 +289,7 @@ def test_failed_write_leaves_file_untouched(self, tmp_path, monkeypatch):
         def fail(*_):
             raise OSError("disk full")
 
-        monkeypatch.setattr(dc.os, "replace", fail)
+        monkeypatch.setattr(os, "replace", fail)
         with pytest.raises(OSError):
             dc.atomic_write_text(target, "new\n")
         assert target.read_text() == "old\n"