Improve tests of fs.wrap wrappers to make sure caching and read-only work

Author: althonos

fs/wrap.py

@@ -92,6 +92,22 @@ class WrapCachedDir(WrapFS[_F], typing.Generic[_F]):
 
     """
 
+    # FIXME (@althonos): The caching data structure can very likely be
+    # improved. With the current implementation, if `scandir` result was
+    # cached for `namespaces=["details", "access"]`, calling `scandir`
+    # again only with `names=["details"]` will miss the cache, even though
+    # we are already storing the totality of the required metadata.
+    #
+    # 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
+    #
+    # Furthermore, `listdir` and `filterdir` calls should be cached as well,
+    # since they can be written as wrappers of `scandir`.
+
     wrap_name = "cached-dir"
 
     def __init__(self, wrap_fs):  # noqa: D107

tests/test_wrap.py

@@ -1,97 +1,159 @@
 from __future__ import unicode_literals
 
+import operator
 import unittest
+try:
+    from unittest import mock
+except ImportError:
+    import mock
 
-from fs import errors
+import six
+
+import fs.wrap
+import fs.errors
 from fs import open_fs
-from fs import wrap
+from fs.info import Info
 
 
-class TestWrap(unittest.TestCase):
-    def test_readonly(self):
-        mem_fs = open_fs("mem://")
-        fs = wrap.read_only(mem_fs)
+class TestWrapReadOnly(unittest.TestCase):
 
-        with self.assertRaises(errors.ResourceReadOnly):
-            fs.open("foo", "w")
+    def setUp(self):
+        self.fs = open_fs("mem://")
+        self.ro = fs.wrap.read_only(self.fs)
 
-        with self.assertRaises(errors.ResourceReadOnly):
-            fs.appendtext("foo", "bar")
+    def tearDown(self):
+        self.fs.close()
 
-        with self.assertRaises(errors.ResourceReadOnly):
-            fs.appendbytes("foo", b"bar")
+    def assertReadOnly(self, func, *args, **kwargs):
+        self.assertRaises(fs.errors.ResourceReadOnly, func, *args, **kwargs)
 
-        with self.assertRaises(errors.ResourceReadOnly):
-            fs.makedir("foo")
+    def test_open_w(self):
+        self.assertReadOnly(self.ro.open, "foo", "w")
 
-        with self.assertRaises(errors.ResourceReadOnly):
-            fs.move("foo", "bar")
+    def test_appendtext(self):
+        self.assertReadOnly(self.ro.appendtext, "foo", "bar")
 
-        with self.assertRaises(errors.ResourceReadOnly):
-            fs.openbin("foo", "w")
+    def test_appendbytes(self):
+        self.assertReadOnly(self.ro.appendbytes, "foo", b"bar")
 
-        with self.assertRaises(errors.ResourceReadOnly):
-            fs.remove("foo")
+    def test_makedir(self):
+        self.assertReadOnly(self.ro.makedir, "foo")
 
-        with self.assertRaises(errors.ResourceReadOnly):
-            fs.removedir("foo")
+    def test_move(self):
+        self.assertReadOnly(self.ro.move, "foo", "bar")
 
-        with self.assertRaises(errors.ResourceReadOnly):
-            fs.setinfo("foo", {})
+    def test_openbin_w(self):
+        self.assertReadOnly(self.ro.openbin, "foo", "w")
 
-        with self.assertRaises(errors.ResourceReadOnly):
-            fs.settimes("foo", {})
+    def test_remove(self):
+        self.assertReadOnly(self.ro.remove, "foo")
 
-        with self.assertRaises(errors.ResourceReadOnly):
-            fs.copy("foo", "bar")
+    def test_removedir(self):
+        self.assertReadOnly(self.ro.removedir, "foo")
 
-        with self.assertRaises(errors.ResourceReadOnly):
-            fs.create("foo")
+    def test_removetree(self):
+        self.assertReadOnly(self.ro.removetree, "foo")
 
-        with self.assertRaises(errors.ResourceReadOnly):
-            fs.writetext("foo", "bar")
+    def test_setinfo(self):
+        self.assertReadOnly(self.ro.setinfo, "foo", {})
 
-        with self.assertRaises(errors.ResourceReadOnly):
-            fs.writebytes("foo", b"bar")
+    def test_settimes(self):
+        self.assertReadOnly(self.ro.settimes, "foo", {})
 
-        with self.assertRaises(errors.ResourceReadOnly):
-            fs.makedirs("foo/bar")
+    def test_copy(self):
+        self.assertReadOnly(self.ro.copy, "foo", "bar")
 
-        with self.assertRaises(errors.ResourceReadOnly):
-            fs.touch("foo")
+    def test_create(self):
+        self.assertReadOnly(self.ro.create, "foo")
 
-        with self.assertRaises(errors.ResourceReadOnly):
-            fs.upload("foo", None)
+    def test_writetext(self):
+        self.assertReadOnly(self.ro.writetext, "foo", "bar")
 
-        with self.assertRaises(errors.ResourceReadOnly):
-            fs.writefile("foo", None)
+    def test_writebytes(self):
+        self.assertReadOnly(self.ro.writebytes, "foo", b"bar")
 
-        self.assertTrue(mem_fs.isempty("/"))
-        mem_fs.writebytes("file", b"read me")
-        with fs.openbin("file") as read_file:
-            self.assertEqual(read_file.read(), b"read me")
+    def test_makedirs(self):
+        self.assertReadOnly(self.ro.makedirs, "foo/bar")
 
-        with fs.open("file", "rb") as read_file:
-            self.assertEqual(read_file.read(), b"read me")
+    def test_touch(self):
+        self.assertReadOnly(self.ro.touch, "foo")
 
-    def test_cachedir(self):
-        mem_fs = open_fs("mem://")
-        mem_fs.makedirs("foo/bar/baz")
-        mem_fs.touch("egg")
+    def test_upload(self):
+        self.assertReadOnly(self.ro.upload, "foo", six.BytesIO())
 
-        fs = wrap.cache_directory(mem_fs)
-        self.assertEqual(sorted(fs.listdir("/")), ["egg", "foo"])
-        self.assertEqual(sorted(fs.listdir("/")), ["egg", "foo"])
-        self.assertTrue(fs.isdir("foo"))
-        self.assertTrue(fs.isdir("foo"))
-        self.assertTrue(fs.isfile("egg"))
-        self.assertTrue(fs.isfile("egg"))
+    def test_writefile(self):
+        self.assertReadOnly(self.ro.writefile, "foo", six.StringIO())
 
-        self.assertEqual(fs.getinfo("foo"), mem_fs.getinfo("foo"))
-        self.assertEqual(fs.getinfo("foo"), mem_fs.getinfo("foo"))
+    def test_openbin_r(self):
+        self.fs.writebytes("file", b"read me")
+        with self.ro.openbin("file") as read_file:
+            self.assertEqual(read_file.read(), b"read me")
+
+    def test_open_r(self):
+        self.fs.writebytes("file", b"read me")
+        with self.ro.open("file", "rb") as read_file:
+            self.assertEqual(read_file.read(), b"read me")
 
-        self.assertEqual(fs.getinfo("/"), mem_fs.getinfo("/"))
-        self.assertEqual(fs.getinfo("/"), mem_fs.getinfo("/"))
 
-        with self.assertRaises(errors.ResourceNotFound):
-            fs.getinfo("/foofoo")
+class TestWrapCachedDir(unittest.TestCase):
+
+    def setUp(self):
+        self.fs = open_fs("mem://")
+        self.fs.makedirs("foo/bar/baz")
+        self.fs.touch("egg")
+        self.cached = fs.wrap.cache_directory(self.fs)
+
+    def tearDown(self):
+        self.fs.close()
+
+    def assertNotFound(self, func, *args, **kwargs):
+        self.assertRaises(fs.errors.ResourceNotFound, func, *args, **kwargs)
+
+    def test_scandir(self):
+        key = operator.attrgetter("name")
+        expected = [
+            Info({"basic": {"name": "egg", "is_dir": False}}),
+            Info({"basic": {"name": "foo", "is_dir": True}}),
+        ]
+        with mock.patch.object(self.fs, "scandir", wraps=self.fs.scandir) as scandir:
+            self.assertEqual(sorted(self.cached.scandir("/"), key=key), expected)
+            scandir.assert_called()
+        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()
+
+    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
+            scandir.assert_called()
+        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"))
+            self.assertFalse(self.cached.isdir("spam"))
+            scandir.assert_not_called()
+
+    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
+            scandir.assert_called()
+        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"))
+            self.assertFalse(self.cached.isfile("spam"))
+            scandir.assert_not_called()
+
+    def test_getinfo(self):
+        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("/"))
+            self.assertNotFound(self.cached.getinfo, "spam")
+            scandir.assert_called()
+        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("/"))
+            self.assertNotFound(self.cached.getinfo, "spam")
+            scandir.assert_not_called()