Method test_scram_threaded

test/test_auth.py:652–667

(self)

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

        # The first thread to call find() will authenticate
        client = self.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(IntegrationTest):