implemented file-system mount identification & handling to FileSet.copy

Author: tclose

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,

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_mounts.py fileformats/core/fs_mount_identifier.pyfileformats/core/fs_mounts.py renamed to fileformats/core/fs_mount_identifier.py

@@ -7,7 +7,7 @@
 from .utils import logger
 
 
-class FileSystemMountIndentifier:
+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)"""
 

fileformats/core/identification.py

@@ -157,11 +157,11 @@ def to_mime(datatype: type, official: bool = True):
 
 def from_paths(
     fspaths: ty.Iterable[Path],
-    *candidates: ty.Tuple[ty.Type[fileformats.core.FileSet]],
+    *candidates: ty.Tuple[ty.Type["fileformats.core.FileSet"]],
     common_ok: bool = False,
     ignore: ty.Optional[str] = None,
     **kwargs,
-) -> ty.List[fileformats.core.FileSet]:
+) -> ty.List["fileformats.core.FileSet"]:
     """Given a list of candidate classes (defaults to all installed in alphabetical order),
     instantiates all possible file-set instances from a collection of file-system paths.
 

fileformats/core/mixin.py

@@ -4,7 +4,8 @@
 import logging
 from . import hook
 from .fileset import FileSet
-from .utils import classproperty, describe_task, to_mime_format_name, matching_source
+from .utils import classproperty, describe_task, matching_source
+from .identification import to_mime_format_name
 from .converter import SubtypeVar
 from .exceptions import FileFormatsError, FormatMismatchError, FormatRecognitionError
 

fileformats/core/sampling.py

