[master] Make host file helpers report failure instead of silently patching

core/libs/commonwealth/src/commonwealth/utils/commands.py

@@ -10,6 +10,10 @@ class KeyNotFound(Exception):
     """Raised when the SSH key is not found."""
 
 
+class HostFileError(Exception):
+    """Raised when a host file cannot be read or written."""
+
+
 def run_command_with_password(command: str, check: bool = True) -> "subprocess.CompletedProcess['str']":
     # attempt to run the command with sshpass
     # used as a fallback if the ssh key is not found
@@ -131,8 +135,10 @@ def upload_file_with_ssh_key(source: str, destination: str, check: bool = True)
 
 
 def load_file(file_name: str) -> str:
-    command = f'cat "{file_name}"'
-    return run_command(command, False).stdout
+    result = run_command(f'cat "{file_name}"', False)
+    if result.returncode != 0:
+        raise HostFileError(f"Failed to read {file_name}: {result.stderr}")
+    return result.stdout
 
 
 def upload_file(file_content: str, destination: str, check: bool = True) -> "subprocess.CompletedProcess['str']":
@@ -147,22 +153,30 @@ def upload_file(file_content: str, destination: str, check: bool = True) -> "sub
         logger.warning("SSH key not found, falling back to password authentication")
         ret = upload_file_with_password(temp_file_in_container, temp_file_in_host, check)
     logger.debug(ret)
-    if ret.returncode == 0:
-        run_command(f"sudo mv {temp_file_in_host} {destination}")
-    else:
+    if ret.returncode != 0:
         logger.error(f"Failed to upload file: {ret.stderr}")
+        return ret
+    moved = run_command(f"sudo mv {temp_file_in_host} {destination}", False)
+    if moved.returncode != 0:
+        logger.error(f"Failed to install uploaded file at {destination}: {moved.stderr}")
+        return moved
     return ret
 
 
 def locate_file(candidates: List[str]) -> Optional[str]:
     # first match will return
     command = f"find {' '.join(candidates)} -type f -print -quit"
-    return run_command(command, False, log_output=False).stdout.strip()
+    found = run_command(command, False, log_output=False).stdout.strip()
+    return found or None
 
 
 def save_file(file_name: str, file_content: str, backup_identifier: str, ensure_newline: bool = True) -> None:
     if ensure_newline and not file_content.endswith("\n"):
         file_content += "\n"
     command = f'sudo cp "{file_name}" "{file_name}.{backup_identifier}.bak"'
-    run_command(command, False)
-    upload_file(file_content, file_name, False)
+    backup = run_command(command, False)
+    if backup.returncode != 0:
+        logger.warning(f"Failed to backup {file_name}: {backup.stderr}")
+    result = upload_file(file_content, file_name, False)
+    if result.returncode != 0:
+        raise HostFileError(f"Failed to save {file_name}: {result.stderr}")

core/libs/commonwealth/src/commonwealth/utils/general.py

@@ -9,7 +9,7 @@
 from typing import Any, AsyncGenerator
 
 import psutil
-from commonwealth.utils.commands import load_file
+from commonwealth.utils.commands import HostFileError, load_file
 from commonwealth.utils.decorators import temporary_cache
 from loguru import logger
 
@@ -47,7 +47,10 @@ def get_cpu_type() -> CpuType:
 
 @cache
 def get_host_os() -> HostOs:
-    os_release = load_file("/etc/os-release")
+    try:
+        os_release = load_file("/etc/os-release")
+    except HostFileError:
+        return HostOs.Other
     if "bookworm" in os_release.lower():
         return HostOs.Bookworm
     if "bullseye" in os_release.lower():

core/libs/commonwealth/src/commonwealth/utils/tests/test_commands.py

