test_fs.py 文件源码

def test_fs_replicas(self):
        mock_clients = MockClients(self.user)
        with mock.patch('main.models.User.get_clients',
                        mock_clients.get_clients):
            fs = get_fs(self.user, chunk_size=3, replicas=2)

            with BytesIO(TEST_FILE) as f:
                file = fs.upload('/foo', f)

            mock_clients.clients[2].data.clear()

            self.assertEqual('/foo', file.path)

            with BytesIO() as o:
                with fs.download('/foo') as f:
                    shutil.copyfileobj(f, o)
                    self.assertEqual(TEST_FILE, o.getvalue())

            with self.assertRaises(FileNotFoundError):
                fs.download('/barfoo')

            fs.delete('/foo')