@@ -45,7 +45,7 @@ def rng(self):
 
     def generate(
         self,
-        file_type: ty.Type[fileformats.core.FileSet],
+        file_type: ty.Type["fileformats.core.FileSet"],
         contents: ty.Union[str, bytes] = None,
         fill: int = 0,
         **kwargs,
@@ -99,7 +99,7 @@ def generate(
 
     def generate_fspath(
         self,
-        file_type: ty.Optional[ty.Type[fileformats.core.FileSet]] = None,
+        file_type: ty.Optional[ty.Type["fileformats.core.FileSet"]] = None,
         fname_stem: ty.Optional[str] = None,
         relpath: ty.Optional[Path] = None,
     ):

fileformats/core/tests/test_fs_mounts.py …s/core/tests/test_fs_mount_identifier.pyfileformats/core/tests/test_fs_mounts.py renamed to fileformats/core/tests/test_fs_mount_identifier.py fileformats/core/tests/test_fs_mounts.py renamed to fileformats/core/tests/test_fs_mount_identifier.py

@@ -1,5 +1,7 @@
+import os.path
 import pytest
-from ..fs_mounts import MountIndentifier
+from fileformats.core.fs_mount_identifier import FsMountIdentifier
+from fileformats.generic import File
 
 
 MOUNT_OUTPUTS = (
@@ -158,12 +160,12 @@
 
 @pytest.mark.parametrize("output, exit_code, expected", MOUNT_OUTPUTS)
 def test_parse_mount_table(output, exit_code, expected):
-    assert MountIndentifier.parse_mount_table(exit_code, output) == expected
+    assert FsMountIdentifier.parse_mount_table(exit_code, output) == expected
 
 
 def test_cifs_check():
-    assert isinstance(MountIndentifier.get_mount_table(), list)
-    assert isinstance(MountIndentifier.on_cifs("/"), bool)
+    assert isinstance(FsMountIdentifier.get_mount_table(), list)
+    assert isinstance(FsMountIdentifier.on_cifs("/"), bool)
     fake_table = [("/scratch/tmp", "ext4"), ("/scratch", "cifs")]
     cifs_targets = [
         ("/scratch/tmp/x/y", False),
@@ -175,10 +177,68 @@ def test_cifs_check():
         ("/", False),
     ]
 
-    with MountIndentifier.patch_table([]):
+    with FsMountIdentifier.patch_table([]):
         for target, _ in cifs_targets:
-            assert MountIndentifier.on_cifs(target) is False
+            assert FsMountIdentifier.on_cifs(target) is False
 
-    with MountIndentifier.patch_table(fake_table):
+    with FsMountIdentifier.patch_table(fake_table):
         for target, expected in cifs_targets:
-            assert MountIndentifier.on_cifs(target) is expected
+            assert FsMountIdentifier.on_cifs(target) is expected
+
+
+def test_copy_constraints(tmp_path):
+
+    ext4_mnt1 = tmp_path / "ext4_mnt1"
+    ext4_mnt2 = tmp_path / "ext4_mnt2"
+    cifs_mnt = tmp_path / "cifs_mnt"
+
+    fake_mount_table = [
+        (str(ext4_mnt1), "ext4"),
+        (str(ext4_mnt2), "ext4"),
+        (str(cifs_mnt), "cifs"),
+    ]
+
+    # Create sample files
+    ext4_file = File.sample(dest_dir=ext4_mnt1, seed=1)
+    cifs_file = File.sample(dest_dir=cifs_mnt, seed=3)
+
+    with FsMountIdentifier.patch_table(fake_mount_table):
+        # Check that symlinks work on ext4
+        copy_modes = File.CopyMode.copy | File.CopyMode.hardlink | File.CopyMode.symlink
+        new_ext4_file = ext4_file.copy(
+            ext4_mnt1 / "dest",
+            mode=copy_modes,
+        )
+
+        assert new_ext4_file.contents == ext4_file.contents
+        assert os.path.islink(new_ext4_file)
+
+        # Symlinks not supported on CIFS
+        new_cifs_file = cifs_file.copy(
+            cifs_mnt / "dest",
+            mode=copy_modes,
+        )
+        assert new_cifs_file.contents == cifs_file.contents
+        assert not os.path.islink(new_cifs_file)
+        assert os.stat(new_cifs_file).st_ino == os.stat(cifs_file).st_ino  # Hardlink
+
+        # Hardlinks not supported across logical volumes
+        new_ext4_file2 = ext4_file.copy(
+            ext4_mnt2 / "dest", mode=File.CopyMode.copy | File.CopyMode.hardlink
+        )
+        assert new_ext4_file2.contents == ext4_file.contents
+        assert not os.path.islink(new_ext4_file2)
+        assert (
+            os.stat(ext4_file).st_ino != os.stat(new_ext4_file2).st_ino
+        )  # Not hardlink
+
+        # Hardlinks not supported across logical volumes 2 (from CIFS)
+        ext4_file_on_cifs = ext4_file.copy(
+            cifs_mnt / "dest",
+            mode=copy_modes,
+        )
+        assert ext4_file_on_cifs.contents == ext4_file.contents
+        assert not os.path.islink(ext4_file_on_cifs)
+        assert (
+            os.stat(ext4_file).st_ino != os.stat(ext4_file_on_cifs).st_ino
+        )  # Not hardlink

fileformats/core/tests/test_mime.py

@@ -1,5 +1,5 @@
 from fileformats.generic import FileSet
-from fileformats.core.utils import from_mime
+from fileformats.core.identification import from_mime
 from fileformats.testing import Classified, U, V
 from fileformats.testing_subpackage import Psi, SubpackageClassified, Zeta, Theta
 

fileformats/core/tests/test_utils.py

@@ -205,13 +205,13 @@ def test_copy_collation_leave_diff_dir(work_dir: Path, dest_dir: Path):
 
     with pytest.raises(
         FileFormatsError,
-        match="using leave mode as it is not supported by the any given the collation specification",
+        match="given the collation specification",
     ):
         fileset.copy(dest_dir=dest_dir, mode="leave", collation="siblings")
 
     with pytest.raises(
         FileFormatsError,
-        match="using leave mode as it is not supported by the any given the collation specification",
+        match="given the collation specification",
     ):
         fileset.copy(dest_dir=dest_dir, mode="leave", collation="adjacent")
 

fileformats/core/utils.py

@@ -79,7 +79,7 @@ def fspaths_converter(
         str,
         os.PathLike,
         bytes,
-        fileformats.core.FileSet,
+        "fileformats.core.FileSet",
     ]
 ):
     """Ensures fs-paths are a set of pathlib.Path"""