Added unit tests for the preserve_time parameter.

At this point in time, all the new tests are failing right now; as they should, because the functionality of `preserve_time` has not been implemented yet.

Author: atollk

setup.cfg

@@ -133,6 +133,7 @@ sitepackages = false
 skip_missing_interpreters = true
 requires =
   setuptools >=38.3.0
+  parameterized ~=0.8
 
 [testenv]
 commands = pytest --cov={toxinidir}/fs {posargs} {toxinidir}/tests

tests/requirements.txt

@@ -14,3 +14,5 @@ pysendfile ~=2.0 ; python_version <= "3.3"
 # mock v4+ doesn't support Python 2.7 anymore
 mock ~=3.0 ; python_version < "3.3"
 
+# parametrized to prevent code duplication in tests.
+parameterized ~=0.8

tests/test_copy.py

@@ -8,44 +8,77 @@
 import shutil
 import calendar
 
+from parameterized import parameterized
+
 import fs.copy
 from fs import open_fs
 
 
 class TestCopy(unittest.TestCase):
-    def test_copy_fs(self):
-        for workers in (0, 1, 2, 4):
-            src_fs = open_fs("mem://")
-            src_fs.makedirs("foo/bar")
-            src_fs.makedirs("foo/empty")
-            src_fs.touch("test.txt")
-            src_fs.touch("foo/bar/baz.txt")
+    @parameterized.expand([(0,), (1,), (2,), (4,)])
+    def test_copy_fs(self, workers):
+        namespaces = ("details", "accessed", "metadata_changed", "modified")
 
-            dst_fs = open_fs("mem://")
-            fs.copy.copy_fs(src_fs, dst_fs, workers=workers)
+        src_fs = open_fs("mem://")
+        src_fs.makedirs("foo/bar")
+        src_fs.makedirs("foo/empty")
+        src_fs.touch("test.txt")
+        src_fs.touch("foo/bar/baz.txt")
+        src_file1_info = src_fs.getinfo("test.txt", namespaces)
+        src_file2_info = src_fs.getinfo("foo/bar/baz.txt", namespaces)
 
-            self.assertTrue(dst_fs.isdir("foo/empty"))
-            self.assertTrue(dst_fs.isdir("foo/bar"))
-            self.assertTrue(dst_fs.isfile("test.txt"))
+        dst_fs = open_fs("mem://")
+        fs.copy.copy_fs(src_fs, dst_fs, workers=workers, preserve_time=True)
+
+        self.assertTrue(dst_fs.isdir("foo/empty"))
+        self.assertTrue(dst_fs.isdir("foo/bar"))
+        self.assertTrue(dst_fs.isfile("test.txt"))
+
+        dst_file1_info = dst_fs.getinfo("test.txt", namespaces)
+        dst_file2_info = dst_fs.getinfo("foo/bar/baz.txt", namespaces)
+        self.assertEqual(dst_file1_info.modified, src_file1_info.modified)
+        self.assertEqual(dst_file1_info.accessed, src_file1_info.accessed)
+        self.assertEqual(
+            dst_file1_info.metadata_changed, src_file1_info.metadata_changed
+        )
+        self.assertEqual(dst_file2_info.modified, src_file2_info.modified)
+        self.assertEqual(dst_file2_info.accessed, src_file2_info.accessed)
+        self.assertEqual(
+            dst_file2_info.metadata_changed, src_file2_info.metadata_changed
+        )
 
     def test_copy_value_error(self):
         src_fs = open_fs("mem://")
         dst_fs = open_fs("mem://")
         with self.assertRaises(ValueError):
             fs.copy.copy_fs(src_fs, dst_fs, workers=-1)
 
-    def test_copy_dir(self):
+    @parameterized.expand([(0,), (1,), (2,), (4,)])
+    def test_copy_dir(self, workers):
+        namespaces = ("details", "accessed", "metadata_changed", "modified")
+
         src_fs = open_fs("mem://")
         src_fs.makedirs("foo/bar")
         src_fs.makedirs("foo/empty")
         src_fs.touch("test.txt")
         src_fs.touch("foo/bar/baz.txt")
