add cleanup_dest_on_error in move_file

Author: tfeldmann

fs/move.py

@@ -46,6 +46,7 @@ def move_file(
     dst_fs,  # type: Union[Text, FS]
     dst_path,  # type: Text
     preserve_time=False,  # type: bool
+    cleanup_dest_on_error=True,  # type: bool
 ):
     # type: (...) -> None
     """Move a file from one filesystem to another.
@@ -57,6 +58,8 @@ def move_file(
         dst_path (str): Path to a file on ``dst_fs``.
         preserve_time (bool): If `True`, try to preserve mtime of the
             resources (defaults to `False`).
+        cleanup_dest_on_error (bool): If `True`, tries to delete the file copied to
+            dst_fs if deleting the file from src_fs fails.
 
     """
     with manage_fs(src_fs, writeable=True) as _src_fs:
@@ -100,7 +103,8 @@ def move_file(
                 except FSError as e:
                     # if the source cannot be removed we delete the copy on the
                     # destination
-                    _dst_fs.remove(dst_path)
+                    if cleanup_dest_on_error:
+                        _dst_fs.remove(dst_path)
                     raise e
 
 

tests/test_move.py

@@ -2,11 +2,17 @@
 
 import unittest
 
+try:
+    from unittest import mock
+except ImportError:
+    import mock
+
 from parameterized import parameterized_class
 
+
 import fs.move
 from fs import open_fs
-from fs.errors import ResourceReadOnly
+from fs.errors import FSError, ResourceReadOnly
 from fs.path import join
 from fs.wrap import read_only
 
@@ -145,22 +151,24 @@ def test_move_file_read_only_mem_dest(self):
             )
             self.assertTrue(src.exists("file.txt"))
 
-    def test_move_dir_cleanup(self):
-        src_fs = open_fs("mem://")
-        src_fs.makedirs("foo/bar")
-        src_fs.touch("foo/bar/baz.txt")
-        src_fs.touch("foo/test.txt")
-
-        dst_fs = open_fs("mem://")
-        dst_fs.create("test.txt")
-
-        ro_src = read_only(src_fs)
-
-        with self.assertRaises(ResourceReadOnly):
-            fs.move.move_dir(ro_src, "/foo", dst_fs, "/")
+    def test_move_file_cleanup_on_error(self):
+        with open_fs("mem://") as src, open_fs("mem://") as dst:
+            src.writetext("file.txt", "Content")
+            with mock.patch.object(src, "remove") as mck:
+                mck.side_effect = FSError
+                with self.assertRaises(FSError):
+                    fs.move.move_file(src, "file.txt", dst, "file.txt")
+            self.assertTrue(src.exists("file.txt"))
+            self.assertFalse(dst.exists("file.txt"))
 
-        self.assertTrue(src_fs.exists("foo/bar/baz.txt"))
-        self.assertTrue(src_fs.exists("foo/test.txt"))
-        self.assertFalse(dst_fs.isdir("bar"))
-        self.assertFalse(dst_fs.exists("bar/baz.txt"))
-        self.assertTrue(dst_fs.exists("test.txt"))
+    def test_move_file_no_cleanup_on_error(self):
+        with open_fs("mem://") as src, open_fs("mem://") as dst:
+            src.writetext("file.txt", "Content")
+            with mock.patch.object(src, "remove") as mck:
+                mck.side_effect = FSError
+                with self.assertRaises(FSError):
+                    fs.move.move_file(
+                        src, "file.txt", dst, "file.txt", cleanup_dest_on_error=False
+                    )
+            self.assertTrue(src.exists("file.txt"))
+            self.assertTrue(dst.exists("file.txt"))