Merge pull request #59 from ArcanaFramework/mount-identification

FS mount identification

Author: tclose

extras/fileformats/extras/application/medical.py

@@ -4,7 +4,7 @@
 from fileformats.core import FileSet
 from fileformats.application import Dicom
 import medimages4tests.dummy.dicom.mri.t1w.siemens.skyra.syngo_d13c
-from fileformats.core.utils import SampleFileGenerator
+from fileformats.core import SampleFileGenerator
 
 
 @FileSet.read_metadata.register

fileformats/application/serialization.py

@@ -5,7 +5,7 @@
 from fileformats.core.mixin import WithClassifiers
 from ..generic import File
 from fileformats.core.exceptions import FormatMismatchError
-from fileformats.core.utils import SampleFileGenerator
+from fileformats.core import SampleFileGenerator
 
 
 class Schema(DataType):

fileformats/core/__init__.py

@@ -3,9 +3,10 @@
 from .datatype import DataType
 from .fileset import FileSet, MockMixin
 from .field import Field
-from .utils import (
+from .identification import (
     to_mime,
     from_mime,
     find_matching,
     from_paths,
 )
+from .sampling import SampleFileGenerator

fileformats/core/datatype.py

@@ -13,6 +13,8 @@
     classproperty,
     subpackages,
     add_exc_note,
+)
+from .identification import (
     to_mime_format_name,
     from_mime_format_name,
     IANA_MIME_TYPE_REGISTRIES,
@@ -125,7 +127,7 @@ def from_mime(cls, mime_string):
         except ValueError:
             raise FormatRecognitionError(
                 f"Format '{mime_string}' is not a valid MIME-like format of <namespace>/<format>"
-            )
+            ) from None
         else:
             namespace = namespace.replace("-", "_")
         # Attempt to load file type using their `iana_mime` attribute

fileformats/core/fileset.py

@@ -16,12 +16,14 @@
 from .utils import (
     classproperty,
     fspaths_converter,
-    to_mime_format_name,
-    IANA_MIME_TYPE_REGISTRIES,
     describe_task,
     matching_source,
     import_extras_module,
-    SampleFileGenerator,
+)
+from .sampling import SampleFileGenerator
+from .identification import (
+    to_mime_format_name,
+    IANA_MIME_TYPE_REGISTRIES,
 )
 from .converter import SubtypeVar
 from .classifier import Classifier
@@ -36,6 +38,8 @@
 )
 from .datatype import DataType
 from . import hook
+from .fs_mount_identifier import FsMountIdentifier
+
 
 try:
     from typing import Self
@@ -1181,6 +1185,7 @@ class CopyMode(Enum):
 
         # All other combinations (typically the result of bit-masking)
 
+        leave_or_copy = 0b1001
         leave_or_symlink = 0b0011
         leave_or_hardlink = 0b0101
         leave_or_link = 0b0111
@@ -1297,20 +1302,49 @@ def copy(
                 if isinstance(collation, str)
                 else collation
             )
-        if new_stem:
+        # Rule out any copy modes that are not supported given the collation mode
+        # and file-system mounts the paths and destination directory reside on
+        constraints = []
+        if FsMountIdentifier.on_cifs(dest_dir) and mode & self.CopyMode.symlink:
+            supported_modes -= self.CopyMode.symlink
+            constraint = (
+                f"Destination directory is on CIFS mount ({dest_dir}) "
+                "and we therefore cannot create a symlink"
+            )
+            logger.debug(constraint)
+            constraints.append(constraint)
+        not_on_same_mount = [
+            p for p in self.fspaths if not FsMountIdentifier.on_same_mount(p, dest_dir)
+        ]
+        if not_on_same_mount and mode & self.CopyMode.hardlink:
+            supported_modes -= self.CopyMode.hardlink
+            constraint = (
+                f"Some paths ({', '.join(str(p) for p in not_on_same_mount)}) are on "
+                f"not on same file-system mount as the destination directory {dest_dir}"
+                "and therefore cannot be hard-linked"
+            )
+            logger.debug(constraint)
+            constraints.append(constraint)
+        if new_stem or (
+            collation >= self.CopyCollation.siblings
+            and not all(p.parent == self.parent for p in self.fspaths)
+        ):
             supported_modes -= self.CopyMode.leave
+
+        # Get the intersection of copy modes that are supported and have been requested
         selected_mode = mode & supported_modes
-        if collation >= self.CopyCollation.siblings:
-            if not all(p.parent == self.parent for p in self.fspaths):
-                selected_mode -= self.CopyMode.leave
         if not selected_mode:
-            raise FileFormatsError(
-                f"Cannot copy {self} using {mode} mode as it is not supported by "
-                f"the {supported_modes} given the collation specification, {collation}"
+            msg = (
+                f"Cannot copy {self} using '{mode}' mode as it is not supported by "
+                f"the '{supported_modes}' given the collation specification, {collation}"
             )
+            if constraints:
+                msg += ", and the following constraints:\n" + "\n".join(constraints)
+            raise FileFormatsError(msg)
         if selected_mode & self.CopyMode.leave:
             return self  # Don't need to do anything
 
+        # Select inner copy/link methods
         if selected_mode & self.CopyMode.symlink:
             copy_dir = copy_file = os.symlink
         elif selected_mode & self.CopyMode.hardlink:
@@ -1339,10 +1373,12 @@ def hardlink_dir(src: Path, dest: Path):
             extension_decomposition=extension_decomposition,
         )
 
