(self)
| 447 | f.close() |
| 448 | |
| 449 | def test_gridfs_find(self): |
| 450 | self.fs.put(b"test2", filename="two") |
| 451 | time.sleep(0.01) |
| 452 | self.fs.put(b"test2+", filename="two") |
| 453 | time.sleep(0.01) |
| 454 | self.fs.put(b"test1", filename="one") |
| 455 | time.sleep(0.01) |
| 456 | self.fs.put(b"test2++", filename="two") |
| 457 | files = self.db.fs.files |
| 458 | self.assertEqual(3, files.count_documents({"filename": "two"})) |
| 459 | self.assertEqual(4, files.count_documents({})) |
| 460 | cursor = self.fs.find(no_cursor_timeout=False).sort("uploadDate", -1).skip(1).limit(2) |
| 461 | gout = cursor.next() |
| 462 | self.assertEqual(b"test1", gout.read()) |
| 463 | cursor.rewind() |
| 464 | gout = cursor.next() |
| 465 | self.assertEqual(b"test1", gout.read()) |
| 466 | gout = cursor.next() |
| 467 | self.assertEqual(b"test2+", gout.read()) |
| 468 | with self.assertRaises(StopIteration): |
| 469 | cursor.__next__() |
| 470 | cursor.rewind() |
| 471 | items = cursor.to_list() |
| 472 | self.assertEqual(len(items), 2) |
| 473 | cursor.rewind() |
| 474 | items = cursor.to_list(1) |
| 475 | self.assertEqual(len(items), 1) |
| 476 | cursor.close() |
| 477 | self.assertRaises(TypeError, self.fs.find, {}, {"_id": True}) |
| 478 | |
| 479 | def test_delete_not_initialized(self): |
| 480 | # Creating a cursor with invalid arguments will not run __init__ |
nothing calls this directly
no test coverage detected