-        for workers in (0, 1, 2, 4):
-            with open_fs("mem://") as dst_fs:
-                fs.copy.copy_dir(src_fs, "/foo", dst_fs, "/", workers=workers)
-                self.assertTrue(dst_fs.isdir("bar"))
-                self.assertTrue(dst_fs.isdir("empty"))
-                self.assertTrue(dst_fs.isfile("bar/baz.txt"))
+        src_file1_info = src_fs.getinfo("test.txt", namespaces)
+        src_file2_info = src_fs.getinfo("foo/bar/baz.txt", namespaces)
+
+        with open_fs("mem://") as dst_fs:
+            fs.copy.copy_dir(
+                src_fs, "/foo", dst_fs, "/", workers=workers, preserve_time=True
+            )
+            self.assertTrue(dst_fs.isdir("bar"))
+            self.assertTrue(dst_fs.isdir("empty"))
+            self.assertTrue(dst_fs.isfile("bar/baz.txt"))
+
+            dst_file2_info = dst_fs.getinfo("bar/baz.txt", namespaces)
+            self.assertEqual(dst_file2_info.modified, src_file2_info.modified)
+            self.assertEqual(dst_file2_info.accessed, src_file2_info.accessed)
+            self.assertEqual(
+                dst_file2_info.metadata_changed, src_file2_info.metadata_changed
+            )
 
     def test_copy_large(self):
         data1 = b"foo" * 512 * 1024

tests/test_memoryfs.py

@@ -66,3 +66,19 @@ def test_close_mem_free(self):
             "Memory usage increased after closing the file system; diff is %0.2f KiB."
             % (diff_close.size_diff / 1024.0),
         )
+
+    def test_copy_preserve_time(self):
+        self.fs.makedir("foo")
+        self.fs.makedir("bar")
+        self.fs.touch("foo/file.txt")
+
+        namespaces = ("details", "accessed", "metadata_changed", "modified")
+        src_info = self.fs.getinfo("foo/file.txt", namespaces)
+
+        self.fs.copy("foo/file.txt", "bar/file.txt", preserve_time=True)
+        self.assertTrue(self.fs.exists("bar/file.txt"))
+
+        dst_info = self.fs.getinfo("bar/file.txt", namespaces)
+        self.assertEqual(dst_info.modified, src_info.modified)
+        self.assertEqual(dst_info.accessed, src_info.accessed)
+        self.assertEqual(dst_info.metadata_changed, src_info.metadata_changed)

tests/test_mirror.py

@@ -2,23 +2,36 @@
 
 import unittest
 
+from parameterized import parameterized_class
+
 from fs.mirror import mirror
 from fs import open_fs
 
 
+@parameterized_class(("WORKERS",), [(0,), (1,), (2,), (4,)])
 class TestMirror(unittest.TestCase):
-    WORKERS = 0  # Single threaded
-
     def _contents(self, fs):
         """Extract an FS in to a simple data structure."""
+        namespaces = ("details", "accessed", "metadata_changed", "modified")
         contents = []
         for path, dirs, files in fs.walk():
             for info in dirs:
                 _path = info.make_path(path)
                 contents.append((_path, "dir", b""))
             for info in files:
                 _path = info.make_path(path)
-                contents.append((_path, "file", fs.readbytes(_path)))
+                _bytes = fs.readbytes(_path)
+                _info = fs.getinfo(_path, namespaces)
+                contents.append(
+                    (
+                        _path,
+                        "file",
+                        _bytes,
+                        _info.modified,
+                        _info.accessed,
+                        _info.metadata_changed,
+                    )
+                )
         return sorted(contents)
 
     def assert_compare_fs(self, fs1, fs2):
@@ -28,36 +41,36 @@ def assert_compare_fs(self, fs1, fs2):
     def test_empty_mirror(self):
         m1 = open_fs("mem://")
         m2 = open_fs("mem://")
-        mirror(m1, m2, workers=self.WORKERS)
+        mirror(m1, m2, workers=self.WORKERS, preserve_time=True)
         self.assert_compare_fs(m1, m2)
 
     def test_mirror_one_file(self):
         m1 = open_fs("mem://")
         m1.writetext("foo", "hello")
         m2 = open_fs("mem://")
