Method test_scram_threaded

test/asynchronous/test_auth.py:656–671

(self)

    @async_client_context.require_sync
    async def test_scram_threaded(self):
        coll = async_client_context.client.db.test
        await coll.drop()
        await coll.insert_one({"_id": 1})

        # The first thread to call find() will authenticate
        client = await self.async_rs_or_single_client()
        coll = client.db.test
        threads = []
        for _ in range(4):
            threads.append(AutoAuthenticateThread(coll))
        for thread in threads:
            thread.start()
        for thread in threads:
            thread.join()
            self.assertTrue(thread.success)


class TestAuthURIOptions(AsyncIntegrationTest):