(self)
| 1111 | self.assertEqual(10, len(db.test.find().batch_size(5).to_list())) |
| 1112 | |
| 1113 | def test_tailable(self): |
| 1114 | db = self.db |
| 1115 | db.drop_collection("test") |
| 1116 | db.create_collection("test", capped=True, size=1000, max=3) |
| 1117 | self.addCleanup(db.drop_collection, "test") |
| 1118 | cursor = db.test.find(cursor_type=CursorType.TAILABLE) |
| 1119 | |
| 1120 | db.test.insert_one({"x": 1}) |
| 1121 | count = 0 |
| 1122 | for doc in cursor: |
| 1123 | count += 1 |
| 1124 | self.assertEqual(1, doc["x"]) |
| 1125 | self.assertEqual(1, count) |
| 1126 | |
| 1127 | db.test.insert_one({"x": 2}) |
| 1128 | count = 0 |
| 1129 | for doc in cursor: |
| 1130 | count += 1 |
| 1131 | self.assertEqual(2, doc["x"]) |
| 1132 | self.assertEqual(1, count) |
| 1133 | |
| 1134 | db.test.insert_one({"x": 3}) |
| 1135 | count = 0 |
| 1136 | for doc in cursor: |
| 1137 | count += 1 |
| 1138 | self.assertEqual(3, doc["x"]) |
| 1139 | self.assertEqual(1, count) |
| 1140 | |
| 1141 | # Capped rollover - the collection can never |
| 1142 | # have more than 3 documents. Just make sure |
| 1143 | # this doesn't raise... |
| 1144 | db.test.insert_many([{"x": i} for i in range(4, 7)]) |
| 1145 | self.assertEqual(0, len(cursor.to_list())) |
| 1146 | |
| 1147 | # and that the cursor doesn't think it's still alive. |
| 1148 | self.assertFalse(cursor.alive) |
| 1149 | |
| 1150 | self.assertEqual(3, db.test.count_documents({})) |
| 1151 | |
| 1152 | # __getitem__(index) |
| 1153 | if _IS_SYNC: |
| 1154 | for cursor in ( |
| 1155 | db.test.find(cursor_type=CursorType.TAILABLE), |
| 1156 | db.test.find(cursor_type=CursorType.TAILABLE_AWAIT), |
| 1157 | ): |
| 1158 | self.assertEqual(4, cursor[0]["x"]) |
| 1159 | self.assertEqual(5, cursor[1]["x"]) |
| 1160 | self.assertEqual(6, cursor[2]["x"]) |
| 1161 | |
| 1162 | cursor.rewind() |
| 1163 | self.assertEqual([4], [doc["x"] for doc in cursor[0:1]]) |
| 1164 | cursor.rewind() |
| 1165 | self.assertEqual([5], [doc["x"] for doc in cursor[1:2]]) |
| 1166 | cursor.rewind() |
| 1167 | self.assertEqual([6], [doc["x"] for doc in cursor[2:3]]) |
| 1168 | cursor.rewind() |
| 1169 | self.assertEqual([4, 5], [doc["x"] for doc in cursor[0:2]]) |
| 1170 | cursor.rewind() |
nothing calls this directly
no test coverage detected