Merge 3d56a46 into f6a6195

CHANGELOG.md

@@ -33,6 +33,9 @@ and this project adheres to [Semantic Versioning](http://semver.org/).
   test suites.
 - `FSTestCases` now builds the large data required for `upload` and `download` tests only
   once in order to reduce the total testing time.
+- FTP servers that do not support the MLST command now try to use the MDTM command to
+  retrieve the last modification timestamp of a resource. 
+  Closes [#456](https://github.com/PyFilesystem/pyfilesystem2/pull/456).
 - `MemoryFS.move` and `MemoryFS.movedir` will now avoid copying data.
   Closes [#452](https://github.com/PyFilesystem/pyfilesystem2/issues/452).
 - `FS.removetree("/")` behaviour has been standardized in all filesystems, and

fs/base.py

@@ -678,6 +678,25 @@ def readtext(
 
     gettext = _new_name(readtext, "gettext")
 
+    def getmodified(self, path):
+        # type: (Text) -> Optional[datetime]
+        """Get the timestamp of the last modifying access of a resource.
+
+        Arguments:
+            path (str): A path to a resource.
+
+        Returns:
+            datetime: The timestamp of the last modification.
+
+        The *modified timestamp* of a file is the point in time
+        that the file was last changed. Depending on the file system,
+        it might only have limited accuracy.
+
+        """
+        if self.getmeta().get("supports_mtime", False):
+            return self.getinfo(path, namespaces=["modified"]).modified
+        return self.getinfo(path, namespaces=["details"]).modified
+
     def getmeta(self, namespace="standard"):
         # type: (Text) -> Mapping[Text, object]
         """Get meta information regarding a filesystem.
@@ -715,6 +734,8 @@ def getmeta(self, namespace="standard"):
         read_only           `True` if this filesystem is read only.
         supports_rename     `True` if this filesystem supports an
                             `os.rename` operation.
+        supports_mtime      `True` if this filesystem supports a native
+                            operation to retreive the "last modified" time.
         =================== ============================================
 
         Most builtin filesystems will provide all these keys, and third-

fs/copy.py

@@ -97,10 +97,9 @@ def _source_is_newer(src_fs, src_path, dst_fs, dst_path):
     """
     try:
         if dst_fs.exists(dst_path):
-            namespace = ("details", "modified")
-            src_modified = src_fs.getinfo(src_path, namespace).modified
+            src_modified = src_fs.getmodified(src_path)
             if src_modified is not None:
-                dst_modified = dst_fs.getinfo(dst_path, namespace).modified
+                dst_modified = dst_fs.getmodified(dst_path)
                 return dst_modified is None or src_modified > dst_modified
         return True
     except FSError:  # pragma: no cover

fs/ftpfs.py

@@ -15,6 +15,7 @@
 from contextlib import contextmanager
 from ftplib import FTP
 
+
 try:
     from ftplib import FTP_TLS
 except ImportError as err:
@@ -667,6 +668,18 @@ def getinfo(self, path, namespaces=None):
                 }
             )
 
+        if "modified" in namespaces:
+            if "basic" in namespaces or "details" in namespaces:
+                raise ValueError(
+                    'Cannot use the "modified" namespace in combination with others.'
+                )
+            with self._lock:
+                with ftp_errors(self, path=path):
+                    cmd = "MDTM " + _encode(self.validatepath(path), self.ftp.encoding)
+                    response = self.ftp.sendcmd(cmd)
+                modified_info = {"modified": self._parse_ftp_time(response.split()[1])}
+                return Info({"modified": modified_info})
+
         if self.supports_mlst:
             with self._lock:
                 with ftp_errors(self, path=path):
@@ -692,6 +705,7 @@ def getmeta(self, namespace="standard"):
         if namespace == "standard":
             _meta = self._meta.copy()
             _meta["unicode_paths"] = "UTF8" in self.features
+            _meta["supports_mtime"] = "MDTM" in self.features
         return _meta
 
     def listdir(self, path):

fs/info.py

@@ -317,9 +317,12 @@ def modified(self):
                 namespace is not in the Info.
 
         """
-        self._require_namespace("details")
-        _time = self._make_datetime(self.get("details", "modified"))
-        return _time
+        try:
+            self._require_namespace("details")
+            return self._make_datetime(self.get("details", "modified"))
+        except MissingInfoNamespace:
+            self._require_namespace("modified")
+            return self._make_datetime(self.get("modified", "modified"))
 
     @property
     def created(self):

fs/osfs.py

@@ -144,6 +144,7 @@ def __init__(
             "network": False,
             "read_only": False,
             "supports_rename": True,
+            "supports_mtime": False,
             "thread_safe": True,
             "unicode_paths": os.path.supports_unicode_filenames,
             "virtual": False,

tests/test_memoryfs.py

@@ -69,7 +69,6 @@ def test_close_mem_free(self):
 
 
 class TestMemoryFile(unittest.TestCase):
-
     def setUp(self):
         self.fs = memoryfs.MemoryFS()
 

tests/test_opener.py

@@ -300,14 +300,26 @@ def test_user_data_opener(self, app_dir):
     def test_open_ftp(self, mock_FTPFS):
         open_fs("ftp://foo:bar@ftp.example.org")
         mock_FTPFS.assert_called_once_with(
-            "ftp.example.org", passwd="bar", port=21, user="foo", proxy=None, timeout=10, tls=False
+            "ftp.example.org",
+            passwd="bar",
+            port=21,
+            user="foo",
+            proxy=None,
+            timeout=10,
+            tls=False,
         )
 
     @mock.patch("fs.ftpfs.FTPFS")
     def test_open_ftps(self, mock_FTPFS):
         open_fs("ftps://foo:bar@ftp.example.org")
         mock_FTPFS.assert_called_once_with(
-            "ftp.example.org", passwd="bar", port=21, user="foo", proxy=None, timeout=10, tls=True
+            "ftp.example.org",
+            passwd="bar",
+            port=21,
+            user="foo",
+            proxy=None,
+            timeout=10,
+            tls=True,
         )
 
     @mock.patch("fs.ftpfs.FTPFS")

tests/test_wrap.py

@@ -177,7 +177,7 @@ def test_scandir(self):
         ]
         with mock.patch.object(self.fs, "scandir", wraps=self.fs.scandir) as scandir:
             self.assertEqual(sorted(self.cached.scandir("/"), key=key), expected)
-            scandir.assert_has_calls([mock.call('/', namespaces=None, page=None)])
+            scandir.assert_has_calls([mock.call("/", namespaces=None, page=None)])
         with mock.patch.object(self.fs, "scandir", wraps=self.fs.scandir) as scandir:
             self.assertEqual(sorted(self.cached.scandir("/"), key=key), expected)
             scandir.assert_not_called()
@@ -187,7 +187,7 @@ def test_isdir(self):
             self.assertTrue(self.cached.isdir("foo"))
             self.assertFalse(self.cached.isdir("egg"))  # is file
             self.assertFalse(self.cached.isdir("spam"))  # doesn't exist
-            scandir.assert_has_calls([mock.call('/', namespaces=None, page=None)])
+            scandir.assert_has_calls([mock.call("/", namespaces=None, page=None)])
         with mock.patch.object(self.fs, "scandir", wraps=self.fs.scandir) as scandir:
             self.assertTrue(self.cached.isdir("foo"))
             self.assertFalse(self.cached.isdir("egg"))
@@ -199,7 +199,7 @@ def test_isfile(self):
             self.assertTrue(self.cached.isfile("egg"))
             self.assertFalse(self.cached.isfile("foo"))  # is dir
             self.assertFalse(self.cached.isfile("spam"))  # doesn't exist
-            scandir.assert_has_calls([mock.call('/', namespaces=None, page=None)])
+            scandir.assert_has_calls([mock.call("/", namespaces=None, page=None)])
         with mock.patch.object(self.fs, "scandir", wraps=self.fs.scandir) as scandir:
             self.assertTrue(self.cached.isfile("egg"))
             self.assertFalse(self.cached.isfile("foo"))
@@ -211,7 +211,7 @@ def test_getinfo(self):
             self.assertEqual(self.cached.getinfo("foo"), self.fs.getinfo("foo"))
             self.assertEqual(self.cached.getinfo("/"), self.fs.getinfo("/"))
             self.assertNotFound(self.cached.getinfo, "spam")
-            scandir.assert_has_calls([mock.call('/', namespaces=None, page=None)])
+            scandir.assert_has_calls([mock.call("/", namespaces=None, page=None)])
         with mock.patch.object(self.fs, "scandir", wraps=self.fs.scandir) as scandir:
             self.assertEqual(self.cached.getinfo("foo"), self.fs.getinfo("foo"))
             self.assertEqual(self.cached.getinfo("/"), self.fs.getinfo("/"))