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

core/libs/commonwealth/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/commonwealth/utils/tests/test_commands.py

@@ -0,0 +1,62 @@
+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_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