added mount identifier class and split up utils module

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/fs_mounts.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 FileSystemMountIndentifier:
+    """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