diff --git a/CHANGELOG.md b/CHANGELOG.md index eca39bdf..c0613e3d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -10,6 +10,8 @@ releases are available on [PyPI](https://pypi.org/project/pytask) and - [#889](https://github.com/pytask-dev/pytask/pull/889) improves typing for tree operations by wrapping optree's pytree utilities with pytask-specific signatures and requiring optree 0.16.0 or newer. +- [#932](https://github.com/pytask-dev/pytask/pull/932) validates unique task signatures + during collection. ## 0.6.0 - 2026-05-01 diff --git a/src/_pytask/collect.py b/src/_pytask/collect.py index 7200c8d4..1d4c697d 100644 --- a/src/_pytask/collect.py +++ b/src/_pytask/collect.py @@ -53,6 +53,7 @@ from _pytask.task_utils import COLLECTED_TASKS from _pytask.task_utils import parse_collected_tasks_with_task_marker from _pytask.task_utils import task as task_decorator +from _pytask.task_utils import validate_unique_task_signatures from _pytask.typing import TaskFunction from _pytask.typing import is_task_function @@ -81,6 +82,7 @@ def pytask_collect(session: Session) -> bool: try: session.hook.pytask_collect_modify_tasks(session=session, tasks=session.tasks) + validate_unique_task_signatures(session.tasks) except Exception: # noqa: BLE001 # pragma: no cover report = CollectionReport.from_exception( outcome=CollectionOutcome.FAIL, exc_info=sys.exc_info() diff --git a/src/_pytask/provisional.py b/src/_pytask/provisional.py index 473b8637..76f724b0 100644 --- a/src/_pytask/provisional.py +++ b/src/_pytask/provisional.py @@ -20,6 +20,7 @@ from _pytask.reports import ExecutionReport from _pytask.task_utils import COLLECTED_TASKS from _pytask.task_utils import parse_collected_tasks_with_task_marker +from _pytask.task_utils import validate_unique_task_signatures from _pytask.tree_util import tree_map from _pytask.tree_util import tree_map_with_path from _pytask.typing import is_task_generator @@ -103,6 +104,7 @@ def pytask_execute_task(session: Session, task: PTask) -> None: session.hook.pytask_collect_modify_tasks( session=session, tasks=session.tasks ) + validate_unique_task_signatures(session.tasks) # Append the last collection report after successful modification if report: session.collection_reports.append(report) @@ -111,6 +113,8 @@ def pytask_execute_task(session: Session, task: PTask) -> None: task=task, exc_info=sys.exc_info() ) session.execution_reports.append(exec_report) + session.should_stop = True + return recreate_dag(session, task) diff --git a/src/_pytask/task_utils.py b/src/_pytask/task_utils.py index e1cd3aac..1d0a7eff 100644 --- a/src/_pytask/task_utils.py +++ b/src/_pytask/task_utils.py @@ -27,6 +27,8 @@ from _pytask.mark import Mark from _pytask.models import CollectionMetadata from _pytask.models import ParsedAfter +from _pytask.nodes import Task +from _pytask.nodes import TaskWithoutPath from _pytask.shared import find_duplicates from _pytask.shared import unwrap_task_function from _pytask.typing import TaskFunction @@ -37,6 +39,8 @@ from pathlib import Path from uuid import UUID + from _pytask.node_protocols import PTask + P = ParamSpec("P") R_co = TypeVar("R_co", covariant=True) @@ -61,6 +65,7 @@ def _is_task_decorator_target(obj: object) -> TypeGuard[Callable[..., Any]]: "parse_collected_tasks_with_task_marker", "parse_keyword_arguments_from_signature_defaults", "task", + "validate_unique_task_signatures", ] @@ -74,6 +79,74 @@ def _is_task_decorator_target(obj: object) -> TypeGuard[Callable[..., Any]]: """ +def validate_unique_task_signatures(tasks: list[PTask]) -> None: + """Raise an error if multiple tasks have the same signature.""" + signature_to_tasks: dict[str, list[PTask]] = defaultdict(list) + for task_ in tasks: + signature_to_tasks[task_.signature].append(task_) + + collisions = { + signature: tasks_ + for signature, tasks_ in signature_to_tasks.items() + if len(tasks_) > 1 + } + if not collisions: + return + + lines = ["Conflicting task identities:"] + groups = sorted( + collisions.values(), key=lambda group: sorted(task_.name for task_ in group) + ) + for conflicting_tasks in groups: + names = ", ".join( + repr(name) for name in sorted({t.name for t in conflicting_tasks}) + ) + lines.extend(("", f"Tasks sharing an identity: {names}")) + lines.extend(f"- {_describe_task(task_)}" for task_ in conflicting_tasks) + if all(type(task_) is TaskWithoutPath for task_ in conflicting_tasks): + guidance = ( + "These tasks have the same name. Choose distinct 'name' values " + "for TaskWithoutPath tasks." + ) + elif all(type(task_) is Task for task_ in conflicting_tasks): + guidance = ( + "Each Task must have a unique combination of 'path' and 'base_name'. " + "Choose distinct 'base_name' values for tasks in the same file; " + "changing the display name does not change task identity." + ) + else: + guidance = ( + "These tasks share a signature. For custom task types, ensure the " + "'signature' implementation returns a distinct, stable value for " + "each task identity." + ) + lines.extend( + ("", guidance + " Alternatively, remove the duplicate registration.") + ) + + raise ValueError("\n".join(lines)) + + +def _describe_task(task_: PTask) -> str: + """Return a task description which distinguishes conflicting definitions.""" + function = unwrap_task_function(task_.function) + while isinstance(function, functools.partial): + function = unwrap_task_function(function.func) + callable_name = getattr(function, "__qualname__", type(function).__qualname__) + module = getattr(function, "__module__", None) + if module: + callable_name = f"{module}.{callable_name}" + + try: + path = get_file(function) + line_number = inspect.getsourcelines(function)[1] + except (OSError, TypeError): + location = "" + else: + location = f"{path.as_posix()}:{line_number}" if path else "" + return f"{task_.name!r}: {callable_name} ({location})" + + @overload def task( name: Callable[P, R_co], diff --git a/tests/test_collect.py b/tests/test_collect.py index 3b6587e7..727b9ce0 100644 --- a/tests/test_collect.py +++ b/tests/test_collect.py @@ -12,17 +12,69 @@ from _pytask.collect import _find_shortest_uniquely_identifiable_name_for_tasks from _pytask.collect import pytask_collect_node from _pytask.node_protocols import PPathNode +from _pytask.pluginmanager import get_plugin_manager +from _pytask.task_utils import validate_unique_task_signatures from pytask import CollectionOutcome from pytask import ExitCode from pytask import NodeInfo +from pytask import PathNode from pytask import PickleNode from pytask import Session from pytask import Task +from pytask import TaskWithoutPath from pytask import build from pytask import cli +from pytask import hookimpl from tests.conftest import noop +def test_duplicate_programmatic_task_signatures_fail_collection(tmp_path): + calls = [] + + def first(): + calls.append("first") + + def second(): + calls.append("second") + + tasks = [ + TaskWithoutPath(name="duplicate", function=first), + TaskWithoutPath(name="duplicate", function=second), + ] + + session = build(tasks=tasks, paths=tmp_path, force=True) + + assert session.exit_code == ExitCode.COLLECTION_FAILED + assert not session.execution_reports + assert not calls + + +def test_duplicate_task_signature_message_contains_all_collision_groups(): + def first(): + pass + + def second(): + pass + + tasks = [ + TaskWithoutPath(name="duplicate-a", function=first), + TaskWithoutPath(name="duplicate-a", function=second), + TaskWithoutPath(name="duplicate-b", function=first), + TaskWithoutPath(name="duplicate-b", function=second), + ] + + with pytest.raises(ValueError, match="Conflicting task identities") as exc_info: + validate_unique_task_signatures(tasks) + + message = str(exc_info.value) + assert "Conflicting task identities" in message + assert "duplicate-a" in message + assert "duplicate-b" in message + assert all(task.signature not in message for task in tasks) + assert "Choose distinct 'name' values" in message + assert f"{Path(__file__).as_posix()}:" in message + + def _make_local_upath_uri(path: Path, protocol: str) -> str: return f"{protocol}:///{path.as_posix().lstrip('/')}" @@ -737,3 +789,103 @@ def test_print_warning_if_non_matching_path_is_passed(runner, tmp_path): assert result.exit_code == ExitCode.OK assert "Collected 0 tasks" in result.output assert "Warning: The path" in result.output + + +@pytest.mark.parametrize("introduce_collision", [True, False]) +def test_validate_task_signatures_after_hook_wrapper( + tmp_path, monkeypatch, introduce_collision +): + class Plugin: + @hookimpl(wrapper=True) + def pytask_collect_modify_tasks(self, tasks): + result = yield + tasks[1].name = "first" if introduce_collision else "second" + return result + + pm = get_plugin_manager() + pm.register(Plugin()) + monkeypatch.setattr("_pytask.build.get_plugin_manager", lambda: pm) + calls = [] + tasks = [ + TaskWithoutPath(name="first", function=lambda: calls.append("first")), + TaskWithoutPath( + name="second" if introduce_collision else "first", + function=lambda: calls.append("second"), + ), + ] + + session = build(tasks=tasks, paths=tmp_path, force=True) + + if introduce_collision: + assert session.exit_code == ExitCode.COLLECTION_FAILED + assert not calls + assert any( + report.exc_info and "Conflicting task identities" in str(report.exc_info[1]) + for report in session.collection_reports + ) + else: + assert session.exit_code == ExitCode.OK + assert sorted(calls) == ["first", "second"] + + +def test_duplicate_task_diagnostics_use_callable_source(tmp_path): + tasks = [ + Task(base_name="duplicate", path=tmp_path / "task_other.py", function=noop), + Task(base_name="duplicate", path=tmp_path / "task_other.py", function=noop), + ] + + with pytest.raises(ValueError, match="Conflicting task identities") as exc_info: + validate_unique_task_signatures(tasks) + + message = str(exc_info.value) + assert "tests.conftest.noop" in message + assert f"{Path(__file__).with_name('conftest.py').as_posix()}:" in message + assert f"({tmp_path.as_posix()}/task_other.py:" not in message + assert "remove the duplicate registration" in message + assert "unique combination of 'path' and 'base_name'" in message + assert "changing the display name does not change task identity" in message + + +def test_duplicate_task_diagnostics_without_source(): + tasks = [TaskWithoutPath(name="duplicate", function=len) for _ in range(2)] + + with pytest.raises(ValueError, match=r"builtins.len \(\)"): + validate_unique_task_signatures(tasks) + + +def test_unique_tasks_can_share_dependency(tmp_path): + dependency = tmp_path / "input.txt" + dependency.write_text("input") + calls = [] + tasks = [ + TaskWithoutPath( + name=name, + function=lambda path: calls.append(path.read_text()), + depends_on={"path": PathNode(path=dependency)}, + ) + for name in ("first", "second") + ] + + session = build(tasks=tasks, paths=tmp_path, force=True) + + assert session.exit_code == ExitCode.OK + assert calls == ["input", "input"] + + +def test_custom_task_identity_conflict_guidance(): + class CustomTask(TaskWithoutPath): + @property + def signature(self): + return "shared-custom-signature" + + tasks = [CustomTask(name=name, function=noop) for name in ("first", "second")] + + with pytest.raises(ValueError, match="Conflicting task identities") as exc_info: + validate_unique_task_signatures(tasks) + + message = str(exc_info.value) + assert "Tasks sharing an identity: 'first', 'second'" in message + assert "shared-custom-signature" not in message + assert "'signature' implementation" in message + assert "distinct, stable value" in message + assert "These tasks have the same name" not in message diff --git a/tests/test_provisional.py b/tests/test_provisional.py index 203e784f..2fcd78a6 100644 --- a/tests/test_provisional.py +++ b/tests/test_provisional.py @@ -2,10 +2,14 @@ import textwrap +import pytest + +from _pytask.pluginmanager import get_plugin_manager from pytask import ExitCode from pytask import TaskOutcome from pytask import build from pytask import cli +from pytask import hookimpl def test_task_that_produces_provisional_path_node(tmp_path): @@ -280,3 +284,51 @@ def task_example( assert result.exit_code == ExitCode.OK assert tmp_path.joinpath("subfolder", "a.txt").exists() assert tmp_path.joinpath("subfolder", "b.txt").exists() + + +@pytest.mark.parametrize("collision_in_wrapper", [False, True]) +def test_generated_task_identity_conflict_stops_before_dag_rebuild( + tmp_path, monkeypatch, collision_in_wrapper +): + source = """ + from pathlib import Path + from pytask import task, mark + + def task_existing(): + Path(__file__).with_name("existing-ran").touch() + + @mark.try_first + @task(is_generator=True) + def task_generator(): + @task(name=GENERATED_NAME) + def generated(): + Path(__file__).with_name("generated-ran").touch() + """ + generated_name = "task_generated" if collision_in_wrapper else "task_existing" + source = source.replace("GENERATED_NAME", repr(generated_name)) + tmp_path.joinpath("task_module.py").write_text(textwrap.dedent(source)) + + class Plugin: + @hookimpl(wrapper=True) + def pytask_collect_modify_tasks(self, tasks): + result = yield + for task in tasks: + if task.base_name == "task_generated": + task.base_name = "task_existing" + return result + + pm = get_plugin_manager() + pm.register(Plugin()) + monkeypatch.setattr("_pytask.build.get_plugin_manager", lambda: pm) + session = build(paths=tmp_path, force=True) + + assert session.exit_code == ExitCode.FAILED + assert session.should_stop + assert not tmp_path.joinpath("existing-ran").exists() + assert not tmp_path.joinpath("generated-ran").exists() + assert any( + report.exc_info and "Conflicting task identities" in str(report.exc_info[1]) + for report in session.execution_reports + ) + assert len(session.tasks) == 3 + assert len(session.dag.nodes) == 2