Method test_single_error_ordered_batch

test/test_bulk.py:504–544

(self)

        )

    def test_single_error_ordered_batch(self):
        self.coll.create_index("a", unique=True)
        self.addCleanup(self.coll.drop_index, [("a", 1)])
        requests: list = [
            InsertOne({"b": 1, "a": 1}),
            UpdateOne({"b": 2}, {"$set": {"a": 1}}, upsert=True),
            InsertOne({"b": 3, "a": 2}),
        ]
        try:
            self.coll.bulk_write(requests)
        except BulkWriteError as exc:
            result = exc.details
            self.assertEqual(exc.code, 65)
        else:
            self.fail("Error not raised")

        self.assertEqualResponse(
            {
                "nMatched": 0,
                "nModified": 0,
                "nUpserted": 0,
                "nInserted": 1,
                "nRemoved": 0,
                "upserted": [],
                "writeConcernErrors": [],
                "writeErrors": [
                    {
                        "index": 1,
                        "code": 11000,
                        "errmsg": "...",
                        "op": {
                            "q": {"b": 2},
                            "u": {"$set": {"a": 1}},
                            "multi": False,
                            "upsert": True,
                        },
                    }
                ],
            },
            result,
        )

    def test_multiple_error_ordered_batch(self):
        self.coll.create_index("a", unique=True)