Skip to content
Merged
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
2 changes: 1 addition & 1 deletion src/_pytask/execute.py
Original file line number Diff line number Diff line change
Expand Up @@ -122,7 +122,7 @@ def pytask_execute_task_protocol(session: Session, task: PTask) -> ExecutionRepo
short_exc_info = remove_traceback_from_exc_info(sys.exc_info())
report = ExecutionReport.from_task_and_exception(task, short_exc_info)
session.should_stop = True
except Exception: # noqa: BLE001
except BaseException: # noqa: BLE001
report = ExecutionReport.from_task_and_exception(task, sys.exc_info())
else:
report = ExecutionReport.from_task(task)
Expand Down
40 changes: 40 additions & 0 deletions tests/test_execute.py
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,7 @@
import pytest

import pytask
from _pytask.mark import MARK_GEN
from pytask import CaptureMethod
from pytask import ExitCode
from pytask import NodeNotFoundError
Expand Down Expand Up @@ -123,6 +124,45 @@ def task_3(paths = [Path("deleted.txt"), Path("out_2.txt")]):
assert isinstance(exc_info[1], NodeNotFoundError)


@pytest.mark.parametrize("exit_code", [0, 7])
def test_system_exit_is_a_normal_task_failure(tmp_path, exit_code):
source = f"""
from pathlib import Path

def task_exits(produces=Path("upstream.txt")):
raise SystemExit({exit_code})

def task_descendant(
path=Path("upstream.txt"), produces=Path("descendant.txt")
):
produces.write_text(path.read_text())

def task_unrelated(produces=Path("unrelated.txt")):
produces.write_text("unrelated")
"""
tmp_path.joinpath("task_module.py").write_text(textwrap.dedent(source))

session = build(paths=tmp_path)

assert session.exit_code == ExitCode.FAILED
assert tmp_path.joinpath("unrelated.txt").read_text() == "unrelated"
assert not tmp_path.joinpath("descendant.txt").exists()
assert {report.outcome for report in session.execution_reports} == {
TaskOutcome.FAIL,
TaskOutcome.SKIP_PREVIOUS_FAILED,
TaskOutcome.SUCCESS,
}
failed_report = next(
report
for report in session.execution_reports
if report.outcome == TaskOutcome.FAIL
)
assert failed_report.exc_info is not None
assert isinstance(failed_report.exc_info[1], SystemExit)
assert failed_report.exc_info[1].code == exit_code
assert MARK_GEN.config is None


def test_depends_on_and_produces_can_be_used_in_task(tmp_path):
source = """
from pathlib import Path
Expand Down