Skip to content

feat: add failure_message option for tasks (#2080) #2081

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wants to merge 2 commits into
base: master
Choose a base branch
from
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 4 additions & 1 deletion copier/_main.py
Original file line number Diff line number Diff line change
Expand Up @@ -389,7 +389,10 @@ def _execute_tasks(self, tasks: Sequence[Task]) -> None:
with local.cwd(working_directory), local.env(**extra_env):
process = subprocess.run(task_cmd, shell=use_shell, env=local.env)
if process.returncode:
raise TaskError.from_process(process)
message: str | None = None
if task.failure_message:
message = self._render_string(task.failure_message, extra_context)
raise TaskError.from_process(process, i, message=message)

def _render_context(self) -> AnyByStrMutableMapping:
"""Produce render context for Jinja."""
Expand Down
6 changes: 6 additions & 0 deletions copier/_template.py
Original file line number Diff line number Diff line change
Expand Up @@ -169,12 +169,17 @@ class Task:
working_directory:
The directory from inside where to execute the task.
If `None`, the project directory will be used.

failure_message:
Provides a message to print if the task fails.
If `None`, the subprocess exception message will be used.
"""

cmd: str | Sequence[str]
extra_vars: dict[str, Any] = field(default_factory=dict)
condition: str | bool = True
working_directory: Path = Path()
failure_message: str | None = None


@dataclass
Expand Down Expand Up @@ -526,6 +531,7 @@ def tasks(self) -> Sequence[Task]:
extra_vars=extra_vars,
condition=task.get("when", "true"),
working_directory=Path(task.get("working_directory", ".")),
failure_message=task.get("failure_message"),
)
)
else:
Expand Down
20 changes: 17 additions & 3 deletions copier/errors.py
Original file line number Diff line number Diff line change
Expand Up @@ -142,27 +142,41 @@ class TaskError(subprocess.CalledProcessError, UserMessageError):

def __init__(
self,
index: int,
command: str | Sequence[str],
returncode: int,
stdout: str | bytes | None,
stderr: str | bytes | None,
message: str | None = None,
):
subprocess.CalledProcessError.__init__(
self, returncode=returncode, cmd=command, output=stdout, stderr=stderr
)
message = f"Task {command!r} returned non-zero exit status {returncode}."
UserMessageError.__init__(self, message)
self.index = index
if not message:
message = subprocess.CalledProcessError.__str__(self)
message = message.rstrip(".")
message = f"Task {index + 1} failed: {message}."
UserMessageError.__init__(self, message=message)

def __str__(self) -> str:
return self.message

@classmethod
def from_process(
cls, process: CompletedProcess[str] | CompletedProcess[bytes]
cls,
process: CompletedProcess[str] | CompletedProcess[bytes],
index: int,
message: str | None = None,
) -> Self:
"""Create a TaskError from a CompletedProcess."""
return cls(
index,
command=process.args,
returncode=process.returncode,
stdout=process.stdout,
stderr=process.stderr,
message=message,
)


Expand Down
4 changes: 4 additions & 0 deletions docs/configuring.md
Original file line number Diff line number Diff line change
Expand Up @@ -1572,6 +1572,8 @@ If a `dict` is given it can contain the following items:
- **when** (optional): Specifies a condition that needs to hold for the task to run.
- **working_directory** (optional): Specifies the directory in which the command will
be run. Defaults to the destination directory.
- **failure_message** (optional): Provides a message to print if the task fails. If
not provided, the subprocess exception message will be shown.

If a `str` or `List[str]` is given as a task it will be treated as `command` with all
other items not present.
Expand Down Expand Up @@ -1599,6 +1601,8 @@ Refer to the example provided below for more information.
when: "{{ _copier_conf.os in ['linux', 'macos'] }}"
- command: Remove-Item {{ name_of_the_project }}\\README.md
when: "{{ _copier_conf.os == 'windows' }}"
- command: finalize-project
failure_message: Couldn't finalize {{ name_of_the_project }}
```

Note: the example assumes you use [Invoke](https://www.pyinvoke.org/) as
Expand Down
36 changes: 36 additions & 0 deletions tests/test_tasks.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@
from pathlib import Path
from typing import Literal

from copier.errors import TaskError
import pytest
import yaml

Expand Down Expand Up @@ -180,3 +181,38 @@ def test_copier_phase_variable(tmp_path_factory: pytest.TempPathFactory) -> None
)
copier.run_copy(str(src), dst, unsafe=True)
assert (dst / "tasks").exists()


@pytest.mark.parametrize(
"task,failure_message,match,data",
[
("false", "Oh, dear! The task failed", r"^Task \d+ failed: Oh, dear! The task failed\.$", None),
("ls non-existing-directory", None, r"^Task \d+ failed: Command 'ls non-existing-directory' returned non-zero exit status 2\.$", None),
("false", "{{ name }} blew up", r"^Task \d+ failed: Wile E. Coyote blew up\.$", {"name": "Wile E. Coyote"}),
],
)
def test_task_failure_message(
tmp_path_factory: pytest.TempPathFactory,
task: str,
failure_message: str | None,
match: str,
data: dict[str, str] | None,
) -> None:
src, dst = map(tmp_path_factory.mktemp, ("src", "dst"))
yaml_dict = {
"_tasks": [
{
"command": task,
}
]
}
if failure_message:
yaml_dict["_tasks"][0]["failure_message"] = failure_message

build_file_tree(
{
(src / "copier.yml"): yaml.safe_dump(yaml_dict)
}
)
with pytest.raises(TaskError, match=match):
copier.run_copy(str(src), dst, unsafe=True, data=data)