Add tests checking WrapReadOnly works with fs.copy, fs.mirror and fs.move

Author: althonos

fs/wrap.py

@@ -100,10 +100,10 @@ class WrapCachedDir(WrapFS[_F], typing.Generic[_F]):
     #
     # A possible solution would be to replaced the cached with a
     #     Dict[Text, Dict[Text, Dict[Text, Info]]]
-    #           ^           ^         ^     ^~~ the actual info object
-    #           |           |         |~~ the path of the directory entry
-    #           |           |~~ the namespace of the info
-    #           |~~ the cached directory entry
+    #           ^           ^         ^     ^-- the actual info object
+    #           |           |         \-- the path of the directory entry
+    #           |           \-- the namespace of the info
+    #           \-- the cached directory entry
     #
     # Furthermore, `listdir` and `filterdir` calls should be cached as well,
     # since they can be written as wrappers of `scandir`.

tests/test_wrap.py

@@ -2,21 +2,24 @@
 
 import operator
 import unittest
+
 try:
     from unittest import mock
 except ImportError:
     import mock
 
 import six
 
+import fs.copy
+import fs.mirror
+import fs.move
 import fs.wrap
 import fs.errors
 from fs import open_fs
 from fs.info import Info
 
 
 class TestWrapReadOnly(unittest.TestCase):
-
     def setUp(self):
         self.fs = open_fs("mem://")
         self.ro = fs.wrap.read_only(self.fs)
@@ -95,8 +98,65 @@ def test_open_r(self):
             self.assertEqual(read_file.read(), b"read me")
 
 
-class TestWrapCachedDir(unittest.TestCase):
+class TestWrapReadOnlySyspath(unittest.TestCase):
+    # If the wrapped fs has a syspath, there is a chance that somewhere
+    # in fs.copy or fs.mirror we try to use it to our advantage, but
+    # we want to make sure these implementations don't circumvent the
+    # wrapper.
+
+    def setUp(self):
+        self.fs = open_fs("temp://")
+        self.ro = fs.wrap.read_only(self.fs)
+        self.src = open_fs("temp://")
+        self.src.touch("foo")
+        self.src.makedir("bar")
+
+    def tearDown(self):
+        self.fs.close()
+        self.src.close()
+
+    def assertReadOnly(self, func, *args, **kwargs):
+        self.assertRaises(fs.errors.ResourceReadOnly, func, *args, **kwargs)
 
+    def test_copy_fs(self):
+        self.assertReadOnly(fs.copy.copy_fs, self.src, self.ro)
+
+    def test_copy_fs_if_newer(self):
+        self.assertReadOnly(fs.copy.copy_fs_if_newer, self.src, self.ro)
+
+    def test_copy_file(self):
+        self.assertReadOnly(fs.copy.copy_file, self.src, "foo", self.ro, "foo")
+
+    def test_copy_file_if_newer(self):
+        self.assertReadOnly(fs.copy.copy_file_if_newer, self.src, "foo", self.ro, "foo")
+
+    def test_copy_structure(self):
+        self.assertReadOnly(fs.copy.copy_structure, self.src, self.ro)
+
+    def test_mirror(self):
+        self.assertReadOnly(fs.mirror.mirror, self.src, self.ro)
+        fs.mirror.mirror(self.src, self.fs)
+        self.fs.touch("baz")
+        self.assertReadOnly(fs.mirror.mirror, self.src, self.ro)
+
+    def test_move_fs(self):
+        self.assertReadOnly(fs.move.move_fs, self.src, self.ro)
+        self.src.removetree("/")
+        self.fs.touch("foo")
+        self.assertReadOnly(fs.move.move_fs, self.ro, self.src)
+
+    def test_move_file(self):
+        self.assertReadOnly(fs.move.move_file, self.src, "foo", self.ro, "foo")
+        self.fs.touch("baz")
+        self.assertReadOnly(fs.move.move_file, self.ro, "baz", self.src, "foo")
+
+    def test_move_dir(self):
+        self.assertReadOnly(fs.move.move_file, self.src, "bar", self.ro, "bar")
+        self.fs.makedir("baz")
+        self.assertReadOnly(fs.move.move_dir, self.ro, "baz", self.src, "baz")
+
+
+class TestWrapCachedDir(unittest.TestCase):
     def setUp(self):
         self.fs = open_fs("mem://")
         self.fs.makedirs("foo/bar/baz")
@@ -126,7 +186,7 @@ def test_isdir(self):
         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"))  # is file
-            self.assertFalse(self.cached.isdir("spam")) # doesn't exist
+            self.assertFalse(self.cached.isdir("spam"))  # doesn't exist
             scandir.assert_called()
         with mock.patch.object(self.fs, "scandir", wraps=self.fs.scandir) as scandir:
             self.assertTrue(self.cached.isdir("foo"))
@@ -138,7 +198,7 @@ def test_isfile(self):
         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"))  # is dir
-            self.assertFalse(self.cached.isfile("spam")) # doesn't exist
+            self.assertFalse(self.cached.isfile("spam"))  # doesn't exist
             scandir.assert_called()
         with mock.patch.object(self.fs, "scandir", wraps=self.fs.scandir) as scandir:
             self.assertTrue(self.cached.isfile("egg"))