nix-required-mounts: correctly handle relative paths in symlink_targets()

pkgs/by-name/ni/nix-required-mounts/nix_required_mounts.py

@@ -49,14 +49,28 @@ class Pattern(TypedDict):
 parser.add_argument("-v", "--verbose", action="count", default=0)
 
 
-def symlink_parents(p: Path) -> List[Path]:
+def symlink_targets(p: Path) -> List[Path]:
+    """
+    >>> from pathlib import Path
+    >>> from tempfile import TemporaryDirectory
+    >>> with TemporaryDirectory() as d:
+    ...     Path(d, "a").touch()
+    ...     Path(d, "b").symlink_to("a")
+    ...     Path(d, "c").symlink_to(Path(d, "b"))
+    ...     targets = [str(p).replace(d, "$TMPDIR") for p in symlink_targets(Path(d, "c"))]
+    >>> targets
+    ['$TMPDIR/b', '$TMPDIR/a']
+    """
     out = []
-    while p.is_symlink() and p not in out:
-        parent = p.readlink()
-        if parent.is_relative_to("."):
-            p = p / parent
+    while p.is_symlink():
+        target = p.readlink()
+        if target.is_absolute():
+            p = target
         else:
-            p = parent
+            p = p.parent / target
+            p = p.absolute()
+        if p in out:
+            break
         out.append(p)
     return out
 
@@ -70,20 +84,26 @@ def get_required_system_features(parsed_drv: dict) -> List[str]:
     # Older versions of Nix store structuredAttrs in the env as a JSON string.
     drv_env = parsed_drv.get("env", {})
     if "__json" in drv_env:
-        return list(json.loads(drv_env["__json"]).get("requiredSystemFeatures", []))
+        return list(
+            json.loads(drv_env["__json"]).get("requiredSystemFeatures", [])
+        )
 
     # Without structuredAttrs, requiredSystemFeatures is a space-separated string in env.
     return drv_env.get("requiredSystemFeatures", "").split()
 
 
-def validate_mounts(pattern: Pattern) -> List[Tuple[PathString, PathString, bool]]:
+def validate_mounts(
+    pattern: Pattern,
+) -> List[Tuple[PathString, PathString, bool]]:
     roots = []
     for mount in pattern["paths"]:
         if isinstance(mount, PathString):
             matches = glob.glob(mount)
             assert matches, f"Specified host paths do not exist: {mount}"
 
-            roots.extend((m, m, pattern["unsafeFollowSymlinks"]) for m in matches)
+            roots.extend(
+                (m, m, pattern["unsafeFollowSymlinks"]) for m in matches
+            )
         else:
             assert isinstance(mount, dict) and "host" in mount, mount
             assert Path(
@@ -136,7 +156,9 @@ def entrypoint():
             "`nix show-derivation`"
             f". Expected JSON, observed: {proc.stdout}",
         )
-        logging.error(textwrap.indent(proc.stdout.decode("utf8"), prefix=" " * 4))
+        logging.error(
+            textwrap.indent(proc.stdout.decode("utf8"), prefix=" " * 4)
+        )
         logging.info("Exiting the nix-required-binds hook")
         return
     [canon_drv_path] = parsed_drv.keys()
@@ -149,13 +171,17 @@ def entrypoint():
 
     parsed_drv = parsed_drv[canon_drv_path]
     required_features = get_required_system_features(parsed_drv)
