Method test_mongodb_x509_auth

test/test_ssl.py:540–645

(self)

    @client_context.require_no_api_version
    @ignore_deprecations
    def test_mongodb_x509_auth(self):
        host, port = client_context.host, client_context.port
        self.addCleanup(remove_all_users, client_context.client["$external"])

        # Give x509 user all necessary privileges.
        client_context.create_user(
            "$external",
            MONGODB_X509_USERNAME,
            roles=[
                {"role": "readWriteAnyDatabase", "db": "admin"},
                {"role": "userAdminAnyDatabase", "db": "admin"},
            ],
        )

        noauth = self.simple_client(
            client_context.pair,
            ssl=True,
            tlsAllowInvalidCertificates=True,
            tlsCertificateKeyFile=CLIENT_PEM,
        )

        with self.assertRaises(OperationFailure):
            noauth.pymongo_test.test.find_one()

        listener = EventListener()
        auth = self.simple_client(
            client_context.pair,
            authMechanism="MONGODB-X509",
            ssl=True,
            tlsAllowInvalidCertificates=True,
            tlsCertificateKeyFile=CLIENT_PEM,
            event_listeners=[listener],
        )

        # No error
        auth.pymongo_test.test.find_one()
        names = listener.started_command_names()
        if client_context.version.at_least(4, 4, -1):
            # Speculative auth skips the authenticate command.
            self.assertEqual(names, ["find"])
        else:
            self.assertEqual(names, ["authenticate", "find"])

        uri = "mongodb://%s@%s:%d/?authMechanism=MONGODB-X509" % (
            quote_plus(MONGODB_X509_USERNAME),
            host,
            port,
        )
        client = self.simple_client(
            uri, ssl=True, tlsAllowInvalidCertificates=True, tlsCertificateKeyFile=CLIENT_PEM
        )
        # No error
        client.pymongo_test.test.find_one()

        uri = "mongodb://%s:%d/?authMechanism=MONGODB-X509" % (host, port)
        client = self.simple_client(
            uri, ssl=True, tlsAllowInvalidCertificates=True, tlsCertificateKeyFile=CLIENT_PEM
        )