Bugfix: File deleted / emptied when moved / copied onto itself (#546)

CHANGELOG.md

@@ -16,7 +16,8 @@ and this project adheres to [Semantic Versioning](http://semver.org/).
 ### Fixed
 - Elaborated documentation of `filter_dirs` and `exclude_dirs` in `fs.walk.Walker`.
   Closes [#371](https://github.com/PyFilesystem/pyfilesystem2/issues/371).
-
+- Fixed a bug where files could be truncated or deleted when moved / copied onto itself.
+  Closes [#546](https://github.com/PyFilesystem/pyfilesystem2/issues/546)
 
 ## [2.4.16] - 2022-05-02
 

fs/base.py

@@ -21,11 +21,11 @@
 from contextlib import closing
 from functools import partial, wraps
 
-from . import copy, errors, fsencode, iotools, tools, walk, wildcard, glob
+from . import copy, errors, fsencode, glob, iotools, tools, walk, wildcard
 from .copy import copy_modified_time
 from .glob import BoundGlobber
 from .mode import validate_open_mode
-from .path import abspath, join, normpath
+from .path import abspath, isbase, join, normpath
 from .time import datetime_to_epoch
 from .walk import Walker
 
@@ -423,13 +423,17 @@ def copy(
 
         """
         with self._lock:
-            if not overwrite and self.exists(dst_path):
+            _src_path = self.validatepath(src_path)
+            _dst_path = self.validatepath(dst_path)
+            if not overwrite and self.exists(_dst_path):
                 raise errors.DestinationExists(dst_path)
-            with closing(self.open(src_path, "rb")) as read_file:
+            if _src_path == _dst_path:
+                raise errors.IllegalDestination(dst_path)
+            with closing(self.open(_src_path, "rb")) as read_file:
                 # FIXME(@althonos): typing complains because open return IO
-                self.upload(dst_path, read_file)  # type: ignore
+                self.upload(_dst_path, read_file)  # type: ignore
             if preserve_time:
-                copy_modified_time(self, src_path, self, dst_path)
+                copy_modified_time(self, _src_path, self, _dst_path)
 
     def copydir(
         self,
@@ -457,11 +461,15 @@ def copydir(
 
         """
         with self._lock:
-            if not create and not self.exists(dst_path):
+            _src_path = self.validatepath(src_path)
+            _dst_path = self.validatepath(dst_path)
+            if isbase(_src_path, _dst_path):
+                raise errors.IllegalDestination(dst_path)
+            if not create and not self.exists(_dst_path):
                 raise errors.ResourceNotFound(dst_path)
-            if not self.getinfo(src_path).is_dir:
+            if not self.getinfo(_src_path).is_dir:
                 raise errors.DirectoryExpected(src_path)
-            copy.copy_dir(self, src_path, self, dst_path, preserve_time=preserve_time)
+            copy.copy_dir(self, _src_path, self, _dst_path, preserve_time=preserve_time)
 
     def create(self, path, wipe=False):
         # type: (Text, bool) -> bool
@@ -1088,6 +1096,12 @@ def movedir(self, src_path, dst_path, create=False, preserve_time=False):
         from .move import move_dir
 
         with self._lock:
+            _src_path = self.validatepath(src_path)
+            _dst_path = self.validatepath(dst_path)
+            if _src_path == _dst_path:
+                return
+            if isbase(_src_path, _dst_path):
+                raise errors.IllegalDestination(dst_path)
             if not create and not self.exists(dst_path):
                 raise errors.ResourceNotFound(dst_path)
             move_dir(self, src_path, self, dst_path, preserve_time=preserve_time)
@@ -1157,14 +1171,19 @@ def move(self, src_path, dst_path, overwrite=False, preserve_time=False):
                 ``dst_path`` does not exist.
 
         """
-        if not overwrite and self.exists(dst_path):
+        _src_path = self.validatepath(src_path)
+        _dst_path = self.validatepath(dst_path)
+        if not overwrite and self.exists(_dst_path):
             raise errors.DestinationExists(dst_path)
-        if self.getinfo(src_path).is_dir:
+        if self.getinfo(_src_path).is_dir:
             raise errors.FileExpected(src_path)
+        if _src_path == _dst_path:
+            # early exit when moving a file onto itself
+            return
         if self.getmeta().get("supports_rename", False):
             try:
-                src_sys_path = self.getsyspath(src_path)
-                dst_sys_path = self.getsyspath(dst_path)
+                src_sys_path = self.getsyspath(_src_path)
+                dst_sys_path = self.getsyspath(_dst_path)
             except errors.NoSysPath:  # pragma: no cover
                 pass
             else:
@@ -1174,15 +1193,15 @@ def move(self, src_path, dst_path, overwrite=False, preserve_time=False):
                     pass
                 else:
                     if preserve_time:
-                        copy_modified_time(self, src_path, self, dst_path)
+                        copy_modified_time(self, _src_path, self, _dst_path)
                     return
         with self._lock:
-            with self.open(src_path, "rb") as read_file:
+            with self.open(_src_path, "rb") as read_file:
                 # FIXME(@althonos): typing complains because open return IO
-                self.upload(dst_path, read_file)  # type: ignore
+                self.upload(_dst_path, read_file)  # type: ignore
             if preserve_time:
-                copy_modified_time(self, src_path, self, dst_path)
-            self.remove(src_path)
+                copy_modified_time(self, _src_path, self, _dst_path)
+            self.remove(_src_path)
 
     def open(
         self,

fs/copy.py

@@ -7,9 +7,9 @@
 
 import warnings
 
-from .errors import ResourceNotFound
+from .errors import IllegalDestination, ResourceNotFound
 from .opener import manage_fs
-from .path import abspath, combine, frombase, normpath
+from .path import abspath, combine, frombase, isbase, normpath
 from .tools import is_thread_safe
 from .walk import Walker
 
@@ -257,9 +257,13 @@ def copy_file_internal(
         lock (bool): Lock both filesystems before copying.
 
     """
+    _src_path = src_fs.validatepath(src_path)
+    _dst_path = dst_fs.validatepath(dst_path)
     if src_fs is dst_fs:
-        # Same filesystem, so we can do a potentially optimized
-        # copy
+        # It's not allowed to copy a file onto itself
+        if _src_path == _dst_path:
+            raise IllegalDestination(dst_path)
+        # Same filesystem, so we can do a potentially optimized copy
         src_fs.copy(src_path, dst_path, overwrite=True, preserve_time=preserve_time)
         return
 
@@ -305,11 +309,18 @@ def copy_structure(
     walker = walker or Walker()
     with manage_fs(src_fs) as _src_fs:
         with manage_fs(dst_fs, create=True) as _dst_fs:
+            _src_root = _src_fs.validatepath(src_root)
+            _dst_root = _dst_fs.validatepath(dst_root)
+
+            # It's not allowed to copy a structure into itself
+            if _src_fs == _dst_fs and isbase(_src_root, _dst_root):
+                raise IllegalDestination(dst_root)
+
             with _src_fs.lock(), _dst_fs.lock():
-                _dst_fs.makedirs(dst_root, recreate=True)
-                for dir_path in walker.dirs(_src_fs, src_root):
+                _dst_fs.makedirs(_dst_root, recreate=True)
+                for dir_path in walker.dirs(_src_fs, _src_root):
                     _dst_fs.makedir(
-                        combine(dst_root, frombase(src_root, dir_path)), recreate=True
+                        combine(_dst_root, frombase(_src_root, dir_path)), recreate=True
                     )
 
 

fs/errors.py

@@ -32,6 +32,7 @@
     "FilesystemClosed",
     "FSError",
     "IllegalBackReference",
+    "IllegalDestination",
     "InsufficientStorage",
     "InvalidCharsInPath",
     "InvalidPath",
@@ -240,6 +241,16 @@ class RemoveRootError(OperationFailed):
     default_message = "root directory may not be removed"
 
 
+class IllegalDestination(OperationFailed):
+    """The given destination cannot be used for the operation.
+
+    This error will occur when attempting to move / copy a folder into itself or copying
+    a file onto itself.
+    """
+
+    default_message = "'{path}' is not a legal destination"
+
+
 class ResourceError(FSError):
     """Base exception class for error associated with a specific resource."""
 

fs/memoryfs.py

@@ -19,7 +19,7 @@
 from .enums import ResourceType, Seek
 from .info import Info
 from .mode import Mode
-from .path import iteratepath, normpath, split
+from .path import isbase, iteratepath, normpath, split
 
 if typing.TYPE_CHECKING:
     from typing import (
@@ -462,6 +462,12 @@ def move(self, src_path, dst_path, overwrite=False, preserve_time=False):
             elif not overwrite and dst_name in dst_dir_entry:
                 raise errors.DestinationExists(dst_path)
 
+            # handle moving a file onto itself
+            if src_dir == dst_dir and src_name == dst_name:
+                if overwrite:
+                    return
+                raise errors.DestinationExists(dst_path)
+
             # move the entry from the src folder to the dst folder
             dst_dir_entry.set_entry(dst_name, src_entry)
             src_dir_entry.remove_entry(src_name)
@@ -472,8 +478,17 @@ def move(self, src_path, dst_path, overwrite=False, preserve_time=False):
                 copy_modified_time(self, src_path, self, dst_path)
 
     def movedir(self, src_path, dst_path, create=False, preserve_time=False):
-        src_dir, src_name = split(self.validatepath(src_path))
-        dst_dir, dst_name = split(self.validatepath(dst_path))
+        _src_path = self.validatepath(src_path)
+        _dst_path = self.validatepath(dst_path)
+        dst_dir, dst_name = split(_dst_path)
+        src_dir, src_name = split(_src_path)
+
+        # move a dir onto itself
+        if _src_path == _dst_path:
+            return
+        # move a dir into itself
+        if isbase(_src_path, _dst_path):
+            raise errors.IllegalDestination(dst_path)
 
         with self._lock:
             src_dir_entry = self._get_dir_entry(src_dir)

fs/osfs.py

@@ -411,6 +411,9 @@ def _check_copy(self, src_path, dst_path, overwrite=False):
         # check dst_path does not exist if we are not overwriting
         if not overwrite and self.exists(_dst_path):
             raise errors.DestinationExists(dst_path)
+        # it's not allowed to copy a file onto itself
+        if _src_path == _dst_path:
+            raise errors.IllegalDestination(dst_path)
         # check parent dir of _dst_path exists and is a directory
         if self.gettype(dirname(dst_path)) is not ResourceType.directory:
             raise errors.DirectoryExpected(dirname(dst_path))

fs/test.py

@@ -1811,6 +1811,40 @@ def test_move_file_mem(self):
     def test_move_file_temp(self):
         self._test_move_file("temp://")
 
+    def test_move_file_onto_itself(self):
+        self.fs.writetext("file.txt", "Hello")
+        self.fs.move("file.txt", "file.txt", overwrite=True)
+        self.assert_text("file.txt", "Hello")
+
+        with self.assertRaises(errors.DestinationExists):
+            self.fs.move("file.txt", "file.txt", overwrite=False)
+
+    def test_move_file_onto_itself_relpath(self):
+        subdir = self.fs.makedir("sub")
+        subdir.writetext("file.txt", "Hello")
+        self.fs.move("sub/file.txt", "sub/../sub/file.txt", overwrite=True)
+        self.assert_text("sub/file.txt", "Hello")
+
+        with self.assertRaises(errors.DestinationExists):
+            self.fs.move("sub/file.txt", "sub/../sub/file.txt", overwrite=False)
+
+    def test_copy_file_onto_itself(self):
+        self.fs.writetext("file.txt", "Hello")
+        with self.assertRaises(errors.IllegalDestination):
+            self.fs.copy("file.txt", "file.txt", overwrite=True)
+        with self.assertRaises(errors.DestinationExists):
+            self.fs.copy("file.txt", "file.txt", overwrite=False)
+        self.assert_text("file.txt", "Hello")
+
+    def test_copy_file_onto_itself_relpath(self):
+        subdir = self.fs.makedir("sub")
+        subdir.writetext("file.txt", "Hello")
+        with self.assertRaises(errors.IllegalDestination):
+            self.fs.copy("sub/file.txt", "sub/../sub/file.txt", overwrite=True)
+        with self.assertRaises(errors.DestinationExists):
+            self.fs.copy("sub/file.txt", "sub/../sub/file.txt", overwrite=False)
+        self.assert_text("sub/file.txt", "Hello")
+
     def test_copydir(self):
         self.fs.makedirs("foo/bar/baz/egg")
         self.fs.writetext("foo/bar/foofoo.txt", "Hello")
@@ -1828,6 +1862,27 @@ def test_copydir(self):
         with self.assertRaises(errors.DirectoryExpected):
             self.fs.copydir("foo2/foofoo.txt", "foofoo.txt", create=True)
 
+    def test_copydir_onto_itself(self):
+        folder = self.fs.makedir("folder")
+        folder.writetext("file1.txt", "Hello1")
+        sub = folder.makedir("sub")
+        sub.writetext("file2.txt", "Hello2")
+
+        with self.assertRaises(errors.IllegalDestination):
+            self.fs.copydir("folder", "folder")
+        self.assert_text("folder/file1.txt", "Hello1")
+        self.assert_text("folder/sub/file2.txt", "Hello2")
+
+    def test_copydir_into_its_own_subfolder(self):
+        folder = self.fs.makedir("folder")
+        folder.writetext("file1.txt", "Hello1")
+        sub = folder.makedir("sub")
+        sub.writetext("file2.txt", "Hello2")
+        with self.assertRaises(errors.IllegalDestination):
+            self.fs.copydir("folder", "folder/sub/")
+        self.assert_text("folder/file1.txt", "Hello1")
+        self.assert_text("folder/sub/file2.txt", "Hello2")
+
     def test_movedir(self):
         self.fs.makedirs("foo/bar/baz/egg")
         self.fs.writetext("foo/bar/foofoo.txt", "Hello")
@@ -1851,6 +1906,27 @@ def test_movedir(self):
         with self.assertRaises(errors.DirectoryExpected):
             self.fs.movedir("foo2/foofoo.txt", "foo2/baz/egg")
 
+    def test_movedir_onto_itself(self):
+        folder = self.fs.makedir("folder")
+        folder.writetext("file1.txt", "Hello1")
+        sub = folder.makedir("sub")
+        sub.writetext("file2.txt", "Hello2")
+
+        self.fs.movedir("folder", "folder")
+        self.assert_text("folder/file1.txt", "Hello1")
+        self.assert_text("folder/sub/file2.txt", "Hello2")
+
+    def test_movedir_into_its_own_subfolder(self):
+        folder = self.fs.makedir("folder")
+        folder.writetext("file1.txt", "Hello1")
+        sub = folder.makedir("sub")
+        sub.writetext("file2.txt", "Hello2")
+
+        with self.assertRaises(errors.IllegalDestination):
+            self.fs.movedir("folder", "folder/sub/")
+        self.assert_text("folder/file1.txt", "Hello1")
+        self.assert_text("folder/sub/file2.txt", "Hello2")
+
     def test_match(self):
         self.assertTrue(self.fs.match(["*.py"], "foo.py"))
         self.assertEqual(