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: 2 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,8 @@ releases are available on [PyPI](https://pypi.org/project/pytask) and

## Unreleased

- [#997](https://github.com/pytask-dev/pytask/pull/997) documents how task generators
behave in dry-run and explain modes and adds regression tests for both modes.
- [#995](https://github.com/pytask-dev/pytask/pull/995) documents that return values
from task generators are ignored and adds a regression test for this behavior.
- [#996](https://github.com/pytask-dev/pytask/pull/996) raises collection errors when
Expand Down
35 changes: 35 additions & 0 deletions tests/test_provisional.py
Original file line number Diff line number Diff line change
Expand Up @@ -275,6 +275,41 @@ def task_child(produces=Path("child.txt")):
assert "return" not in generator.produces


@pytest.mark.parametrize("mode", ["dry_run", "explain"])
def test_task_generator_in_simulation_mode(tmp_path, mode):
source = """
from pathlib import Path
from pytask import task

@task(is_generator=True)
def task_generator():
counter = Path(__file__).parent / "counter.txt"
count = int(counter.read_text()) if counter.exists() else 0
counter.write_text(str(count + 1))

@task
def task_generated(produces=Path(__file__).parent / "generated.txt"):
produces.write_text("generated")
"""
tmp_path.joinpath("task_module.py").write_text(textwrap.dedent(source))

session = build(
paths=tmp_path, dry_run=mode == "dry_run", explain=mode == "explain"
)

outcomes = {
report.task.name.rsplit("::", maxsplit=1)[-1]: report.outcome
for report in session.execution_reports
}
assert session.exit_code == ExitCode.OK
assert outcomes == {
"task_generator": TaskOutcome.WOULD_BE_EXECUTED,
"task_generated": TaskOutcome.WOULD_BE_EXECUTED,
}
assert tmp_path.joinpath("counter.txt").read_text() == "1"
assert not tmp_path.joinpath("generated.txt").exists()


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