| 358 | self.assertEqual(b"hello world", (self.fs.get(oid)).read()) |
| 359 | |
| 360 | def test_file_exists(self): |
| 361 | oid = self.fs.put(b"hello") |
| 362 | with self.assertRaises(FileExists): |
| 363 | self.fs.put(b"world", _id=oid) |
| 364 | |
| 365 | one = self.fs.new_file(_id=123) |
| 366 | one.write(b"some content") |
| 367 | one.close() |
| 368 | |
| 369 | # Attempt to upload a file with more chunks to the same _id. |
| 370 | with patch("gridfs.synchronous.grid_file._UPLOAD_BUFFER_SIZE", DEFAULT_CHUNK_SIZE): |
| 371 | two = self.fs.new_file(_id=123) |
| 372 | with self.assertRaises(FileExists): |
| 373 | two.write(b"x" * DEFAULT_CHUNK_SIZE * 3) |
| 374 | # Original file is still readable (no extra chunks were uploaded). |
| 375 | self.assertEqual((self.fs.get(123)).read(), b"some content") |
| 376 | |
| 377 | two = self.fs.new_file(_id=123) |
| 378 | two.write(b"some content") |
| 379 | with self.assertRaises(FileExists): |
| 380 | two.close() |
| 381 | # Original file is still readable. |
| 382 | self.assertEqual((self.fs.get(123)).read(), b"some content") |
| 383 | |
| 384 | def test_exists(self): |
| 385 | oid = self.fs.put(b"hello") |