Method test_gssapi_threaded

test/asynchronous/test_auth.py:216–273

(self)

    @ignore_deprecations
    @async_client_context.require_sync
    async def test_gssapi_threaded(self):
        client = self.simple_client(
            GSSAPI_HOST,
            GSSAPI_PORT,
            username=GSSAPI_PRINCIPAL,
            password=GSSAPI_PASS,
            authMechanism="GSSAPI",
            authMechanismProperties=self.mech_properties,
        )

        # Authentication succeeded?
        await client.server_info()
        db = client[GSSAPI_DB]

        # Need one document in the collection. AutoAuthenticateThread does
        # collection.find_one with a 1-second delay, forcing it to check out
        # multiple connections from the pool concurrently, proving that
        # auto-authentication works with GSSAPI.
        collection = db.test
        if not await collection.count_documents({}):
            try:
                await collection.drop()
                await collection.insert_one({"_id": 1})
            except OperationFailure:
                raise SkipTest("User must be able to write.")

        threads = []
        for _ in range(4):
            threads.append(AutoAuthenticateThread(collection))
        for thread in threads:
            thread.start()
        for thread in threads:
            thread.join()
            self.assertTrue(thread.success)

        set_name = async_client_context.replica_set_name
        if set_name:
            client = self.simple_client(
                GSSAPI_HOST,
                GSSAPI_PORT,
                username=GSSAPI_PRINCIPAL,
                password=GSSAPI_PASS,
                authMechanism="GSSAPI",
                authMechanismProperties=self.mech_properties,
                replicaSet=set_name,
            )

            # Succeeded?
            await client.server_info()

            threads = []
            for _ in range(4):
                threads.append(AutoAuthenticateThread(collection))
            for thread in threads:
                thread.start()
            for thread in threads:
                thread.join()
                self.assertTrue(thread.success)