Route MultiFS removal through the writable filesystem

CHANGELOG.md

@@ -15,6 +15,8 @@ and this project adheres to [Semantic Versioning](http://semver.org/).
   Closes [#459](https://github.com/PyFilesystem/pyfilesystem2/issues/459).
 
 ### Fixed
+- Restricted MultiFS file and directory removal to the configured write filesystem.
+  Closes [#572](https://github.com/PyFilesystem/pyfilesystem2/issues/572).
 - Elaborated documentation of `filter_dirs` and `exclude_dirs` in `fs.walk.Walker`.
   Closes [#371](https://github.com/PyFilesystem/pyfilesystem2/issues/371).
 - Fixes a backward incompatibility where `fs.move.move_file` raises `DestinationExists`

CONTRIBUTORS.md

@@ -20,6 +20,7 @@ Many thanks to the following developers for contributing to this project:
 - [George Macon](https://github.com/gmacon)
 - [Giampaolo Cimino](https://github.com/gpcimino)
 - [@Hoboneer](https://github.com/Hoboneer)
+- [Hugohong258](https://github.com/Hugohong258)
 - [Jen Hagg](https://github.com/jenhagg)
 - [Joseph Atkins-Turkish](https://github.com/Spacerat)
 - [Joshua Tauberer](https://github.com/JoshData)

fs/multifs.py

@@ -249,13 +249,13 @@ def openbin(self, path, mode="r", buffering=-1, **options):
     def remove(self, path):
         # type: (Text) -> None
         self.check()
-        fs = self._delegate_required(path)
+        fs = self._writable_required(path)
         return fs.remove(path)
 
     def removedir(self, path):
         # type: (Text) -> None
         self.check()
-        fs = self._delegate_required(path)
+        fs = self._writable_required(path)
         return fs.removedir(path)
 
     def scandir(

tests/test_multifs.py

@@ -119,6 +119,58 @@ def test_priority(self):
         multi_fs.add_fs("m3", m3, priority=10)
         self.assertEqual(multi_fs.readbytes("name"), b"m1")
 
+    def test_remove_requires_writable_filesystem(self):
+        with MultiFS() as fs:
+            readonly = MemoryFS()
+            readonly.writebytes("file", b"keep")
+            fs.add_fs("readonly", readonly)
+            with self.assertRaises(errors.ResourceReadOnly):
+                fs.remove("file")
+            self.assertEqual(readonly.readbytes("file"), b"keep")
+
+    def test_removedir_requires_writable_filesystem(self):
+        with MultiFS() as fs:
+            readonly = MemoryFS()
+            readonly.makedir("directory")
+            fs.add_fs("readonly", readonly)
+            with self.assertRaises(errors.ResourceReadOnly):
+                fs.removedir("directory")
+            self.assertTrue(readonly.isdir("directory"))
+
+    def test_remove_uses_writable_filesystem(self):
+        readonly = MemoryFS()
+        readonly.writebytes("file", b"keep")
+        self.mem_fs.writebytes("file", b"remove")
+        self.fs.add_fs("readonly", readonly, priority=1)
+        self.fs.remove("file")
+        self.assertFalse(self.mem_fs.exists("file"))
+        self.assertEqual(readonly.readbytes("file"), b"keep")
+
+    def test_removedir_uses_writable_filesystem(self):
+        readonly = MemoryFS()
+        readonly.makedir("directory")
+        self.mem_fs.makedir("directory")
+        self.fs.add_fs("readonly", readonly, priority=1)
+        self.fs.removedir("directory")
+        self.assertFalse(self.mem_fs.exists("directory"))
+        self.assertTrue(readonly.isdir("directory"))
+
+    def test_remove_does_not_fall_back_to_readonly_filesystem(self):
+        readonly = MemoryFS()
+        readonly.writebytes("file", b"keep")
+        self.fs.add_fs("readonly", readonly)
+        with self.assertRaises(errors.ResourceNotFound):
+            self.fs.remove("file")
+        self.assertEqual(readonly.readbytes("file"), b"keep")
+
+    def test_removedir_does_not_fall_back_to_readonly_filesystem(self):
+        readonly = MemoryFS()
+        readonly.makedir("directory")
+        self.fs.add_fs("readonly", readonly)
+        with self.assertRaises(errors.ResourceNotFound):
+            self.fs.removedir("directory")
+        self.assertTrue(readonly.isdir("directory"))
+
     def test_no_writable(self):
         fs = MultiFS()
         with self.assertRaises(errors.ResourceReadOnly):