Added support for MFMT command to FTPFS.setinfo to set the last modified time of a file.

Author: atollk

fs/copy.py

@@ -461,7 +461,7 @@ def copy_mtime(
         dst_path (str): Path to a directory on the destination filesystem.
 
     """
-    namespaces = ("details", "metadata_changed", "modified")
+    namespaces = ("details",)
     with manage_fs(src_fs, writeable=False) as _src_fs:
         with manage_fs(dst_fs, create=True) as _dst_fs:
             src_meta = _src_fs.getinfo(src_path, namespaces)

fs/ftpfs.py

@@ -6,6 +6,7 @@
 
 import array
 import calendar
+import datetime
 import io
 import itertools
 import socket
@@ -19,8 +20,7 @@
     from ftplib import FTP_TLS
 except ImportError as err:
     FTP_TLS = err  # type: ignore
-from ftplib import error_perm
-from ftplib import error_temp
+from ftplib import error_perm, error_temp
 from typing import cast
 
 from six import PY2
@@ -836,8 +836,27 @@ def writebytes(self, path, contents):
 
     def setinfo(self, path, info):
         # type: (Text, RawInfo) -> None
-        if not self.exists(path):
-            raise errors.ResourceNotFound(path)
+        current_info = self.getinfo(path)
+        if current_info.is_file and "MFMT" in self.features:
+            mtime = 0.0
+            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
 
     def readbytes(self, path):
         # type: (Text) -> bytes

tests/test_ftpfs.py

@@ -11,6 +11,7 @@
 import time
 import unittest
 import uuid
+import datetime
 
 try:
     from unittest import mock
@@ -144,7 +145,6 @@ def test_manager_with_host(self):
 @mark.slow
 @unittest.skipIf(platform.python_implementation() == "PyPy", "ftp unreliable with PyPy")
 class TestFTPFS(FSTestCases, unittest.TestCase):
-
     user = "user"
     pasw = "1234"
 
@@ -162,7 +162,7 @@ def setUpClass(cls):
         cls.server.shutdown_after = -1
         cls.server.handler.authorizer = DummyAuthorizer()
         cls.server.handler.authorizer.add_user(
-            cls.user, cls.pasw, cls._temp_path, perm="elradfmw"
+            cls.user, cls.pasw, cls._temp_path, perm="elradfmwT"
         )
         cls.server.handler.authorizer.add_anonymous(cls._temp_path)
         cls.server.start()
@@ -223,6 +223,27 @@ 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)
+        )
+
     def test_host(self):
         self.assertEqual(self.fs.host, self.server.host)
 
@@ -301,7 +322,6 @@ def test_features(self):
 @mark.slow
 @unittest.skipIf(platform.python_implementation() == "PyPy", "ftp unreliable with PyPy")
 class TestAnonFTPFS(FSTestCases, unittest.TestCase):
-
     user = "anonymous"
     pasw = ""