Accept broken symlinks as path arguments of dandi validate

dandi/cli/base.py

@@ -1,8 +1,10 @@
 from enum import Enum
 from functools import wraps
 import os
+from typing import Any
 
 import click
+from click.utils import format_filename
 
 from .. import get_logger
 
@@ -70,6 +72,44 @@ def get_metavar(self, param, ctx=None):
         return "[" + ",".join(self.values) + ",all]"
 
 
+class LinkAwarePath(click.Path):
+    """
+    A ``click.Path`` for commands that should accept broken symlinks
+
+    The typical broken symlink is a git-annex (or DataLad) file whose content
+    has not been fetched.  A command like ``dandi validate`` should still take
+    such a path, to report on it or skip it, rather than refuse it as missing.
+    """
+
+    def __init__(self, *, lexists: bool = False, **kwargs: Any) -> None:
+        """
+        Parameters
+        ----------
+        lexists
+            If True, the path must exist, but a symlink to a missing target
+            counts as existing (unlike with ``exists=True``).  Cannot be
+            combined with ``resolve_path=True``.
+        **kwargs
+            Passed to ``click.Path``.
+        """
+        if lexists and kwargs.get("resolve_path"):
+            raise ValueError("lexists=True cannot be combined with resolve_path=True")
+        super().__init__(**kwargs)
+        self.lexists = lexists
+
+    def convert(
+        self, value: Any, param: click.Parameter | None, ctx: click.Context | None
+    ) -> Any:
+        is_dash = self.file_okay and self.allow_dash and value in ("-", b"-")
+        if self.lexists and not is_dash and not os.path.lexists(value):
+            self.fail(
+                f"{self.name.title()} {format_filename(value)!r} does not exist.",
+                param,
+                ctx,
+            )
+        return super().convert(value, param, ctx)
+
+
 # ???: could make them always available but hidden
 #  via  hidden=True.
 def devel_option(*args, **kwargs):

dandi/cli/cmd_validate.py

@@ -11,7 +11,12 @@
 
 import click
 
-from .base import devel_debug_option, devel_option, map_to_click_exceptions
+from .base import (
+    LinkAwarePath,
+    devel_debug_option,
+    devel_option,
+    map_to_click_exceptions,
+)
 from .formatter import JSONFormatter, JSONLinesFormatter, TextFormatter, YAMLFormatter
 from ..utils import pluralize
 from ..validate._core import validate as validate_
@@ -176,7 +181,7 @@ def _filter_results(
     multiple=True,
     default=(),
 )
-@click.argument("paths", nargs=-1, type=click.Path(exists=True, dir_okay=True))
+@click.argument("paths", nargs=-1, type=LinkAwarePath(lexists=True, dir_okay=True))
 @click.pass_context
 @devel_debug_option()
 @map_to_click_exceptions

dandi/cli/tests/test_base.py

@@ -1,10 +1,12 @@
 from enum import StrEnum
+import os
+from pathlib import Path
 
 import click
 from click.testing import CliRunner
 import pytest
 
-from ..base import EnumChoice
+from ..base import EnumChoice, LinkAwarePath
 
 
 class _Existing(StrEnum):
@@ -69,3 +71,40 @@ def cmd(existing):
     r = CliRunner().invoke(cmd, [])
     assert r.exit_code == 0, r.output
     assert captured["existing"] is _Existing.SKIP
+
+
+@pytest.mark.ai_generated
+def test_path_lexists_rejects_resolve_path():
+    with pytest.raises(ValueError, match="resolve_path"):
+        LinkAwarePath(lexists=True, resolve_path=True)
+
+
+@pytest.mark.ai_generated
+@pytest.mark.parametrize(
+    "kwargs,broken_ok",
+    [
+        ({"lexists": True}, True),
+        ({"lexists": True, "exists": True}, False),
+        ({"exists": True}, False),
+    ],
+)
+def test_path_lexists(tmp_path: Path, kwargs: dict, broken_ok: bool) -> None:
+    @click.command()
+    @click.argument("path", type=LinkAwarePath(allow_dash=True, **kwargs))
+    def cmd(path):
+        click.echo(f"got:{path}")
+
+    (tmp_path / "file.txt").write_text("content")
+    broken = tmp_path / "broken"
+    try:
+        os.symlink(tmp_path / "missing", broken)
+    except OSError:
+        pytest.skip("symlinks are not supported here")
+    for ok in [str(tmp_path / "file.txt"), "-"] + ([str(broken)] if broken_ok else []):
+        r = CliRunner().invoke(cmd, [ok])
+        assert r.exit_code == 0, r.output
+        assert r.output == f"got:{ok}\n"
+    for bad in [str(tmp_path / "missing")] + ([] if broken_ok else [str(broken)]):
+        r = CliRunner().invoke(cmd, [bad])
+        assert r.exit_code == 2
+        assert f"Path {bad!r} does not exist." in r.output

dandi/cli/tests/test_cmd_validate.py

@@ -996,3 +996,25 @@ def test_validate_missing_file_content_no_broken_symlinks(tmp_path: Path) -> Non
     assert r_default.exit_code == r_skip.exit_code == 0
     assert "FILE_CONTENT_MISSING" not in r_default.output
     assert "FILE_CONTENT_MISSING" not in r_skip.output
+
+
+@pytest.mark.ai_generated
+def test_validate_broken_symlink_path_argument(tmp_path: Path) -> None:
+    """A broken symlink can be given directly as a path to validate.
+
+    The --missing-file-content policies then apply to it just as they do when
+    the containing directory is given.
+    """
+    ds = _make_dandiset_with_broken_symlinks(tmp_path)
+    link = ds / "sub-001" / "sub-001.nwb"
+    r = CliRunner().invoke(validate, [str(link)])
+    assert r.exit_code == 1
+    assert "does not exist" not in r.output
+    assert "FILE_CONTENT_MISSING" in r.output
+    r = CliRunner().invoke(validate, ["--missing-file-content", "skip", str(link)])
+    assert r.exit_code == 0
+    assert "FILE_CONTENT_MISSING_SKIPPED" in r.output
+    # Paths that truly do not exist are still rejected
+    r = CliRunner().invoke(validate, [str(ds / "sub-001" / "nonexistent.nwb")])
+    assert r.exit_code == 2
+    assert "does not exist" in r.output