✨ Add TaskError exception for subprocess management

copier/errors.py

@@ -1,7 +1,10 @@
 """Custom exceptions used by Copier."""
 
+from __future__ import annotations
+
 from pathlib import Path
-from typing import TYPE_CHECKING, Sequence
+from subprocess import CompletedProcess
+from typing import TYPE_CHECKING, Self, Sequence
 
 from .tools import printf_exception
 from .types import PathSeq
@@ -120,6 +123,36 @@ class MultipleYieldTagsError(CopierError):
     """Multiple yield tags are used in one path name, but it is not allowed."""
 
 
+class TaskError(CopierError):
+    """Exception raised when a task fails."""
+
+    def __init__(
+        self,
+        task_command: str | Sequence[str],
+        returncode: int,
+        stdout: str | bytes | None,
+        stderr: str | bytes | None,
+    ):
+        self.task_command = task_command
+        self.returncode = returncode
+        self.stdout = stdout
+        self.stderr = stderr
+        message = f"Task {task_command!r} returned non-zero exit status {returncode}."
+        super().__init__(message)
+
+    @classmethod
+    def from_process(
+        cls, process: CompletedProcess[str] | CompletedProcess[bytes]
+    ) -> Self:
+        """Create a TaskError from a CompletedProcess."""
+        return cls(
+            task_command=process.args,
+            returncode=process.returncode,
+            stdout=process.stdout,
+            stderr=process.stderr,
+        )
+
+
 # Warnings
 class CopierWarning(Warning):
     """Base class for all other Copier warnings."""

copier/main.py

@@ -41,6 +41,7 @@
 from .errors import (
     CopierAnswersInterrupt,
     ExtensionNotFoundError,
+    TaskError,
     UnsafeTemplateError,
     UserMessageError,
     YieldTagInFileError,
@@ -366,7 +367,9 @@ def _execute_tasks(self, tasks: Sequence[Task]) -> None:
 
             extra_env = {k.upper(): str(v) for k, v in task.extra_vars.items()}
             with local.cwd(working_directory), local.env(**extra_env):
-                subprocess.run(task_cmd, shell=use_shell, check=True, env=local.env)
+                process = subprocess.run(task_cmd, shell=use_shell, env=local.env)
+                if process.returncode:
+                    raise TaskError.from_process(process)
 
     def _system_render_context(self) -> AnyByStrMutableMapping:
         """System reserved render context.

tests/test_cleanup.py

@@ -1,5 +1,4 @@
 from pathlib import Path
-from subprocess import CalledProcessError
 
 import pytest
 from plumbum import local
@@ -10,15 +9,15 @@
 def test_cleanup(tmp_path: Path) -> None:
     """Copier creates dst_path, fails to copy and removes it."""
     dst = tmp_path / "new_folder"
-    with pytest.raises(CalledProcessError):
+    with pytest.raises(copier.errors.TaskError):
         copier.run_copy("./tests/demo_cleanup", dst, quiet=True, unsafe=True)
     assert not dst.exists()
 
 
 def test_do_not_cleanup(tmp_path: Path) -> None:
     """Copier creates dst_path, fails to copy and keeps it."""
     dst = tmp_path / "new_folder"
-    with pytest.raises(CalledProcessError):
+    with pytest.raises(copier.errors.TaskError):
         copier.run_copy(
             "./tests/demo_cleanup", dst, quiet=True, unsafe=True, cleanup_on_error=False
         )
@@ -29,7 +28,7 @@ def test_no_cleanup_when_folder_existed(tmp_path: Path) -> None:
     """Copier will not delete a folder if it didn't create it."""
     preexisting_file = tmp_path / "something"
     preexisting_file.touch()
-    with pytest.raises(CalledProcessError):
+    with pytest.raises(copier.errors.TaskError):
         copier.run_copy(
             "./tests/demo_cleanup",
             tmp_path,