-    required_features = list(filter(known_features.__contains__, required_features))
+    required_features = list(
+        filter(known_features.__contains__, required_features)
+    )
 
     patterns: List[Pattern] = list(
         pattern
         for pattern in allowed_patterns.values()
         for path in pattern["paths"]
-        if any(feature in required_features for feature in pattern["onFeatures"])
+        if any(
+            feature in required_features for feature in pattern["onFeatures"]
+        )
     )  # noqa: E501
 
     queue: Deque[Tuple[PathString, PathString, bool]] = deque(
@@ -180,10 +206,12 @@ def entrypoint():
 
         # assert host_path_str == guest_path_str, (host_path_str, guest_path_str)
 
-        for child in host_path.iterdir() if host_path.is_dir() else [host_path]:
-            for parent in symlink_parents(child):
-                parent_str = parent.absolute().as_posix()
-                queue.append((parent_str, parent_str, follow_symlinks))
+        for child in (
+            host_path.iterdir() if host_path.is_dir() else [host_path]
+        ):
+            for target in symlink_targets(child):
+                target_str = target.absolute().as_posix()
+                queue.append((target_str, target_str, follow_symlinks))
 
     # the pre-build-hook command
     if args.issue_command == "always" or (

pkgs/by-name/ni/nix-required-mounts/package.nix

@@ -43,6 +43,13 @@ python3Packages.buildPythonApplication {
     makeWrapper
     python3Packages.setuptools
   ];
+  checkInputs = [
+    python3Packages.pytestCheckHook
+  ];
+  doCheck = true;
+  pythonImportsCheck = [
+    "nix_required_mounts"
+  ];
 
   postFixup = ''
     wrapProgram $out/bin/${pname} \

pkgs/by-name/ni/nix-required-mounts/pyproject.toml

@@ -18,3 +18,6 @@ nix-required-mounts = "nix_required_mounts:entrypoint"
 
 [tool.black]
 line-length = 79
+
+[tool.pytest.ini_options]
+addopts = ["--doctest-modules"]

pkgs/by-name/ni/nix-required-mounts/scripts/nix_required_mounts_closure.py

@@ -1,45 +1,47 @@
 import json
 import os
 
-store_dir = os.environ["storeDir"]
-
-with open(os.environ["shallowConfigPath"], "r") as f:
-    config = json.load(f)
-
-cache = {}
-
-
-def read_edges(path: str | dict) -> list[str | dict]:
-    if isinstance(path, dict):
-        return [path]
-    assert isinstance(path, str)
-
-    if not path.startswith(store_dir):
-        return [path]
-    if path in cache:
-        return cache[path]
-
-    name = f"references-{path.removeprefix(store_dir)}"
-
-    assert os.path.exists(name)
-
-    with open(name, "r") as f:
-        return [p.strip() for p in f.readlines() if p.startswith(store_dir)]
-
-
-def host_path(mount: str | dict) -> str:
-    if isinstance(mount, dict):
-        return mount["host"]
-    assert isinstance(mount, str), mount
-    return mount
-
-
-for pattern in config:
-    closure = []
-    for path in config[pattern]["paths"]:
-        closure.append(path)
-        closure.extend(read_edges(path))
-    config[pattern]["paths"] = list({host_path(m): m for m in closure}.values())
-
-with open(os.environ["out"], "w") as f:
-    json.dump(config, f)
+if __name__ == "__main__":
+    store_dir = os.environ.get("storeDir", "/nix/store")
+
+    with open(os.environ["shallowConfigPath"], "r") as f:
+        config = json.load(f)
+
+    cache = {}
+
+    def read_edges(path: str | dict) -> list[str | dict]:
+        if isinstance(path, dict):
+            return [path]
+        assert isinstance(path, str)
+
+        if not path.startswith(store_dir):
+            return [path]
+        if path in cache:
+            return cache[path]
+
+        name = f"references-{path.removeprefix(store_dir)}"
+
+        assert os.path.exists(name)
+
+        with open(name, "r") as f:
+            return [
+                p.strip() for p in f.readlines() if p.startswith(store_dir)
+            ]
+
+    def host_path(mount: str | dict) -> str:
+        if isinstance(mount, dict):
+            return mount["host"]
+        assert isinstance(mount, str), mount
+        return mount
+
+    for pattern in config:
+        closure = []
+        for path in config[pattern]["paths"]:
+            closure.append(path)
+            closure.extend(read_edges(path))
+        config[pattern]["paths"] = list(
+            {host_path(m): m for m in closure}.values()
+        )
+
+    with open(os.environ["out"], "w") as f:
+        json.dump(config, f)