(self)
| 705 | db.drop_collection("test") |
| 706 | |
| 707 | def test_insert_one(self): |
| 708 | db = self.db |
| 709 | db.test.drop() |
| 710 | |
| 711 | document: dict[str, Any] = {"_id": 1000} |
| 712 | result = db.test.insert_one(document) |
| 713 | self.assertIsInstance(result, InsertOneResult) |
| 714 | self.assertIsInstance(result.inserted_id, int) |
| 715 | self.assertEqual(document["_id"], result.inserted_id) |
| 716 | self.assertTrue(result.acknowledged) |
| 717 | self.assertIsNotNone(db.test.find_one({"_id": document["_id"]})) |
| 718 | self.assertEqual(1, db.test.count_documents({})) |
| 719 | |
| 720 | document = {"foo": "bar"} |
| 721 | result = db.test.insert_one(document) |
| 722 | self.assertIsInstance(result, InsertOneResult) |
| 723 | self.assertIsInstance(result.inserted_id, ObjectId) |
| 724 | self.assertEqual(document["_id"], result.inserted_id) |
| 725 | self.assertTrue(result.acknowledged) |
| 726 | self.assertIsNotNone(db.test.find_one({"_id": document["_id"]})) |
| 727 | self.assertEqual(2, db.test.count_documents({})) |
| 728 | |
| 729 | db = db.client.get_database(db.name, write_concern=WriteConcern(w=0)) |
| 730 | result = db.test.insert_one(document) |
| 731 | self.assertIsInstance(result, InsertOneResult) |
| 732 | self.assertIsInstance(result.inserted_id, ObjectId) |
| 733 | self.assertEqual(document["_id"], result.inserted_id) |
| 734 | self.assertFalse(result.acknowledged) |
| 735 | # The insert failed duplicate key... |
| 736 | |
| 737 | def async_lambda(): |
| 738 | return db.test.count_documents({}) == 2 |
| 739 | |
| 740 | wait_until(async_lambda, "forcing duplicate key error") |
| 741 | |
| 742 | document = RawBSONDocument(encode({"_id": ObjectId(), "foo": "bar"})) |
| 743 | result = db.test.insert_one(document) |
| 744 | self.assertIsInstance(result, InsertOneResult) |
| 745 | self.assertEqual(result.inserted_id, None) |
| 746 | |
| 747 | def test_insert_many(self): |
| 748 | db = self.db |
nothing calls this directly
no test coverage detected