(self)
| 209 | self.assertCollationInLastCommand() |
| 210 | |
| 211 | def test_bulk_write(self): |
| 212 | self.db.test.collection.bulk_write( |
| 213 | [ |
| 214 | DeleteOne({"noCollation": 42}), |
| 215 | DeleteMany({"noCollation": 42}), |
| 216 | DeleteOne({"foo": 42}, collation=self.collation), |
| 217 | DeleteMany({"foo": 42}, collation=self.collation), |
| 218 | ReplaceOne({"noCollation": 24}, {"bar": 42}), |
| 219 | UpdateOne({"noCollation": 84}, {"$set": {"bar": 10}}, upsert=True), |
| 220 | UpdateMany({"noCollation": 45}, {"$set": {"bar": 42}}), |
| 221 | ReplaceOne({"foo": 24}, {"foo": 42}, collation=self.collation), |
| 222 | UpdateOne( |
| 223 | {"foo": 84}, {"$set": {"foo": 10}}, upsert=True, collation=self.collation |
| 224 | ), |
| 225 | UpdateMany({"foo": 45}, {"$set": {"foo": 42}}, collation=self.collation), |
| 226 | ] |
| 227 | ) |
| 228 | |
| 229 | delete_cmd = self.listener.started_events[0].command |
| 230 | update_cmd = self.listener.started_events[1].command |
| 231 | |
| 232 | def check_ops(ops): |
| 233 | for op in ops: |
| 234 | if "noCollation" in op["q"]: |
| 235 | self.assertNotIn("collation", op) |
| 236 | else: |
| 237 | self.assertEqual(self.collation.document, op["collation"]) |
| 238 | |
| 239 | check_ops(delete_cmd["deletes"]) |
| 240 | check_ops(update_cmd["updates"]) |
| 241 | |
| 242 | def test_indexes_same_keys_different_collations(self): |
| 243 | self.db.test.drop() |
nothing calls this directly
no test coverage detected