Optimization in FTPFS.setinfo.

Before, `setinfo` would always start by sending a MLST or LIST command to find whether the resources exists. This is actually not necessary if MFMT is sent anyway, so the command is now skipped in that case.

Author: atollk

fs/ftpfs.py

@@ -836,27 +836,32 @@ def writebytes(self, path, contents):
 
     def setinfo(self, path, info):
         # type: (Text, RawInfo) -> None
-        current_info = self.getinfo(path)
-        if current_info.is_file and "MFMT" in self.features:
-            mtime = 0.0
+        use_mfmt = False
+        if "MFMT" in self.features:
+            info_details = None
             if "modified" in info:
-                mtime = float(cast(float, info["modified"]["modified"]))
-            if "details" in info:
-                mtime = float(cast(float, info["details"]["modified"]))
-            if mtime:
-                with ftp_errors(self, path):
-                    cmd = (
-                        "MFMT "
-                        + datetime.datetime.fromtimestamp(mtime).strftime(
-                            "%Y%m%d%H%M%S"
-                        )
-                        + " "
-                        + _encode(path, self.ftp.encoding)
-                    )
-                    try:
-                        self.ftp.sendcmd(cmd)
-                    except error_perm:
-                        pass
+                info_details = info["modified"]
+            elif "details" in info:
+                info_details = info["details"]
+            if info_details and "modified" in info_details:
+                use_mfmt = True
+                mtime = cast(float, info_details["modified"])
+
+        if use_mfmt:
+            with ftp_errors(self, path):
+                cmd = (
+                    "MFMT "
+                    + datetime.datetime.utcfromtimestamp(mtime).strftime("%Y%m%d%H%M%S")
+                    + " "
+                    + _encode(path, self.ftp.encoding)
+                )
+                try:
+                    self.ftp.sendcmd(cmd)
+                except error_perm:
+                    pass
+        else:
+            if not self.exists(path):
+                raise errors.ResourceNotFound(path)
 
     def readbytes(self, path):
         # type: (Text) -> bytes

tests/test_ftpfs.py

@@ -3,6 +3,7 @@
 from __future__ import print_function
 from __future__ import unicode_literals
 
+import calendar
 import os
 import platform
 import shutil
@@ -226,23 +227,19 @@ def test_geturl(self):
     def test_setinfo(self):
         # TODO: temporary test, since FSTestCases.test_setinfo is broken.
         self.fs.create("bar")
-        time1 = time.mktime(datetime.datetime.now().timetuple())
-        time2 = time1 + 60
-        self.fs.setinfo("bar", {"details": {"modified": time1}})
-        mtime1 = self.fs.getinfo("bar", ("details",)).modified
-        self.fs.setinfo("bar", {"details": {"modified": time2}})
-        mtime2 = self.fs.getinfo("bar", ("details",)).modified
-        replacement = {}
-        if mtime1.microsecond == 0 or mtime2.microsecond == 0:
-            mtime1 = mtime1.replace(microsecond=0)
-            mtime2 = mtime2.replace(microsecond=0)
-        if mtime1.second == 0 or mtime2.second == 0:
-            mtime1 = mtime1.replace(second=0)
-            mtime2 = mtime2.replace(second=0)
-        mtime2_modified = mtime2.replace(minute=mtime2.minute - 1)
-        self.assertEqual(
-            mtime1.replace(**replacement), mtime2_modified.replace(**replacement)
-        )
+        original_modified = self.fs.getinfo("bar", ("details",)).modified
+        new_modified = original_modified - datetime.timedelta(hours=1)
+        new_modified_stamp = calendar.timegm(new_modified.timetuple())
+        self.fs.setinfo("bar", {"details": {"modified": new_modified_stamp}})
+        new_modified_get = self.fs.getinfo("bar", ("details",)).modified
+        if original_modified.microsecond == 0 or new_modified_get.microsecond == 0:
+            original_modified = original_modified.replace(microsecond=0)
+            new_modified_get = new_modified_get.replace(microsecond=0)
+        if original_modified.second == 0 or new_modified_get.second == 0:
+            original_modified = original_modified.replace(second=0)
+            new_modified_get = new_modified_get.replace(second=0)
+        new_modified_get = new_modified_get + datetime.timedelta(hours=1)
+        self.assertEqual(original_modified, new_modified_get)
 
     def test_host(self):
         self.assertEqual(self.fs.host, self.server.host)