(self)
| 463 | |
| 464 | @patch("gridfs.asynchronous.grid_file._UPLOAD_BUFFER_SIZE", 5) |
| 465 | async def test_abort(self): |
| 466 | gin = self.fs.open_upload_stream("test_filename", chunk_size_bytes=5) |
| 467 | await gin.write(b"test1") |
| 468 | await gin.write(b"test2") |
| 469 | await gin.write(b"test3") |
| 470 | self.assertEqual(3, await self.db.fs.chunks.count_documents({"files_id": gin._id})) |
| 471 | await gin.abort() |
| 472 | self.assertTrue(gin.closed) |
| 473 | with self.assertRaises(ValueError): |
| 474 | await gin.write(b"test4") |
| 475 | self.assertEqual(0, await self.db.fs.chunks.count_documents({"files_id": gin._id})) |
| 476 | |
| 477 | async def test_download_to_stream(self): |
| 478 | file1 = BytesIO(b"hello world") |
nothing calls this directly
no test coverage detected