Method test_no_remove

test/test_bulk.py:907–923

(self)

            coll.bulk_write([InsertOne({"x": 1})])

    def test_no_remove(self):
        # We test that an authorization failure aborts the batch and is raised
        # as OperationFailure.
        cli = self.rs_or_single_client_noauth(
            username="noremove", password="pw", authSource="pymongo_test"
        )
        coll = cli.pymongo_test.test
        coll.find_one()
        requests = [
            InsertOne({"x": 1}),
            ReplaceOne({"x": 2}, {"x": 2}, upsert=True),
            DeleteMany({}),  # Prohibited.
            InsertOne({"x": 3}),  # Never attempted.
        ]
        with self.assertRaises(OperationFailure):
            coll.bulk_write(requests)  # type: ignore[arg-type]
        self.assertEqual({1, 2}, set(self.coll.distinct("x")))


class TestBulkWriteConcern(BulkTestBase):