Fix lockfile validation scope

Author: tobiasraabe

src/_pytask/lockfile.py

@@ -59,6 +59,14 @@ class _JournalEntry(msgspec.Struct):
     produces: dict[str, str] = msgspec.field(default_factory=dict)
 
 
+def _should_initialize_lockfile_state(command: str | None) -> bool:
+    return command in (None, "build")
+
+
+def _should_validate_lockfile_ids(command: str | None) -> bool:
+    return command in (None, "build", "collect")
+
+
 def _encode_node_path(path: tuple[str | int, ...]) -> str:
     return msgspec.json.encode(path).decode()
 
@@ -276,16 +284,6 @@ def _raise_error_if_lockfile_ids_are_ambiguous(tasks: list[PTask], root: Path) -
         )
 
         for kind, node in chain(dependencies, products):
-            try:
-                node_state = node.state()
-            except Exception:  # noqa: BLE001
-                # Preserve existing behavior where state() errors are raised during
-                # task execution, not during collection-time lockfile validation.
-                node_state = None
-
-            if node_state is None:
-                continue
-
             node_id = build_portable_node_id(node, root)
             current = (node.signature, kind, node.name)
             previous = seen.get(node_id)
@@ -306,9 +304,8 @@ def _raise_error_if_lockfile_ids_are_ambiguous(tasks: list[PTask], root: Path) -
 
     if errors:
         msg = (
-            "Ambiguous lockfile ids detected. Each dependency/product that contributes "
-            "state must map to a unique lockfile id within a task.\n\n"
-            + "\n".join(errors)
+            "Ambiguous lockfile ids detected. Each dependency/product must map to a "
+            "unique lockfile id within a task.\n\n" + "\n".join(errors)
         )
         raise ValueError(msg)
 
@@ -423,6 +420,8 @@ def flush(self) -> None:
 @hookimpl
 def pytask_post_parse(config: dict[str, Any]) -> None:
     """Initialize the lockfile state."""
+    if not _should_initialize_lockfile_state(config.get("command")):
+        return
     path = config["root"] / "pytask.lock"
     config["lockfile_path"] = path
     config["lockfile_state"] = LockfileState.from_path(path, config["root"])
@@ -431,6 +430,8 @@ def pytask_post_parse(config: dict[str, Any]) -> None:
 @hookimpl(trylast=True)
 def pytask_collect_modify_tasks(session: Session, tasks: list[PTask]) -> None:
     """Validate that lockfile ids are unambiguous for collected tasks."""
+    if not _should_validate_lockfile_ids(session.config.get("command")):
+        return
     _raise_error_if_lockfile_ids_are_ambiguous(tasks, session.config["root"])
 
 

tests/test_lockfile.py

@@ -100,7 +100,14 @@ def test_python_node_id_is_collision_free(tmp_path):
     assert left_id != right_id
 
 
-def test_collection_fails_for_ambiguous_lockfile_ids(runner, tmp_path):
+@pytest.mark.parametrize(
+    "args",
+    [
+        pytest.param(lambda path: [path.as_posix()], id="build"),
+        pytest.param(lambda path: ["collect", path.as_posix()], id="collect"),
+    ],
+)
+def test_collection_fails_for_ambiguous_lockfile_ids(runner, tmp_path, args):
     source = """
     from dataclasses import dataclass, field
     from pathlib import Path
@@ -131,6 +138,63 @@ def task_example(
     """
     tmp_path.joinpath("task_module.py").write_text(textwrap.dedent(source))
 
+    result = runner.invoke(cli, args(tmp_path))
+
+    assert result.exit_code == ExitCode.COLLECTION_FAILED
+    assert "Ambiguous lockfile ids detected" in result.output
+    assert "lockfile id 'dup'" in result.output
+
+
+def test_markers_command_ignores_invalid_lockfile(runner, tmp_path):
+    tmp_path.joinpath("pytask.lock").write_text("{not toml")
+
+    result = runner.invoke(cli, ["markers", tmp_path.as_posix()])
+
+    assert result.exit_code == ExitCode.OK
+    assert "persist" in result.output
+
+
+def test_collection_fails_for_ambiguous_lockfile_ids_with_missing_product_state(
+    runner, tmp_path
+):
+    source = """
+    from dataclasses import dataclass, field
+    from pathlib import Path
+    from typing import Annotated, Any
+
+    from pytask import Product
+
+    @dataclass
+    class CustomNode:
+        name: str
+        filepath: Path
+        signature: str
+        attributes: dict[Any, Any] = field(default_factory=dict)
+
+        def state(self):
+            if not self.filepath.exists():
+                return None
+            return self.filepath.read_text()
+
+        def load(self, is_product=False):
+            return self if is_product else self.filepath.read_text()
+
+        def save(self, value):
+            self.filepath.write_text(value)
+
+    def task_example(
+        dependency=CustomNode(
+            name="dup", filepath=Path("in.txt"), signature="signature-a"
+        ),
+        product: Annotated[CustomNode, Product] = CustomNode(
+            name="dup", filepath=Path("out.txt"), signature="signature-b"
+        ),
+    ):
+        product.save(dependency.upper())
+    """
+    tmp_path.joinpath("task_module.py").write_text(textwrap.dedent(source))
+    tmp_path.joinpath("in.txt").write_text("hello")
+
     result = runner.invoke(cli, [tmp_path.as_posix()])
 
     assert result.exit_code == ExitCode.COLLECTION_FAILED