perf: test suite improvements

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.

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):

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,62 +694,38 @@ 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.
 
     This test ensures that when multiple tasks with the same function name are passed
     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())