-        dest_dir = Path(dest_dir)  # ensure a Path not a string
+        # Prepare destination directory
+        dest_dir = Path(dest_dir)
         if make_dirs:
             dest_dir.mkdir(parents=True, exist_ok=True)
 
+        # Iterate through the paths to copy, copying them to the destination directory
         new_paths = []
         for fspath in fspaths_to_copy:
             new_path, fspath = self._new_copy_path(

fileformats/core/fs_mount_identifier.py

@@ -0,0 +1,143 @@
+import os
+import typing as ty
+from pathlib import Path
+import re
+from contextlib import contextmanager
+import subprocess as sp
+from .utils import logger
+
+
+class FsMountIdentifier:
+    """Used to check the mount type that given file paths reside on in order to determine
+    features that can be used (e.g. symlinks)"""
+
+    @classmethod
+    def on_cifs(cls, path: os.PathLike) -> bool:
+        """
+        Check whether a file path is on a CIFS filesystem mounted in a POSIX host.
+
+        POSIX hosts are assumed to have the ``mount`` command.
+
+        On Windows, Docker mounts host directories into containers through CIFS
+        shares, which has support for Minshall+French symlinks, or text files that
+        the CIFS driver exposes to the OS as symlinks.
+        We have found that under concurrent access to the filesystem, this feature
+        can result in failures to create or read recently-created symlinks,
+        leading to inconsistent behavior and ``FileNotFoundError`` errors.
+
+        This check is written to support disabling symlinks on CIFS shares.
+
+        NB: This function and sub-functions are copied from the nipype.utils.filemanip module
+
+
+        NB: Adapted from https://github.com/nipy/nipype
+        """
+        return cls.get_mount(path)[1] == "cifs"
+
+    @classmethod
+    def on_same_mount(cls, path1: os.PathLike, path2: os.PathLike) -> bool:
+        """Checks whether two or paths are on the same logical file system"""
+        return cls.get_mount(path1)[0] == cls.get_mount(path2)[0]
+
+    @classmethod
+    def get_mount(cls, path: os.PathLike) -> ty.Tuple[Path, str]:
+        """Get the mount point for a given file-system path
+
+        Parameters
+        ----------
+        path: os.PathLike
+            the file-system path to identify the mount of
+
+        Returns
+        -------
+        mount_point: os.PathLike
+            the root of the mount the path sits on
+        fstype : str
+            the type of the file-system (e.g. ext4 or cifs)"""
+        try:
+            # Only the first match (most recent parent) counts, mount table sorted longest
+            # to shortest
+            return next(
+                (Path(p), t)
+                for p, t in cls.get_mount_table()
+                if str(path).startswith(p)
+            )
+        except StopIteration:
+            return (Path("/"), "ext4")
+
+    @classmethod
+    def generate_cifs_table(cls) -> ty.List[ty.Tuple[str, str]]:
+        """
+        Construct a reverse-length-ordered list of mount points that fall under a CIFS mount.
+
+        This precomputation allows efficient checking for whether a given path
+        would be on a CIFS filesystem.
+        On systems without a ``mount`` command, or with no CIFS mounts, returns an
+        empty list.
+
+        """
+        exit_code, output = sp.getstatusoutput("mount")
+        return cls.parse_mount_table(exit_code, output)
+
+    @classmethod
+    def parse_mount_table(
+        cls, exit_code: int, output: str
+    ) -> ty.List[ty.Tuple[str, str]]:
+        """
+        Parse the output of ``mount`` to produce (path, fs_type) pairs.
+
+        Separated from _generate_cifs_table to enable testing logic with real
+        outputs
+
+        """
+        # Not POSIX
+        if exit_code != 0:
+            return []
+
+        # Linux mount example:  sysfs on /sys type sysfs (rw,nosuid,nodev,noexec)
+        #                          <PATH>^^^^      ^^^^^<FSTYPE>
+        # OSX mount example:    /dev/disk2 on / (hfs, local, journaled)
+        #                               <PATH>^  ^^^<FSTYPE>
+        pattern = re.compile(r".*? on (/.*?) (?:type |\()([^\s,\)]+)")
+
+        # Keep line and match for error reporting (match == None on failure)
+        # Ignore empty lines
+        matches = [(ll, pattern.match(ll)) for ll in output.strip().splitlines() if ll]
+
+        # (path, fstype) tuples, sorted by path length (longest first)
+        mount_info = sorted(
+            (match.groups() for _, match in matches if match is not None),
+            key=lambda x: len(x[0]),
+            reverse=True,
+        )
+        cifs_paths = [path for path, fstype in mount_info if fstype.lower() == "cifs"]
+
+        # Report failures as warnings
+        for line, match in matches:
+            if match is None:
+                logger.debug("Cannot parse mount line: '%s'", line)
+
+        return [
+            mount
+            for mount in mount_info
+            if any(mount[0].startswith(path) for path in cifs_paths)
+        ]
+
+    @classmethod
+    def get_mount_table(cls) -> ty.List[ty.Tuple[str, str]]:
+        if cls._mount_table is None:
+            cls._mount_table = cls.generate_cifs_table()
+        return cls._mount_table
+
+    @classmethod
+    @contextmanager
+    def patch_table(cls, mount_table: ty.List[ty.Tuple[str, str]]):
+        """Patch the mount table with new values. Used in test routines"""
+        orig_table = cls._mount_table
+        cls._mount_table = list(mount_table)
+        try:
+            yield
+        finally:
+            cls._mount_table = orig_table
+
+    _mount_table: ty.Optional[ty.List[ty.Tuple[str, str]]] = None