| 47 | return GridFSFile(self.collection, doc) |
| 48 | |
| 49 | def test_insert(self): |
| 50 | def test_insert_file(data, filename, read_size): |
| 51 | # insert file |
| 52 | id = self.fs.put(data, filename=filename, encoding='utf8') |
| 53 | doc = self.collection.files.find_one(id) |
| 54 | f = self.get_file(doc) |
| 55 | |
| 56 | # test metadata |
| 57 | self.assertEqual(id, f._id) |
| 58 | self.assertEqual(filename, f.filename) |
| 59 | |
| 60 | # test data |
| 61 | result = [] |
| 62 | while True: |
| 63 | s = f.read(read_size) |
| 64 | if len(s) > 0: |
| 65 | result.append(s.decode('utf8')) |
| 66 | if read_size >= 0: |
| 67 | self.assertLessEqual(len(s), read_size) |
| 68 | else: |
| 69 | break |
| 70 | result = "".join(result) |
| 71 | self.assertEqual(f.length, len(result)) |
| 72 | self.assertEqual(data, result) |
| 73 | |
| 74 | # test with 1-chunk files |
| 75 | test_insert_file("hello world", "hello.txt", -1) |
| 76 | test_insert_file("hello world 2", "hello.txt", 10) |
| 77 | test_insert_file("hello world 3", "hello.txt", 100) |
| 78 | |
| 79 | # test with multiple-chunk files |
| 80 | size = 4 * 1024 * 1024 |
| 81 | bigger = "".join([chr(ord('a') + (n % 26)) for n in range(size)]) |
| 82 | test_insert_file(bigger, "bigger.txt", -1) |
| 83 | test_insert_file(bigger, "bigger.txt", 1024) |
| 84 | test_insert_file(bigger, "bigger.txt", 1024 * 1024) |
| 85 | |
| 86 | def test_missing_chunk(self): |
| 87 | data = "test data" |