Method test_scram

test/asynchronous/test_auth.py:476–565

(self)

    @async_client_context.require_no_fips
    async def test_scram(self):
        # Step 1: create users
        await async_client_context.create_user(
            "testscram", "sha1", "pwd", roles=["dbOwner"], mechanisms=["SCRAM-SHA-1"]
        )
        await async_client_context.create_user(
            "testscram", "sha256", "pwd", roles=["dbOwner"], mechanisms=["SCRAM-SHA-256"]
        )
        await async_client_context.create_user(
            "testscram",
            "both",
            "pwd",
            roles=["dbOwner"],
            mechanisms=["SCRAM-SHA-1", "SCRAM-SHA-256"],
        )

        # Step 2: verify auth success cases
        client = await self.async_rs_or_single_client_noauth(
            username="sha1", password="pwd", authSource="testscram"
        )
        await client.testscram.command("dbstats")

        client = await self.async_rs_or_single_client_noauth(
            username="sha1", password="pwd", authSource="testscram", authMechanism="SCRAM-SHA-1"
        )
        await client.testscram.command("dbstats")

        client = await self.async_rs_or_single_client_noauth(
            username="sha256", password="pwd", authSource="testscram"
        )
        await client.testscram.command("dbstats")

        client = await self.async_rs_or_single_client_noauth(
            username="sha256", password="pwd", authSource="testscram", authMechanism="SCRAM-SHA-256"
        )
        await client.testscram.command("dbstats")

        # Step 2: SCRAM-SHA-1 and SCRAM-SHA-256
        client = await self.async_rs_or_single_client_noauth(
            username="both", password="pwd", authSource="testscram", authMechanism="SCRAM-SHA-1"
        )
        await client.testscram.command("dbstats")
        client = await self.async_rs_or_single_client_noauth(
            username="both", password="pwd", authSource="testscram", authMechanism="SCRAM-SHA-256"
        )
        await client.testscram.command("dbstats")

        self.listener.reset()
        client = await self.async_rs_or_single_client_noauth(
            username="both", password="pwd", authSource="testscram", event_listeners=[self.listener]
        )
        await client.testscram.command("dbstats")
        if async_client_context.version.at_least(4, 4, -1):
            # Speculative authentication in 4.4+ sends saslStart with the
            # handshake.
            self.assertEqual(self.listener.started_events, [])
        else:
            started = self.listener.started_events[0]