-        mirror(m1, m2, workers=self.WORKERS)
+        mirror(m1, m2, workers=self.WORKERS, preserve_time=True)
         self.assert_compare_fs(m1, m2)
 
     def test_mirror_one_file_one_dir(self):
         m1 = open_fs("mem://")
         m1.writetext("foo", "hello")
         m1.makedir("bar")
         m2 = open_fs("mem://")
-        mirror(m1, m2, workers=self.WORKERS)
+        mirror(m1, m2, workers=self.WORKERS, preserve_time=True)
         self.assert_compare_fs(m1, m2)
 
     def test_mirror_delete_replace(self):
         m1 = open_fs("mem://")
         m1.writetext("foo", "hello")
         m1.makedir("bar")
         m2 = open_fs("mem://")
-        mirror(m1, m2, workers=self.WORKERS)
+        mirror(m1, m2, workers=self.WORKERS, preserve_time=True)
         self.assert_compare_fs(m1, m2)
         m2.remove("foo")
-        mirror(m1, m2, workers=self.WORKERS)
+        mirror(m1, m2, workers=self.WORKERS, preserve_time=True)
         self.assert_compare_fs(m1, m2)
         m2.removedir("bar")
-        mirror(m1, m2, workers=self.WORKERS)
+        mirror(m1, m2, workers=self.WORKERS, preserve_time=True)
         self.assert_compare_fs(m1, m2)
 
     def test_mirror_extra_dir(self):
@@ -66,7 +79,7 @@ def test_mirror_extra_dir(self):
         m1.makedir("bar")
         m2 = open_fs("mem://")
         m2.makedir("baz")
-        mirror(m1, m2, workers=self.WORKERS)
+        mirror(m1, m2, workers=self.WORKERS, preserve_time=True)
         self.assert_compare_fs(m1, m2)
 
     def test_mirror_extra_file(self):
@@ -76,7 +89,7 @@ def test_mirror_extra_file(self):
         m2 = open_fs("mem://")
         m2.makedir("baz")
         m2.touch("egg")
-        mirror(m1, m2, workers=self.WORKERS)
+        mirror(m1, m2, workers=self.WORKERS, preserve_time=True)
         self.assert_compare_fs(m1, m2)
 
     def test_mirror_wrong_type(self):
@@ -86,28 +99,16 @@ def test_mirror_wrong_type(self):
         m2 = open_fs("mem://")
         m2.makedir("foo")
         m2.touch("bar")
-        mirror(m1, m2, workers=self.WORKERS)
+        mirror(m1, m2, workers=self.WORKERS, preserve_time=True)
         self.assert_compare_fs(m1, m2)
 
     def test_mirror_update(self):
         m1 = open_fs("mem://")
         m1.writetext("foo", "hello")
         m1.makedir("bar")
         m2 = open_fs("mem://")
-        mirror(m1, m2, workers=self.WORKERS)
+        mirror(m1, m2, workers=self.WORKERS, preserve_time=True)
         self.assert_compare_fs(m1, m2)
         m2.appendtext("foo", " world!")
-        mirror(m1, m2, workers=self.WORKERS)
+        mirror(m1, m2, workers=self.WORKERS, preserve_time=True)
         self.assert_compare_fs(m1, m2)
-
-
-class TestMirrorWorkers1(TestMirror):
-    WORKERS = 1
-
-
-class TestMirrorWorkers2(TestMirror):
-    WORKERS = 2
-
-
-class TestMirrorWorkers4(TestMirror):
-    WORKERS = 4

tests/test_move.py

@@ -2,33 +2,67 @@
 
 import unittest
 
+from parameterized import parameterized_class
+
 import fs.move
 from fs import open_fs
 
 
+@parameterized_class(("preserve_time",), [(True,), (False,)])
 class TestMove(unittest.TestCase):
     def test_move_fs(self):
+        namespaces = ("details", "accessed", "metadata_changed", "modified")
+
         src_fs = open_fs("mem://")
         src_fs.makedirs("foo/bar")
         src_fs.touch("test.txt")
         src_fs.touch("foo/bar/baz.txt")
