+ self.assertRegexpMatches(collection2.manifest_text(), r"\./testdir2 86fb269d190d2c85f6e0468ceca42a20\+12\+A\S+ 0:12:file1\.txt$")
+
+ collection1.stop_threads()
+ collection2.stop_threads()
+
+def fuseProjectMkdirTestHelper1(mounttmp):
+ class Test(unittest.TestCase):
+ def runTest(self):
+ os.mkdir(os.path.join(mounttmp, "testcollection"))
+ with self.assertRaises(OSError):
+ os.mkdir(os.path.join(mounttmp, "testcollection"))
+ Test().runTest()
+
+def fuseProjectMkdirTestHelper2(mounttmp):
+ class Test(unittest.TestCase):
+ def runTest(self):
+ with open(os.path.join(mounttmp, "testcollection", "file1.txt"), "w") as f:
+ f.write("Hello world!")
+ with self.assertRaises(OSError):
+ os.rmdir(os.path.join(mounttmp, "testcollection"))
+ os.remove(os.path.join(mounttmp, "testcollection", "file1.txt"))
+ with self.assertRaises(OSError):
+ os.remove(os.path.join(mounttmp, "testcollection"))
+ os.rmdir(os.path.join(mounttmp, "testcollection"))
+ Test().runTest()
+
+class FuseProjectMkdirRmdirTest(MountTestBase):
+ def runTest(self):
+ self.make_mount(fuse.ProjectDirectory,
+ project_object=self.api.users().current().execute())
+
+ d1 = llfuse.listdir(self.mounttmp)
+ self.assertNotIn('testcollection', d1)
+
+ self.pool.apply(fuseProjectMkdirTestHelper1, (self.mounttmp,))
+
+ d1 = llfuse.listdir(self.mounttmp)
+ self.assertIn('testcollection', d1)
+
+ self.pool.apply(fuseProjectMkdirTestHelper2, (self.mounttmp,))
+
+ d1 = llfuse.listdir(self.mounttmp)
+ self.assertNotIn('testcollection', d1)
+
+
+def fuseProjectMvTestHelper1(mounttmp):
+ class Test(unittest.TestCase):
+ def runTest(self):
+ d1 = llfuse.listdir(mounttmp)
+ self.assertNotIn('testcollection', d1)
+
+ os.mkdir(os.path.join(mounttmp, "testcollection"))
+
+ d1 = llfuse.listdir(mounttmp)
+ self.assertIn('testcollection', d1)
+
+ with self.assertRaises(OSError):
+ os.rename(os.path.join(mounttmp, "testcollection"), os.path.join(mounttmp, 'Unrestricted public data'))
+
+ os.rename(os.path.join(mounttmp, "testcollection"), os.path.join(mounttmp, 'Unrestricted public data', 'testcollection'))
+
+ d1 = llfuse.listdir(mounttmp)
+ self.assertNotIn('testcollection', d1)
+
+ d1 = llfuse.listdir(os.path.join(mounttmp, 'Unrestricted public data'))
+ self.assertIn('testcollection', d1)
+
+ Test().runTest()
+
+class FuseProjectMvTest(MountTestBase):
+ def runTest(self):
+ self.make_mount(fuse.ProjectDirectory,
+ project_object=self.api.users().current().execute())
+
+ self.pool.apply(fuseProjectMvTestHelper1, (self.mounttmp,))
+
+
+def fuseFsyncTestHelper(mounttmp, k):
+ class Test(unittest.TestCase):
+ def runTest(self):
+ fd = os.open(os.path.join(mounttmp, k), os.O_RDONLY)
+ os.fsync(fd)
+ os.close(fd)
+
+ Test().runTest()
+
+class FuseFsyncTest(FuseMagicTest):
+ def runTest(self):
+ self.make_mount(fuse.MagicDirectory)
+ self.pool.apply(fuseFsyncTestHelper, (self.mounttmp, self.testcollection))
+
+
+class MagicDirApiError(FuseMagicTest):
+ def setUp(self):
+ api = mock.MagicMock()
+ super(MagicDirApiError, self).setUp(api=api)
+ api.collections().get().execute.side_effect = iter([Exception('API fail'), {"manifest_text": self.test_manifest}])
+ api.keep.get.side_effect = Exception('Keep fail')
+
+ def runTest(self):
+ self.make_mount(fuse.MagicDirectory)
+
+ self.operations.inodes.inode_cache.cap = 1
+ self.operations.inodes.inode_cache.min_entries = 2
+
+ with self.assertRaises(OSError):
+ llfuse.listdir(os.path.join(self.mounttmp, self.testcollection))
+
+ llfuse.listdir(os.path.join(self.mounttmp, self.testcollection))