(self)
| 267 | self.assertEqual(self.coll.count_documents({}), 1) |
| 268 | |
| 269 | def test_upsert(self): |
| 270 | expected = { |
| 271 | "nMatched": 0, |
| 272 | "nModified": 0, |
| 273 | "nUpserted": 1, |
| 274 | "nInserted": 0, |
| 275 | "nRemoved": 0, |
| 276 | "upserted": [{"index": 0, "_id": "..."}], |
| 277 | } |
| 278 | |
| 279 | result = self.coll.bulk_write([ReplaceOne({}, {"foo": "bar"}, upsert=True)]) |
| 280 | self.assertEqualResponse(expected, result.bulk_api_result) |
| 281 | self.assertEqual(1, result.upserted_count) |
| 282 | assert result.upserted_ids is not None |
| 283 | self.assertEqual(1, len(result.upserted_ids)) |
| 284 | self.assertIsInstance(result.upserted_ids.get(0), ObjectId) |
| 285 | |
| 286 | self.assertEqual(self.coll.count_documents({"foo": "bar"}), 1) |
| 287 | |
| 288 | def test_numerous_inserts(self): |
| 289 | # Ensure we don't exceed server's maxWriteBatchSize size limit. |
nothing calls this directly
no test coverage detected