+        src_file1_info = src_fs.getinfo("test.txt", namespaces)
+        src_file2_info = src_fs.getinfo("foo/bar/baz.txt", namespaces)
 
         dst_fs = open_fs("mem://")
         fs.move.move_fs(src_fs, dst_fs)
 
         self.assertTrue(dst_fs.isdir("foo/bar"))
         self.assertTrue(dst_fs.isfile("test.txt"))
+        self.assertTrue(dst_fs.isfile("foo/bar/baz.txt"))
         self.assertTrue(src_fs.isempty("/"))
 
-    def test_copy_dir(self):
+        if self.preserve_time:
+            dst_file1_info = dst_fs.getinfo("test.txt", namespaces)
+            dst_file2_info = dst_fs.getinfo("foo/bar/baz.txt", namespaces)
+            self.assertEqual(dst_file1_info.modified, src_file1_info.modified)
+            self.assertEqual(dst_file1_info.accessed, src_file1_info.accessed)
+            self.assertEqual(
+                dst_file1_info.metadata_changed, src_file1_info.metadata_changed
+            )
+            self.assertEqual(dst_file2_info.modified, src_file2_info.modified)
+            self.assertEqual(dst_file2_info.accessed, src_file2_info.accessed)
+            self.assertEqual(
+                dst_file2_info.metadata_changed, src_file2_info.metadata_changed
+            )
+
+    def test_move_dir(self):
+        namespaces = ("details", "accessed", "metadata_changed", "modified")
+
         src_fs = open_fs("mem://")
         src_fs.makedirs("foo/bar")
         src_fs.touch("test.txt")
         src_fs.touch("foo/bar/baz.txt")
+        src_file2_info = src_fs.getinfo("foo/bar/baz.txt", namespaces)
 
         dst_fs = open_fs("mem://")
         fs.move.move_dir(src_fs, "/foo", dst_fs, "/")
 
         self.assertTrue(dst_fs.isdir("bar"))
         self.assertTrue(dst_fs.isfile("bar/baz.txt"))
         self.assertFalse(src_fs.exists("foo"))
+        self.assertTrue(src_fs.isfile("test.txt"))
+
+        if self.preserve_time:
+            dst_file2_info = dst_fs.getinfo("bar/baz.txt", namespaces)
+            self.assertEqual(dst_file2_info.modified, src_file2_info.modified)
+            self.assertEqual(dst_file2_info.accessed, src_file2_info.accessed)
+            self.assertEqual(
+                dst_file2_info.metadata_changed, src_file2_info.metadata_changed
+            )

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_osfs.py

@@ -7,6 +7,7 @@
 import shutil
 import tempfile
 import sys
+import time
 import unittest
 
 from fs import osfs, open_fs
@@ -87,6 +88,25 @@ def test_expand_vars(self):
         self.assertIn("TYRIONLANISTER", fs1.getsyspath("/"))
         self.assertNotIn("TYRIONLANISTER", fs2.getsyspath("/"))
 
+    def test_copy_preserve_time(self):
+        self.fs.makedir("foo")
+        self.fs.makedir("bar")
+        now = time.time() - 10000
+        if not self.fs.create("foo/file.txt"):
+            raw_info = {"details": {"accessed": now, "modified": now}}
+            self.fs.setinfo("foo/file.txt", raw_info)
+
+        namespaces = ("details", "accessed", "metadata_changed", "modified")
+        src_info = self.fs.getinfo("foo/file.txt", namespaces)
+
+        self.fs.copy("foo/file.txt", "bar/file.txt", preserve_time=True)
+        self.assertTrue(self.fs.exists("bar/file.txt"))
+
+        dst_info = self.fs.getinfo("bar/file.txt", namespaces)
+        self.assertEqual(dst_info.modified, src_info.modified)
+        self.assertEqual(dst_info.accessed, src_info.accessed)
+        self.assertEqual(dst_info.metadata_changed, src_info.metadata_changed)
+
     @unittest.skipUnless(osfs.sendfile, "sendfile not supported")
     @unittest.skipIf(
         sys.version_info >= (3, 8),