@@ -0,0 +1,71 @@
+import subprocess
+from unittest.mock import mock_open, patch
+
+import pytest
+
+from .. import commands
+
+
+def _result(returncode: int, stdout: str = "", stderr: str = "") -> subprocess.CompletedProcess[str]:
+    return subprocess.CompletedProcess(args=[], returncode=returncode, stdout=stdout, stderr=stderr)
+
+
+def test_load_file_returns_contents_when_cat_succeeds(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr(commands, "run_command", lambda *_args, **_kwargs: _result(0, stdout="dtoverlay=spi0-led\n"))
+    assert commands.load_file("/boot/config.txt") == "dtoverlay=spi0-led\n"
+
+
+def test_load_file_returns_empty_string_for_empty_file(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr(commands, "run_command", lambda *_args, **_kwargs: _result(0, stdout=""))
+    assert commands.load_file("/boot/config.txt") == ""
+
+
+def test_load_file_raises_when_cat_fails(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr(
+        commands, "run_command", lambda *_args, **_kwargs: _result(1, stdout="", stderr="cat: No such file")
+    )
+    with pytest.raises(commands.HostFileError, match="Failed to read /boot/config.txt"):
+        commands.load_file("/boot/config.txt")
+
+
+def test_locate_file_returns_first_match(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr(
+        commands, "run_command", lambda *_args, **_kwargs: _result(0, stdout="/boot/firmware/config.txt\n")
+    )
+    assert commands.locate_file(["/boot/firmware/config.txt", "/boot/config.txt"]) == "/boot/firmware/config.txt"
+
+
+def test_locate_file_returns_none_when_find_prints_nothing(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr(commands, "run_command", lambda *_args, **_kwargs: _result(1, stdout="", stderr="No such file"))
+    assert commands.locate_file(["/missing/a", "/missing/b"]) is None
+
+
+def test_locate_file_returns_match_when_find_exits_nonzero(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr(
+        commands,
+        "run_command",
+        lambda *_args, **_kwargs: _result(1, stdout="/boot/config.txt\n", stderr="No such file"),
+    )
+    assert commands.locate_file(["/boot/firmware/config.txt", "/boot/config.txt"]) == "/boot/config.txt"
+
+
+def test_save_file_raises_when_upload_fails(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr(commands, "run_command", lambda *_args, **_kwargs: _result(0))
+    monkeypatch.setattr(commands, "upload_file", lambda *_args, **_kwargs: _result(1, stderr="Permission denied"))
+    with pytest.raises(commands.HostFileError, match="Failed to save /boot/config.txt"):
+        commands.save_file("/boot/config.txt", "[pi4]\n", "before_test")
+
+
+def test_save_file_succeeds_when_upload_succeeds(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr(commands, "run_command", lambda *_args, **_kwargs: _result(0))
+    monkeypatch.setattr(commands, "upload_file", lambda *_args, **_kwargs: _result(0))
+    commands.save_file("/boot/config.txt", "[pi4]\n", "before_test")
+
+
+def test_upload_file_returns_mv_failure(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr(commands, "upload_file_with_ssh_key", lambda *_args, **_kwargs: _result(0))
+    monkeypatch.setattr(commands, "run_command", lambda *_args, **_kwargs: _result(1, stderr="Read-only file system"))
+    with patch("builtins.open", mock_open()):
+        result = commands.upload_file("[pi4]\n", "/boot/config.txt", False)
+    assert result.returncode == 1
+    assert "Read-only file system" in result.stderr

core/libs/commonwealth/src/commonwealth/utils/tests/test_general.py

@@ -9,6 +9,7 @@
 from pyfakefs.fake_filesystem import FakeFilesystem
 
 from .. import general
+from ..commands import HostFileError
 from ..general import CpuType, HostOs, get_cpu_type
 
 CPUINFO_TEMPLATE = """processor	: 0
@@ -58,6 +59,17 @@ def test_get_host_os(os_release: str, expected_host_os: HostOs, monkeypatch: pyt
     general.get_host_os.cache_clear()
 
 
+def test_get_host_os_returns_other_when_load_file_fails(monkeypatch: pytest.MonkeyPatch) -> None:
+    general.get_host_os.cache_clear()
+
+    def raise_host_file_error(_file_name: str) -> str:
+        raise HostFileError("Failed to read /etc/os-release")
+
+    monkeypatch.setattr(general, "load_file", raise_host_file_error)
+    assert general.get_host_os() == HostOs.Other
+    general.get_host_os.cache_clear()
+
+
 def test_blueos_version(monkeypatch: pytest.MonkeyPatch) -> None:
     general.blueos_version.cache_clear()
     monkeypatch.setenv("GIT_DESCRIBE_TAGS", "1.5.0-10-gabcdef12")

core/services/ardupilot_manager/flight_controller_detector/linux/navigator.py

@@ -1,7 +1,7 @@
 import platform
 from typing import Any, ClassVar, Dict, List, Optional, Tuple
 
-from commonwealth.utils.commands import load_file
+from commonwealth.utils.commands import HostFileError, load_file
 from commonwealth.utils.general import CpuType, get_cpu_type
 from elftools.elf.elffile import ELFFile
 from flight_controller_detector.linux.linux_boards import LinuxFlightController
@@ -77,7 +77,10 @@ class NavigatorPi4(Navigator):
 
     def get_serials(self) -> List[Serial]:
         release = "Bullseye"
-        os_release = load_file("/etc/os-release")
+        try:
+            os_release = load_file("/etc/os-release")
+        except HostFileError:
+            os_release = ""
         if "bookworm" in os_release.lower():
             release = "Bookworm"
 

core/tools/blueos_startup_update/blueos_startup_update.py

@@ -292,6 +292,9 @@ def revert_update_dwc2() -> bool:
     Removes dwc2 configuration from cmdline.txt
     This was being wrongly applied on Pi3 due to a bad host_cpu check.
     """
+    if cmdline_file is None:
+        logging.warning("cmdline.txt not found. skipping dwc2 revert")
+        return False
 
     # Remove dwc2 module configuration from cmdline
     unpatched_cmdline_content = load_file(cmdline_file).replace("\n", "").split(" ")
@@ -314,6 +317,9 @@ def clean_config_pi3() -> bool:
     Removes any tagged configurations from config.txt on Pi3
     This was being wrongly applied due to a bad host_cpu check.
     """
+    if config_file is None:
+        logging.warning("config.txt not found. skipping pi3 config cleanup")
+        return False
     config_content = load_file(config_file).splitlines()
     unpatched_config_content = config_content.copy()
 
@@ -811,7 +817,13 @@ def main() -> int:
 
     enabled_patches = [(name, patch) for name, patch in patches_to_apply if name not in disabled_patches]
 
-    patches_requiring_restart = [name for name, patch in enabled_patches if patch()]
+    patches_requiring_restart = []
+    for name, patch in enabled_patches:
+        try:
+            if patch():
+                patches_requiring_restart.append(name)
+        except Exception as patch_error:
+            logger.error(f"Patch {name} failed: {patch_error}")
     if patches_requiring_restart:
         logger.warning("The system will restart in 10 seconds because the following applied patches required restart:")
         for patch in patches_requiring_restart: