diff --git a/tests/test_capture.py b/tests/test_capture.py index c899401f..fcbc319d 100644 --- a/tests/test_capture.py +++ b/tests/test_capture.py @@ -21,6 +21,7 @@ from _pytask.capture import _get_multicapture from pytask import CaptureMethod from pytask import ExitCode +from pytask import build from pytask import cli from tests.conftest import enter_directory from tests.conftest import run_in_subprocess @@ -139,27 +140,14 @@ def task_show_capture(): raise NotImplementedError -@pytest.mark.xfail( - sys.platform == "win32", - reason="from pytask ... cannot be found", - raises=AssertionError, -) -def test_wrong_capture_method(tmp_path): - source = """ - from pytask import build - import sys +def test_wrong_capture_method(tmp_path, capsys): + session = build(paths=tmp_path, show_capture="a") # type: ignore[arg-type] - if __name__ == "__main__": - session = build(tasks=[], show_capture="a") - sys.exit(session.exit_code) - """ - tmp_path.joinpath("workflow.py").write_text(textwrap.dedent(source)) - - result = run_in_subprocess((sys.executable, "workflow.py"), cwd=tmp_path) - assert result.exit_code == ExitCode.CONFIGURATION_FAILED - assert "Value 'a' is not a valid" in result.stdout - assert "Traceback" not in result.stdout - assert not result.stderr + captured = capsys.readouterr() + assert session.exit_code == ExitCode.CONFIGURATION_FAILED + assert "Value 'a' is not a valid" in captured.out + assert "Traceback" not in captured.out + assert not captured.err # Following tests are copied from pytest. diff --git a/tests/test_config.py b/tests/test_config.py index 418780ff..d09e03ba 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -1,15 +1,14 @@ from __future__ import annotations -import os -import sys import textwrap +from pathlib import Path import pytest from pytask import ExitCode from pytask import build from pytask import cli -from tests.conftest import run_in_subprocess +from tests.conftest import enter_directory def test_debug_pytask(capsys, tmp_path): @@ -74,7 +73,7 @@ def test_not_existing_path_in_config(runner, tmp_path): assert result.exit_code == ExitCode.CONFIGURATION_FAILED -def test_paths_are_relative_to_configuration_file_cli(tmp_path): +def test_paths_are_relative_to_configuration_file_cli(tmp_path, runner): tmp_path.joinpath("src").mkdir() tmp_path.joinpath("tasks").mkdir() config = """ @@ -86,15 +85,12 @@ def test_paths_are_relative_to_configuration_file_cli(tmp_path): source = "def task_example(): ..." tmp_path.joinpath("tasks", "task_example.py").write_text(source) - result = run_in_subprocess(("pytask", "src"), cwd=tmp_path) + with enter_directory(tmp_path): + result = runner.invoke(cli, ["src"]) assert result.exit_code == ExitCode.OK - assert "1 Succeeded" in result.stdout + assert "1 Succeeded" in result.output -@pytest.mark.skipif( - sys.platform == "win32" and os.environ.get("CI") == "true", - reason="Windows does not pick up the right Python interpreter.", -) def test_paths_are_relative_to_configuration_file(tmp_path): tmp_path.joinpath("src").mkdir() tmp_path.joinpath("tasks").mkdir() @@ -107,16 +103,11 @@ def test_paths_are_relative_to_configuration_file(tmp_path): source = "def task_example(): ..." tmp_path.joinpath("tasks", "task_example.py").write_text(source) - source = """ - from pytask import build - from pathlib import Path + with enter_directory(tmp_path): + session = build(paths=[Path("src")]) - session = build(paths=[Path("src")]) - """ - tmp_path.joinpath("script.py").write_text(textwrap.dedent(source)) - result = run_in_subprocess((sys.executable, "script.py"), cwd=tmp_path) - assert result.exit_code == ExitCode.OK - assert "1 Succeeded" in result.stdout + assert session.exit_code == ExitCode.OK + assert len(session.tasks) == 1 def test_create_gitignore_file_in_pytask_directory(tmp_path): diff --git a/tests/test_execute.py b/tests/test_execute.py index c8795684..5067d6ac 100644 --- a/tests/test_execute.py +++ b/tests/test_execute.py @@ -1,17 +1,19 @@ from __future__ import annotations import json -import os import pickle import re import subprocess import sys import textwrap +from pathlib import Path +from typing import Annotated import pytest import pytask from _pytask.mark import MARK_GEN +from _pytask.path import HashPathCache from pytask import CaptureMethod from pytask import ExitCode from pytask import NodeNotFoundError @@ -620,67 +622,51 @@ def task_create_file( assert tmp_path.joinpath("file.txt").read_text() == "This is the text." -@pytest.mark.xfail(sys.platform == "win32", reason="Decoding issues in Gitlab Actions.") -def test_execute_tasks_via_functional_api(tmp_path): - source = """ - import sys - from pathlib import Path - from typing import Annotated - from pytask import PathNode, PythonNode, build +def test_execute_tasks_via_functional_api(monkeypatch, tmp_path): + node_text = pytask.PythonNode() + output_path = tmp_path / "file.txt" - node_text = PythonNode() - - def create_text() -> Annotated[int, node_text]: + def create_text(): return "This is the text." - def create_file( - content: Annotated[str, node_text] - ) -> Annotated[str, Path("file.txt")]: + def create_file(content): return content - if __name__ == "__main__": - session = build(tasks=[create_file, create_text]) - assert len(session.tasks) == 2 - assert len(session.dag.nodes) == 5 - sys.exit(session.exit_code) - """ - tmp_path.joinpath("task_module.py").write_text(textwrap.dedent(source)) - result = subprocess.run( - (sys.executable, tmp_path.joinpath("task_module.py").as_posix()), - check=False, - cwd=tmp_path, + monkeypatch.setitem(create_text.__globals__, "_functional_api_node", node_text) + monkeypatch.setitem( + create_text.__globals__, + "_functional_api_output", + Annotated[str, output_path], ) - assert result.returncode == ExitCode.OK - assert tmp_path.joinpath("file.txt").read_text() == "This is the text." + create_text.__annotations__ = {"return": "Annotated[int, _functional_api_node]"} + create_file.__annotations__ = { + "content": "Annotated[str, _functional_api_node]", + "return": "_functional_api_output", + } + + session = build(tasks=[create_file, create_text], paths=tmp_path) + + assert session.exit_code == ExitCode.OK + assert len(session.tasks) == 2 + assert len(session.dag.nodes) == 5 + assert output_path.read_text() == "This is the text." -@pytest.mark.skipif( - sys.platform == "win32" and os.environ.get("CI") == "true", - reason="Windows does not pick up the right Python interpreter.", -) def test_execute_tasks_multiple_times_via_api(tmp_path): """See #625.""" - source = """ - import pathlib - from typing import Annotated - from pytask import build, task - import sys - @task - def task1() -> None: pass - def task2() -> None: pass + @pytask.task + def task1() -> None: + pass - if __name__ == "__main__": - session1 = build(tasks=[task1, task2]) - session2 = build(tasks=[task1, task2]) - sys.exit(session2.exit_code) - """ - tmp_path.joinpath("task_module.py").write_text(textwrap.dedent(source)) - result = run_in_subprocess( - (sys.executable, tmp_path.joinpath("task_module.py").as_posix()), - cwd=tmp_path, - ) - assert result.exit_code == ExitCode.OK + def task2() -> None: + pass + + session1 = build(tasks=[task1, task2], paths=tmp_path) + session2 = build(tasks=[task1, task2], paths=tmp_path) + + assert session1.exit_code == ExitCode.OK + assert session2.exit_code == ExitCode.OK @pytest.mark.xfail( @@ -708,10 +694,6 @@ def test_pass_non_task_to_functional_api_that_are_ignored(): assert len(session.tasks) == 0 -@pytest.mark.skipif( - sys.platform == "win32" and os.environ.get("CI") == "true", - reason="Windows does not pick up the right Python interpreter.", -) def test_repeated_tasks_via_functional_interface(tmp_path): """Test that repeated tasks with the same function name work correctly. @@ -719,51 +701,31 @@ def test_repeated_tasks_via_functional_interface(tmp_path): to pytask.build(), they all get unique IDs and execute correctly, similar to how file-based collection handles repeated tasks. """ - source = """ - from pathlib import Path - from typing import Annotated - from pytask import Product, task, build, ExitCode - import sys - - # Create repeated tasks with the same function name tasks = [] for i in range(3): + def create_data( value: int = i * 10, - produces: Annotated[Path, Product] = Path(f"output_{i}.txt") + produces=tmp_path / f"output_{i}.txt", ) -> None: - '''Generate data based on a value.''' + """Generate data based on a value.""" produces.write_text(str(value)) + create_data.__annotations__["produces"] = Annotated[Path, pytask.Product] tasks.append(create_data) - if __name__ == "__main__": - session = build(tasks=tasks) + session = build(tasks=tasks, paths=tmp_path) - # Verify all tasks were collected and executed - assert session.exit_code == ExitCode.OK, f"Exit code: {session.exit_code}" - assert len(session.tasks) == 3, f"Expected 3 tasks, got {len(session.tasks)}" - assert len(session.execution_reports) == 3 + assert session.exit_code == ExitCode.OK, f"Exit code: {session.exit_code}" + assert len(session.tasks) == 3, f"Expected 3 tasks, got {len(session.tasks)}" + assert len(session.execution_reports) == 3 + assert tmp_path.joinpath("output_0.txt").read_text() == "0" + assert tmp_path.joinpath("output_1.txt").read_text() == "10" + assert tmp_path.joinpath("output_2.txt").read_text() == "20" - # Verify each task executed and produced the correct output - assert Path("output_0.txt").read_text() == "0" - assert Path("output_1.txt").read_text() == "10" - assert Path("output_2.txt").read_text() == "20" - - # Verify tasks have unique names with repeated task IDs - task_names = [task.name for task in session.tasks] - assert len(task_names) == len(set(task_names)), "Task names should be unique" - assert all("create_data[" in name for name in task_names), \\ - f"Task names should contain repeated task IDs: {task_names}" - - sys.exit(session.exit_code) - """ - tmp_path.joinpath("task_module.py").write_text(textwrap.dedent(source)) - result = run_in_subprocess( - (sys.executable, tmp_path.joinpath("task_module.py").as_posix()), - cwd=tmp_path, - ) - assert result.exit_code == ExitCode.OK + task_names = [task.name for task in session.tasks] + assert len(task_names) == len(set(task_names)), "Task names should be unique" + assert all("create_data[" in name for name in task_names) def test_multiple_product_annotations(runner, tmp_path): @@ -832,8 +794,8 @@ def task_example( assert "_pytask/execute.py" not in result.output -def test_hashing_works(tmp_path): - """Use subprocess or otherwise the cache is filled from other tests.""" +def test_hashing_works(tmp_path, monkeypatch, runner): + monkeypatch.setattr(HashPathCache, "_cache", {}) source = """ from pathlib import Path from typing import Annotated @@ -843,13 +805,13 @@ def task_example() -> Annotated[str, Path("file.txt")]: """ tmp_path.joinpath("task_example.py").write_text(textwrap.dedent(source)) - result = run_in_subprocess(("pytask",), cwd=tmp_path) + result = runner.invoke(cli, [tmp_path.as_posix()]) assert result.exit_code == ExitCode.OK hashes = json.loads(tmp_path.joinpath(".pytask", "file_hashes.json").read_text()) assert len(hashes) == 2 - result = run_in_subprocess(("pytask",), cwd=tmp_path) + result = runner.invoke(cli, [tmp_path.as_posix()]) assert result.exit_code == ExitCode.OK hashes_ = json.loads(tmp_path.joinpath(".pytask", "file_hashes.json").read_text()) diff --git a/tests/test_hook_module.py b/tests/test_hook_module.py index 4b5a75cc..79f5bd30 100644 --- a/tests/test_hook_module.py +++ b/tests/test_hook_module.py @@ -1,12 +1,13 @@ from __future__ import annotations -import subprocess import sys import textwrap import pytest from pytask import ExitCode +from pytask import cli +from tests.conftest import enter_directory from tests.conftest import run_in_subprocess @@ -62,32 +63,28 @@ def pytask_extend_command_line_interface(cli): assert "--new-option" in result.stdout -def test_error_when_hook_module_path_does_not_exist(tmp_path): - result = subprocess.run( # noqa: PLW1510 - ("pytask", "build", "--hook-module", "hooks.py", "--help"), - cwd=tmp_path, - capture_output=True, - ) - assert result.returncode == ExitCode.CONFIGURATION_FAILED - assert b"Error: Invalid value for '--hook-module'" in result.stderr +def test_error_when_hook_module_path_does_not_exist(tmp_path, runner): + with enter_directory(tmp_path): + result = runner.invoke(cli, ["build", "--hook-module", "hooks.py", "--help"]) + assert result.exit_code == ExitCode.CONFIGURATION_FAILED + assert "Error: Invalid value for '--hook-module'" in result.output -def test_error_when_hook_module_module_does_not_exist(tmp_path): - result = subprocess.run( # noqa: PLW1510 - ("pytask", "build", "--hook-module", "hooks", "--help"), - cwd=tmp_path, - capture_output=True, - ) - assert result.returncode == ExitCode.CONFIGURATION_FAILED - assert b"Error: Invalid value for '--hook-module':" in result.stderr +def test_error_when_hook_module_module_does_not_exist(tmp_path, runner): + with enter_directory(tmp_path): + result = runner.invoke(cli, ["build", "--hook-module", "hooks", "--help"]) + + assert result.exit_code == ExitCode.CONFIGURATION_FAILED + assert "Error: Invalid value for '--hook-module':" in result.output -def test_error_when_hook_module_is_no_iterable(tmp_path): + +def test_error_when_hook_module_is_no_iterable(tmp_path, runner): tmp_path.joinpath("pyproject.toml").write_text( "[tool.pytask.ini_options]\nhook_module = 'hooks'" ) - result = subprocess.run( # noqa: PLW1510 - ("pytask", "build", "--help"), cwd=tmp_path, capture_output=True - ) - assert result.returncode == ExitCode.CONFIGURATION_FAILED - assert b"Error: Invalid value for '--hook-module':" in result.stderr + with enter_directory(tmp_path): + result = runner.invoke(cli, ["build", "--help"]) + + assert result.exit_code == ExitCode.CONFIGURATION_FAILED + assert "Error: Invalid value for '--hook-module':" in result.output diff --git a/tests/test_profile.py b/tests/test_profile.py index 7454d997..634c5263 100644 --- a/tests/test_profile.py +++ b/tests/test_profile.py @@ -1,5 +1,7 @@ from __future__ import annotations +import csv +import json import textwrap import pytest @@ -13,23 +15,20 @@ def test_duration_is_stored_in_task(tmp_path): source = """ - import time - def task_example(): time.sleep(2) + def task_example(): pass """ tmp_path.joinpath("task_example.py").write_text(textwrap.dedent(source)) - session = build(paths=tmp_path) assert session.exit_code == ExitCode.OK assert len(session.tasks) == 1 task = session.tasks[0] duration = task.attributes["duration"] - assert duration[1] - duration[0] > 2 + assert duration[0] <= duration[1] runtime_state = RuntimeState.from_root(tmp_path) - duration = runtime_state.get_duration(task) - assert duration is not None - assert duration > 2 + stored_duration = runtime_state.get_duration(task) + assert stored_duration == pytest.approx(duration[1] - duration[0]) def test_profile_if_no_tasks_are_collected(tmp_path, runner): @@ -40,8 +39,7 @@ def test_profile_if_no_tasks_are_collected(tmp_path, runner): def test_profile_if_there_is_no_information_on_collected_tasks(tmp_path, runner): source = """ - import time - def task_example(): time.sleep(2) + def task_example(): pass """ tmp_path.joinpath("task_example.py").write_text(textwrap.dedent(source)) @@ -54,11 +52,9 @@ def task_example(): time.sleep(2) def test_profile_if_there_is_information_on_collected_tasks(tmp_path, runner): source = """ - import time from pathlib import Path def task_example(produces=Path("out.txt")): - time.sleep(2) produces.write_text("There are nine billion bicycles in Beijing.") """ tmp_path.joinpath("task_example.py").write_text(textwrap.dedent(source)) @@ -72,8 +68,7 @@ def task_example(produces=Path("out.txt")): @pytest.mark.parametrize("export", ["csv", "json"]) def test_export_of_profile(tmp_path, runner, export): source = """ - import time - def task_example(): time.sleep(2) + def task_example(): pass """ tmp_path.joinpath("task_example.py").write_text(textwrap.dedent(source)) @@ -81,7 +76,19 @@ def task_example(): time.sleep(2) result = runner.invoke(cli, ["profile", tmp_path.as_posix(), "--export", export]) assert result.exit_code == ExitCode.OK - assert tmp_path.joinpath(f"profile.{export}").exists() + export_path = tmp_path.joinpath(f"profile.{export}") + if export == "csv": + with export_path.open(newline="") as file: + rows = list(csv.DictReader(file)) + assert len(rows) == 1 + assert rows[0]["Task"].endswith("task_example") + assert float(rows[0]["Duration (in s)"]) >= 0 + else: + profile = json.loads(export_path.read_text()) + assert len(profile) == 1 + task_name, task_profile = next(iter(profile.items())) + assert task_name.endswith("task_example") + assert task_profile["Duration (in s)"] >= 0 @pytest.mark.parametrize( diff --git a/tests/test_task.py b/tests/test_task.py index b60a0195..3da34695 100644 --- a/tests/test_task.py +++ b/tests/test_task.py @@ -1,14 +1,14 @@ from __future__ import annotations -import subprocess -import sys import textwrap +from typing import Annotated import pytest from pytask import ExitCode from pytask import build from pytask import cli +from pytask import task @pytest.mark.parametrize("func_name", ["task_example", "func"]) @@ -642,40 +642,29 @@ def task_second(): @pytest.mark.parametrize("decorator", ["", "@task"]) -@pytest.mark.xfail( - reason="Wrong python interpreter picked up in CI?", - condition=sys.platform == "win32", - strict=True, -) def test_task_will_be_executed_after_another_one_with_function_session( - tmp_path, decorator + monkeypatch, tmp_path, decorator ): - source = f""" - from pytask import task, ExitCode, build - from pathlib import Path - from typing import Annotated + output_path = tmp_path / "out.txt" - {decorator} - def task_first() -> Annotated[str, Path("out.txt")]: + def task_first(): return "Hello, World!" - @task(after=task_first) def task_second(): - assert Path(__file__).parent.joinpath("out.txt").exists() + assert output_path.exists() - session = build(tasks=[task_first, task_second]) - assert session.exit_code == ExitCode.OK - """ - tmp_path.joinpath("task_example.py").write_text(textwrap.dedent(source)) - - result = subprocess.run( - (sys.executable, "task_example.py"), - cwd=tmp_path, - capture_output=True, - check=False, + monkeypatch.setitem( + task_first.__globals__, "_task_output_annotation", Annotated[str, output_path] ) - assert "2 Succeeded" in result.stdout.decode() - assert result.returncode == ExitCode.OK + task_first.__annotations__ = {"return": "_task_output_annotation"} + first_task = task(task_first) if decorator else task_first + second_task = task(after=first_task)(task_second) + + session = build(tasks=[first_task, second_task], paths=tmp_path) + + assert session.exit_code == ExitCode.OK + assert len(session.execution_reports) == 2 + assert output_path.read_text() == "Hello, World!" def test_raise_error_for_wrong_after_expression(runner, tmp_path): diff --git a/tests/test_tree_util.py b/tests/test_tree_util.py index c2978923..5ce77afc 100644 --- a/tests/test_tree_util.py +++ b/tests/test_tree_util.py @@ -47,14 +47,12 @@ def task_example({arg_name}=complex): def test_profile_with_pytree(tmp_path, runner): source = """ - import time from pytask.tree_util import tree_leaves from pathlib import Path def task_example( produces=[{"out_1": Path("out_1.txt")}, {"out_2": Path("out_2.txt")}] ): - time.sleep(2) for p in tree_leaves(produces): p.write_text("There are nine billion bicycles in Beijing.") """