Fix file move time preservation (#558)

CHANGELOG.md

@@ -21,6 +21,7 @@ and this project adheres to [Semantic Versioning](http://semver.org/).
   ([#535](https://github.com/PyFilesystem/pyfilesystem2/issues/535)).
 - Fixed a bug where files could be truncated or deleted when moved / copied onto itself.
   Closes [#546](https://github.com/PyFilesystem/pyfilesystem2/issues/546)
+- Fixed a bug in `FS.move` and `MemoryFS.move`, where `peserve_time=True` resulted in an `ResourceNotFound` error. Closes [#558](https://github.com/PyFilesystem/pyfilesystem2/issues/558)
 
 ## [2.4.16] - 2022-05-02
 

CONTRIBUTORS.md

@@ -25,6 +25,7 @@ Many thanks to the following developers for contributing to this project:
 - [Joshua Tauberer](https://github.com/JoshData)
 - [Justin Charlong](https://github.com/jcharlong)
 - [Louis Sautier](https://github.com/sbraz)
+- [Marcel Johannesmann](https://github.com/mj0nez)
 - [Martin Durant](https://github.com/martindurant)
 - [Martin Larralde](https://github.com/althonos)
 - [Masaya Nakamura](https://github.com/mashabow)

fs/base.py

@@ -22,7 +22,7 @@
 from functools import partial, wraps
 
 from . import copy, errors, fsencode, glob, iotools, tools, walk, wildcard
-from .copy import copy_modified_time
+from .copy import copy_modified_time, read_modified_time, update_details_info
 from .glob import BoundGlobber
 from .mode import validate_open_mode
 from .path import abspath, isbase, join, normpath
@@ -1187,14 +1187,21 @@ def move(self, src_path, dst_path, overwrite=False, preserve_time=False):
             except errors.NoSysPath:  # pragma: no cover
                 pass
             else:
+                # get the time info to preserve it (see #558)
+                if preserve_time:
+                    modification_details = read_modified_time(self, _src_path)
+
                 try:
                     os.rename(src_sys_path, dst_sys_path)
                 except OSError:
                     pass
                 else:
+                    # update the meta info, after moving
                     if preserve_time:
-                        copy_modified_time(self, _src_path, self, _dst_path)
+                        update_details_info(self, _dst_path, modification_details)
+
                     return
+
         with self._lock:
             with self.open(_src_path, "rb") as read_file:
                 # FIXME(@althonos): typing complains because open return IO

fs/copy.py

@@ -14,7 +14,7 @@
 from .walk import Walker
 
 if typing.TYPE_CHECKING:
-    from typing import Callable, Optional, Text, Union
+    from typing import Callable, Optional, Text, Union, Dict
 
     from .base import FS
 
@@ -536,3 +536,43 @@ def copy_modified_time(
                 if value in src_details:
                     dst_details[value] = src_details[value]
             _dst_fs.setinfo(dst_path, {"details": dst_details})
+
+
+def read_modified_time(
+    src_fs,  # type: Union[FS, Text]
+    src_path,  # type: Text
+):
+    # type: (...) -> Dict
+    """Read modified time metadata from a file.
+
+    Arguments:
+        src_fs (FS or str): Source filesystem (instance or URL).
+        src_path (str): Path to a directory on the source filesystem.
+
+    """
+    namespaces = ("details",)
+    with manage_fs(src_fs, writeable=False) as _src_fs:
+        src_meta = _src_fs.getinfo(src_path, namespaces)
+        src_details = src_meta.raw.get("details", {})
+        mod_details = {}
+        for value in ("metadata_changed", "modified"):
+            if value in src_details:
+                mod_details[value] = src_details[value]
+    return mod_details
+
+
+def update_details_info(
+    dst_fs,  # type: Union[FS, Text]
+    dst_path,  # type: Text
+    details_dic,  # type: Dict
+):
+    # type: (...) -> None
+    """Update file metadata from a dict.
+
+    Arguments:
+        dst_fs (FS or str): Destination filesystem (instance or URL).
+        dst_path (str): Path to a directory on the destination filesystem.
+
+    """
+    with manage_fs(dst_fs, create=True) as _dst_fs:
+        _dst_fs.setinfo(dst_path, {"details": details_dic})

fs/memoryfs.py

@@ -15,7 +15,7 @@
 from . import errors
 from ._typing import overload
 from .base import FS
-from .copy import copy_modified_time
+from .copy import copy_modified_time, read_modified_time, update_details_info
 from .enums import ResourceType, Seek
 from .info import Info
 from .mode import Mode
@@ -468,14 +468,19 @@ def move(self, src_path, dst_path, overwrite=False, preserve_time=False):
                     return
                 raise errors.DestinationExists(dst_path)
 
+            # get the time info to preserve it (see #558)
+            if preserve_time:
+                modification_details = read_modified_time(self, src_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)
             # make sure to update the entry name itself (see #509)
             src_entry.name = dst_name
 
+            # update the meta info, after moving
             if preserve_time:
-                copy_modified_time(self, src_path, self, dst_path)
+                update_details_info(self, dst_path, modification_details)
 
     def movedir(self, src_path, dst_path, create=False, preserve_time=False):
         _src_path = self.validatepath(src_path)

fs/test.py

@@ -1789,6 +1789,20 @@ def test_move_file_same_fs(self):
         self.assertEqual(self.fs.listdir("foo"), ["test2.txt"])
         self.assertEqual(next(self.fs.scandir("foo")).name, "test2.txt")
 
+    def test_move_file_same_fs_preserve_time(self):
+        text = "Hello, World"
+        self.fs.makedir("foo").writetext("test.txt", text)
+        self.assert_text("foo/test.txt", text)
+
+        fs.move.move_file(
+            self.fs, "foo/test.txt", self.fs, "foo/test2.txt", preserve_time=True
+        )
+        self.assert_not_exists("foo/test.txt")
+        self.assert_text("foo/test2.txt", text)
+
+        self.assertEqual(self.fs.listdir("foo"), ["test2.txt"])
+        self.assertEqual(next(self.fs.scandir("foo")).name, "test2.txt")
+
     def _test_move_file(self, protocol):
         other_fs = open_fs(protocol)