ydb/contrib/python/google-auth/py3/tests/test_pluggable.py

# Copyright 2022 Google LLC
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#      http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

import json
import os
import subprocess

import mock
import pytest  # type: ignore

from google.auth import exceptions
from google.auth import pluggable
from .test__default import WORKFORCE_AUDIENCE

CLIENT_ID = "username"
CLIENT_SECRET = "password"
# Base64 encoding of "username:password".
BASIC_AUTH_ENCODING = "dXNlcm5hbWU6cGFzc3dvcmQ="
SERVICE_ACCOUNT_EMAIL = "service-1234@service-name.iam.gserviceaccount.com"
SERVICE_ACCOUNT_IMPERSONATION_URL_BASE = (
    "https://us-east1-iamcredentials.googleapis.com"
)
SERVICE_ACCOUNT_IMPERSONATION_URL_ROUTE = "/v1/projects/-/serviceAccounts/{}:generateAccessToken".format(
    SERVICE_ACCOUNT_EMAIL
)
SERVICE_ACCOUNT_IMPERSONATION_URL = (
    SERVICE_ACCOUNT_IMPERSONATION_URL_BASE + SERVICE_ACCOUNT_IMPERSONATION_URL_ROUTE
)
QUOTA_PROJECT_ID = "QUOTA_PROJECT_ID"
SCOPES = ["scope1", "scope2"]
SUBJECT_TOKEN_FIELD_NAME = "access_token"

TOKEN_URL = "https://sts.googleapis.com/v1/token"
TOKEN_INFO_URL = "https://sts.googleapis.com/v1/introspect"
SUBJECT_TOKEN_TYPE = "urn:ietf:params:oauth:token-type:jwt"
AUDIENCE = "//iam.googleapis.com/projects/123456/locations/global/workloadIdentityPools/POOL_ID/providers/PROVIDER_ID"
DEFAULT_UNIVERSE_DOMAIN = "googleapis.com"

VALID_TOKEN_URLS = [
    "https://sts.googleapis.com",
    "https://us-east-1.sts.googleapis.com",
    "https://US-EAST-1.sts.googleapis.com",
    "https://sts.us-east-1.googleapis.com",
    "https://sts.US-WEST-1.googleapis.com",
    "https://us-east-1-sts.googleapis.com",
    "https://US-WEST-1-sts.googleapis.com",
    "https://us-west-1-sts.googleapis.com/path?query",
    "https://sts-us-east-1.p.googleapis.com",
]
INVALID_TOKEN_URLS = [
    "https://iamcredentials.googleapis.com",
    "sts.googleapis.com",
    "https://",
    "http://sts.googleapis.com",
    "https://st.s.googleapis.com",
    "https://us-eas\t-1.sts.googleapis.com",
    "https:/us-east-1.sts.googleapis.com",
    "https://US-WE/ST-1-sts.googleapis.com",
    "https://sts-us-east-1.googleapis.com",
    "https://sts-US-WEST-1.googleapis.com",
    "testhttps://us-east-1.sts.googleapis.com",
    "https://us-east-1.sts.googleapis.comevil.com",
    "https://us-east-1.us-east-1.sts.googleapis.com",
    "https://us-ea.s.t.sts.googleapis.com",
    "https://sts.googleapis.comevil.com",
    "hhttps://us-east-1.sts.googleapis.com",
    "https://us- -1.sts.googleapis.com",
    "https://-sts.googleapis.com",
    "https://us-east-1.sts.googleapis.com.evil.com",
    "https://sts.pgoogleapis.com",
    "https://p.googleapis.com",
    "https://sts.p.com",
    "http://sts.p.googleapis.com",
    "https://xyz-sts.p.googleapis.com",
    "https://sts-xyz.123.p.googleapis.com",
    "https://sts-xyz.p1.googleapis.com",
    "https://sts-xyz.p.foo.com",
    "https://sts-xyz.p.foo.googleapis.com",
]
VALID_SERVICE_ACCOUNT_IMPERSONATION_URLS = [
    "https://iamcredentials.googleapis.com",
    "https://us-east-1.iamcredentials.googleapis.com",
    "https://US-EAST-1.iamcredentials.googleapis.com",
    "https://iamcredentials.us-east-1.googleapis.com",
    "https://iamcredentials.US-WEST-1.googleapis.com",
    "https://us-east-1-iamcredentials.googleapis.com",
    "https://US-WEST-1-iamcredentials.googleapis.com",
    "https://us-west-1-iamcredentials.googleapis.com/path?query",
    "https://iamcredentials-us-east-1.p.googleapis.com",
]
INVALID_SERVICE_ACCOUNT_IMPERSONATION_URLS = [
    "https://sts.googleapis.com",
    "iamcredentials.googleapis.com",
    "https://",
    "http://iamcredentials.googleapis.com",
    "https://iamcre.dentials.googleapis.com",
    "https://us-eas\t-1.iamcredentials.googleapis.com",
    "https:/us-east-1.iamcredentials.googleapis.com",
    "https://US-WE/ST-1-iamcredentials.googleapis.com",
    "https://iamcredentials-us-east-1.googleapis.com",
    "https://iamcredentials-US-WEST-1.googleapis.com",
    "testhttps://us-east-1.iamcredentials.googleapis.com",
    "https://us-east-1.iamcredentials.googleapis.comevil.com",
    "https://us-east-1.us-east-1.iamcredentials.googleapis.com",
    "https://us-ea.s.t.iamcredentials.googleapis.com",
    "https://iamcredentials.googleapis.comevil.com",
    "hhttps://us-east-1.iamcredentials.googleapis.com",
    "https://us- -1.iamcredentials.googleapis.com",
    "https://-iamcredentials.googleapis.com",
    "https://us-east-1.iamcredentials.googleapis.com.evil.com",
    "https://iamcredentials.pgoogleapis.com",
    "https://p.googleapis.com",
    "https://iamcredentials.p.com",
    "http://iamcredentials.p.googleapis.com",
    "https://xyz-iamcredentials.p.googleapis.com",
    "https://iamcredentials-xyz.123.p.googleapis.com",
    "https://iamcredentials-xyz.p1.googleapis.com",
    "https://iamcredentials-xyz.p.foo.com",
    "https://iamcredentials-xyz.p.foo.googleapis.com",
]


class TestCredentials(object):
    CREDENTIAL_SOURCE_EXECUTABLE_COMMAND = (
        "/fake/external/excutable --arg1=value1 --arg2=value2"
    )
    CREDENTIAL_SOURCE_EXECUTABLE_OUTPUT_FILE = "fake_output_file"
    CREDENTIAL_SOURCE_EXECUTABLE = {
        "command": CREDENTIAL_SOURCE_EXECUTABLE_COMMAND,
        "timeout_millis": 30000,
        "interactive_timeout_millis": 300000,
        "output_file": CREDENTIAL_SOURCE_EXECUTABLE_OUTPUT_FILE,
    }
    CREDENTIAL_SOURCE = {"executable": CREDENTIAL_SOURCE_EXECUTABLE}
    EXECUTABLE_OIDC_TOKEN = "FAKE_ID_TOKEN"
    EXECUTABLE_SUCCESSFUL_OIDC_RESPONSE_ID_TOKEN = {
        "version": 1,
        "success": True,
        "token_type": "urn:ietf:params:oauth:token-type:id_token",
        "id_token": EXECUTABLE_OIDC_TOKEN,
        "expiration_time": 9999999999,
    }
    EXECUTABLE_SUCCESSFUL_OIDC_NO_EXPIRATION_TIME_RESPONSE_ID_TOKEN = {
        "version": 1,
        "success": True,
        "token_type": "urn:ietf:params:oauth:token-type:id_token",
        "id_token": EXECUTABLE_OIDC_TOKEN,
    }
    EXECUTABLE_SUCCESSFUL_OIDC_RESPONSE_JWT = {
        "version": 1,
        "success": True,
        "token_type": "urn:ietf:params:oauth:token-type:jwt",
        "id_token": EXECUTABLE_OIDC_TOKEN,
        "expiration_time": 9999999999,
    }
    EXECUTABLE_SUCCESSFUL_OIDC_NO_EXPIRATION_TIME_RESPONSE_JWT = {
        "version": 1,
        "success": True,
        "token_type": "urn:ietf:params:oauth:token-type:jwt",
        "id_token": EXECUTABLE_OIDC_TOKEN,
    }
    EXECUTABLE_SAML_TOKEN = "FAKE_SAML_RESPONSE"
    EXECUTABLE_SUCCESSFUL_SAML_RESPONSE = {
        "version": 1,
        "success": True,
        "token_type": "urn:ietf:params:oauth:token-type:saml2",
        "saml_response": EXECUTABLE_SAML_TOKEN,
        "expiration_time": 9999999999,
    }
    EXECUTABLE_SUCCESSFUL_SAML_NO_EXPIRATION_TIME_RESPONSE = {
        "version": 1,
        "success": True,
        "token_type": "urn:ietf:params:oauth:token-type:saml2",
        "saml_response": EXECUTABLE_SAML_TOKEN,
    }
    EXECUTABLE_FAILED_RESPONSE = {
        "version": 1,
        "success": False,
        "code": "401",
        "message": "Permission denied. Caller not authorized",
    }
    CREDENTIAL_URL = "http://fakeurl.com"

    @classmethod
    def make_pluggable(
        cls,
        audience=AUDIENCE,
        subject_token_type=SUBJECT_TOKEN_TYPE,
        token_url=TOKEN_URL,
        token_info_url=TOKEN_INFO_URL,
        client_id=None,
        client_secret=None,
        quota_project_id=None,
        scopes=None,
        default_scopes=None,
        service_account_impersonation_url=None,
        credential_source=None,
        workforce_pool_user_project=None,
        interactive=None,
    ):
        return pluggable.Credentials(
            audience=audience,
            subject_token_type=subject_token_type,
            token_url=token_url,
            token_info_url=token_info_url,
            service_account_impersonation_url=service_account_impersonation_url,
            credential_source=credential_source,
            client_id=client_id,
            client_secret=client_secret,
            quota_project_id=quota_project_id,
            scopes=scopes,
            default_scopes=default_scopes,
            workforce_pool_user_project=workforce_pool_user_project,
            interactive=interactive,
        )

    def test_from_constructor_and_injection(self):
        credentials = pluggable.Credentials(
            audience=AUDIENCE,
            subject_token_type=SUBJECT_TOKEN_TYPE,
            token_url=TOKEN_URL,
            token_info_url=TOKEN_INFO_URL,
            credential_source=self.CREDENTIAL_SOURCE,
            interactive=True,
        )
        setattr(credentials, "_tokeninfo_username", "mock_external_account_id")

        assert isinstance(credentials, pluggable.Credentials)
        assert credentials.interactive
        assert credentials.external_account_id == "mock_external_account_id"

    @mock.patch.object(pluggable.Credentials, "__init__", return_value=None)
    def test_from_info_full_options(self, mock_init):
        credentials = pluggable.Credentials.from_info(
            {
                "audience": AUDIENCE,
                "subject_token_type": SUBJECT_TOKEN_TYPE,
                "token_url": TOKEN_URL,
                "token_info_url": TOKEN_INFO_URL,
                "service_account_impersonation_url": SERVICE_ACCOUNT_IMPERSONATION_URL,
                "service_account_impersonation": {"token_lifetime_seconds": 2800},
                "client_id": CLIENT_ID,
                "client_secret": CLIENT_SECRET,
                "quota_project_id": QUOTA_PROJECT_ID,
                "credential_source": self.CREDENTIAL_SOURCE,
            }
        )

        # Confirm pluggable.Credentials instantiated with expected attributes.
        assert isinstance(credentials, pluggable.Credentials)
        mock_init.assert_called_once_with(
            audience=AUDIENCE,
            subject_token_type=SUBJECT_TOKEN_TYPE,
            token_url=TOKEN_URL,
            token_info_url=TOKEN_INFO_URL,
            service_account_impersonation_url=SERVICE_ACCOUNT_IMPERSONATION_URL,
            service_account_impersonation_options={"token_lifetime_seconds": 2800},
            client_id=CLIENT_ID,
            client_secret=CLIENT_SECRET,
            credential_source=self.CREDENTIAL_SOURCE,
            quota_project_id=QUOTA_PROJECT_ID,
            workforce_pool_user_project=None,
            universe_domain=DEFAULT_UNIVERSE_DOMAIN,
        )

    @mock.patch.object(pluggable.Credentials, "__init__", return_value=None)
    def test_from_info_required_options_only(self, mock_init):
        credentials = pluggable.Credentials.from_info(
            {
                "audience": AUDIENCE,
                "subject_token_type": SUBJECT_TOKEN_TYPE,
                "token_url": TOKEN_URL,
                "credential_source": self.CREDENTIAL_SOURCE,
            }
        )

        # Confirm pluggable.Credentials instantiated with expected attributes.
        assert isinstance(credentials, pluggable.Credentials)
        mock_init.assert_called_once_with(
            audience=AUDIENCE,
            subject_token_type=SUBJECT_TOKEN_TYPE,
            token_url=TOKEN_URL,
            token_info_url=None,
            service_account_impersonation_url=None,
            service_account_impersonation_options={},
            client_id=None,
            client_secret=None,
            credential_source=self.CREDENTIAL_SOURCE,
            quota_project_id=None,
            workforce_pool_user_project=None,
            universe_domain=DEFAULT_UNIVERSE_DOMAIN,
        )

    @mock.patch.object(pluggable.Credentials, "__init__", return_value=None)
    def test_from_file_full_options(self, mock_init, tmpdir):
        info = {
            "audience": AUDIENCE,
            "subject_token_type": SUBJECT_TOKEN_TYPE,
            "token_url": TOKEN_URL,
            "token_info_url": TOKEN_INFO_URL,
            "service_account_impersonation_url": SERVICE_ACCOUNT_IMPERSONATION_URL,
            "service_account_impersonation": {"token_lifetime_seconds": 2800},
            "client_id": CLIENT_ID,
            "client_secret": CLIENT_SECRET,
            "quota_project_id": QUOTA_PROJECT_ID,
            "credential_source": self.CREDENTIAL_SOURCE,
        }
        config_file = tmpdir.join("config.json")
        config_file.write(json.dumps(info))
        credentials = pluggable.Credentials.from_file(str(config_file))

        # Confirm pluggable.Credentials instantiated with expected attributes.
        assert isinstance(credentials, pluggable.Credentials)
        mock_init.assert_called_once_with(
            audience=AUDIENCE,
            subject_token_type=SUBJECT_TOKEN_TYPE,
            token_url=TOKEN_URL,
            token_info_url=TOKEN_INFO_URL,
            service_account_impersonation_url=SERVICE_ACCOUNT_IMPERSONATION_URL,
            service_account_impersonation_options={"token_lifetime_seconds": 2800},
            client_id=CLIENT_ID,
            client_secret=CLIENT_SECRET,
            credential_source=self.CREDENTIAL_SOURCE,
            quota_project_id=QUOTA_PROJECT_ID,
            workforce_pool_user_project=None,
            universe_domain=DEFAULT_UNIVERSE_DOMAIN,
        )

    @mock.patch.object(pluggable.Credentials, "__init__", return_value=None)
    def test_from_file_required_options_only(self, mock_init, tmpdir):
        info = {
            "audience": AUDIENCE,
            "subject_token_type": SUBJECT_TOKEN_TYPE,
            "token_url": TOKEN_URL,
            "credential_source": self.CREDENTIAL_SOURCE,
        }
        config_file = tmpdir.join("config.json")
        config_file.write(json.dumps(info))
        credentials = pluggable.Credentials.from_file(str(config_file))

        # Confirm pluggable.Credentials instantiated with expected attributes.
        assert isinstance(credentials, pluggable.Credentials)
        mock_init.assert_called_once_with(
            audience=AUDIENCE,
            subject_token_type=SUBJECT_TOKEN_TYPE,
            token_url=TOKEN_URL,
            token_info_url=None,
            service_account_impersonation_url=None,
            service_account_impersonation_options={},
            client_id=None,
            client_secret=None,
            credential_source=self.CREDENTIAL_SOURCE,
            quota_project_id=None,
            workforce_pool_user_project=None,
            universe_domain=DEFAULT_UNIVERSE_DOMAIN,
        )

    def test_constructor_invalid_options(self):
        credential_source = {"unsupported": "value"}

        with pytest.raises(ValueError) as excinfo:
            self.make_pluggable(credential_source=credential_source)

        assert excinfo.match(r"Missing credential_source")

    def test_constructor_invalid_credential_source(self):
        with pytest.raises(ValueError) as excinfo:
            self.make_pluggable(credential_source="non-dict")

        assert excinfo.match(r"Missing credential_source")

    def test_info_with_credential_source(self):
        credentials = self.make_pluggable(
            credential_source=self.CREDENTIAL_SOURCE.copy()
        )

        assert credentials.info == {
            "type": "external_account",
            "audience": AUDIENCE,
            "subject_token_type": SUBJECT_TOKEN_TYPE,
            "token_url": TOKEN_URL,
            "token_info_url": TOKEN_INFO_URL,
            "credential_source": self.CREDENTIAL_SOURCE,
            "universe_domain": DEFAULT_UNIVERSE_DOMAIN,
        }

    def test_token_info_url(self):
        credentials = self.make_pluggable(
            credential_source=self.CREDENTIAL_SOURCE.copy()
        )

        assert credentials.token_info_url == TOKEN_INFO_URL

    def test_token_info_url_custom(self):
        for url in VALID_TOKEN_URLS:
            credentials = self.make_pluggable(
                credential_source=self.CREDENTIAL_SOURCE.copy(),
                token_info_url=(url + "/introspect"),
            )

            assert credentials.token_info_url == url + "/introspect"

    def test_token_info_url_negative(self):
        credentials = self.make_pluggable(
            credential_source=self.CREDENTIAL_SOURCE.copy(), token_info_url=None
        )

        assert not credentials.token_info_url

    def test_token_url_custom(self):
        for url in VALID_TOKEN_URLS:
            credentials = self.make_pluggable(
                credential_source=self.CREDENTIAL_SOURCE.copy(),
                token_url=(url + "/token"),
            )

            assert credentials._token_url == (url + "/token")

    def test_service_account_impersonation_url_custom(self):
        for url in VALID_SERVICE_ACCOUNT_IMPERSONATION_URLS:
            credentials = self.make_pluggable(
                credential_source=self.CREDENTIAL_SOURCE.copy(),
                service_account_impersonation_url=(
                    url + SERVICE_ACCOUNT_IMPERSONATION_URL_ROUTE
                ),
            )

            assert credentials._service_account_impersonation_url == (
                url + SERVICE_ACCOUNT_IMPERSONATION_URL_ROUTE
            )

    @mock.patch.dict(os.environ, {"GOOGLE_EXTERNAL_ACCOUNT_ALLOW_EXECUTABLES": "1"})
    def test_retrieve_subject_token_successfully(self, tmpdir):
        ACTUAL_CREDENTIAL_SOURCE_EXECUTABLE_OUTPUT_FILE = tmpdir.join(
            "actual_output_file"
        )
        ACTUAL_CREDENTIAL_SOURCE_EXECUTABLE = {
            "command": "command",
            "interactive_timeout_millis": 300000,
            "output_file": ACTUAL_CREDENTIAL_SOURCE_EXECUTABLE_OUTPUT_FILE,
        }
        ACTUAL_CREDENTIAL_SOURCE = {"executable": ACTUAL_CREDENTIAL_SOURCE_EXECUTABLE}

        testData = {
            "subject_token_oidc_id_token": {
                "stdout": json.dumps(
                    self.EXECUTABLE_SUCCESSFUL_OIDC_RESPONSE_ID_TOKEN
                ).encode("UTF-8"),
                "impersonation_url": SERVICE_ACCOUNT_IMPERSONATION_URL,
                "file_content": self.EXECUTABLE_SUCCESSFUL_OIDC_NO_EXPIRATION_TIME_RESPONSE_ID_TOKEN,
                "expect_token": self.EXECUTABLE_OIDC_TOKEN,
            },
            "subject_token_oidc_id_token_interacitve_mode": {
                "audience": WORKFORCE_AUDIENCE,
                "file_content": self.EXECUTABLE_SUCCESSFUL_OIDC_NO_EXPIRATION_TIME_RESPONSE_ID_TOKEN,
                "interactive": True,
                "expect_token": self.EXECUTABLE_OIDC_TOKEN,
            },
            "subject_token_oidc_jwt": {
                "stdout": json.dumps(
                    self.EXECUTABLE_SUCCESSFUL_OIDC_RESPONSE_JWT
                ).encode("UTF-8"),
                "impersonation_url": SERVICE_ACCOUNT_IMPERSONATION_URL,
                "file_content": self.EXECUTABLE_SUCCESSFUL_OIDC_NO_EXPIRATION_TIME_RESPONSE_JWT,
                "expect_token": self.EXECUTABLE_OIDC_TOKEN,
            },
            "subject_token_oidc_jwt_interactive_mode": {
                "audience": WORKFORCE_AUDIENCE,
                "file_content": self.EXECUTABLE_SUCCESSFUL_OIDC_NO_EXPIRATION_TIME_RESPONSE_JWT,
                "interactive": True,
                "expect_token": self.EXECUTABLE_OIDC_TOKEN,
            },
            "subject_token_saml": {
                "stdout": json.dumps(self.EXECUTABLE_SUCCESSFUL_SAML_RESPONSE).encode(
                    "UTF-8"
                ),
                "impersonation_url": SERVICE_ACCOUNT_IMPERSONATION_URL,
                "file_content": self.EXECUTABLE_SUCCESSFUL_SAML_NO_EXPIRATION_TIME_RESPONSE,
                "expect_token": self.EXECUTABLE_SAML_TOKEN,
            },
            "subject_token_saml_interactive_mode": {
                "audience": WORKFORCE_AUDIENCE,
                "file_content": self.EXECUTABLE_SUCCESSFUL_SAML_NO_EXPIRATION_TIME_RESPONSE,
                "interactive": True,
                "expect_token": self.EXECUTABLE_SAML_TOKEN,
            },
        }

        for data in testData.values():
            with open(
                ACTUAL_CREDENTIAL_SOURCE_EXECUTABLE_OUTPUT_FILE, "w"
            ) as output_file:
                json.dump(data.get("file_content"), output_file)

            with mock.patch(
                "subprocess.run",
                return_value=subprocess.CompletedProcess(
                    args=[], stdout=data.get("stdout"), returncode=0
                ),
            ):
                credentials = self.make_pluggable(
                    audience=data.get("audience", AUDIENCE),
                    service_account_impersonation_url=data.get("impersonation_url"),
                    credential_source=ACTUAL_CREDENTIAL_SOURCE,
                    interactive=data.get("interactive", False),
                )
                subject_token = credentials.retrieve_subject_token(None)
                assert subject_token == data.get("expect_token")
            os.remove(ACTUAL_CREDENTIAL_SOURCE_EXECUTABLE_OUTPUT_FILE)

    @mock.patch.dict(os.environ, {"GOOGLE_EXTERNAL_ACCOUNT_ALLOW_EXECUTABLES": "1"})
    def test_retrieve_subject_token_saml(self):
        with mock.patch(
            "subprocess.run",
            return_value=subprocess.CompletedProcess(
                args=[],
                stdout=json.dumps(self.EXECUTABLE_SUCCESSFUL_SAML_RESPONSE).encode(
                    "UTF-8"
                ),
                returncode=0,
            ),
        ):
            credentials = self.make_pluggable(credential_source=self.CREDENTIAL_SOURCE)

            subject_token = credentials.retrieve_subject_token(None)

            assert subject_token == self.EXECUTABLE_SAML_TOKEN

    @mock.patch.dict(os.environ, {"GOOGLE_EXTERNAL_ACCOUNT_ALLOW_EXECUTABLES": "1"})
    def test_retrieve_subject_token_saml_interactive_mode(self, tmpdir):

        ACTUAL_CREDENTIAL_SOURCE_EXECUTABLE_OUTPUT_FILE = tmpdir.join(
            "actual_output_file"
        )
        ACTUAL_CREDENTIAL_SOURCE_EXECUTABLE = {
            "command": "command",
            "interactive_timeout_millis": 300000,
            "output_file": ACTUAL_CREDENTIAL_SOURCE_EXECUTABLE_OUTPUT_FILE,
        }
        ACTUAL_CREDENTIAL_SOURCE = {"executable": ACTUAL_CREDENTIAL_SOURCE_EXECUTABLE}
        with open(ACTUAL_CREDENTIAL_SOURCE_EXECUTABLE_OUTPUT_FILE, "w") as output_file:
            json.dump(
                self.EXECUTABLE_SUCCESSFUL_SAML_NO_EXPIRATION_TIME_RESPONSE, output_file
            )

        with mock.patch(
            "subprocess.run",
            return_value=subprocess.CompletedProcess(args=[], returncode=0),
        ):
            credentials = self.make_pluggable(
                audience=WORKFORCE_AUDIENCE,
                credential_source=ACTUAL_CREDENTIAL_SOURCE,
                interactive=True,
            )

            subject_token = credentials.retrieve_subject_token(None)

            assert subject_token == self.EXECUTABLE_SAML_TOKEN
            os.remove(ACTUAL_CREDENTIAL_SOURCE_EXECUTABLE_OUTPUT_FILE)

    @mock.patch.dict(os.environ, {"GOOGLE_EXTERNAL_ACCOUNT_ALLOW_EXECUTABLES": "1"})
    def test_retrieve_subject_token_failed(self):
        with mock.patch(
            "subprocess.run",
            return_value=subprocess.CompletedProcess(
                args=[],
                stdout=json.dumps(self.EXECUTABLE_FAILED_RESPONSE).encode("UTF-8"),
                returncode=0,
            ),
        ):
            credentials = self.make_pluggable(credential_source=self.CREDENTIAL_SOURCE)

            with pytest.raises(exceptions.RefreshError) as excinfo:
                _ = credentials.retrieve_subject_token(None)

            assert excinfo.match(
                r"Executable returned unsuccessful response: code: 401, message: Permission denied. Caller not authorized."
            )

    @mock.patch.dict(
        os.environ,
        {
            "GOOGLE_EXTERNAL_ACCOUNT_ALLOW_EXECUTABLES": "1",
            "GOOGLE_EXTERNAL_ACCOUNT_INTERACTIVE": "1",
        },
    )
    def test_retrieve_subject_token_failed_interactive_mode(self, tmpdir):
        ACTUAL_CREDENTIAL_SOURCE_EXECUTABLE_OUTPUT_FILE = tmpdir.join(
            "actual_output_file"
        )
        ACTUAL_CREDENTIAL_SOURCE_EXECUTABLE = {
            "command": "command",
            "interactive_timeout_millis": 300000,
            "output_file": ACTUAL_CREDENTIAL_SOURCE_EXECUTABLE_OUTPUT_FILE,
        }
        ACTUAL_CREDENTIAL_SOURCE = {"executable": ACTUAL_CREDENTIAL_SOURCE_EXECUTABLE}
        with open(
            ACTUAL_CREDENTIAL_SOURCE_EXECUTABLE_OUTPUT_FILE, "w", encoding="utf-8"
        ) as output_file:
            json.dump(self.EXECUTABLE_FAILED_RESPONSE, output_file)

        with mock.patch(
            "subprocess.run",
            return_value=subprocess.CompletedProcess(args=[], returncode=0),
        ):
            credentials = self.make_pluggable(
                audience=WORKFORCE_AUDIENCE,
                credential_source=ACTUAL_CREDENTIAL_SOURCE,
                interactive=True,
            )

            with pytest.raises(exceptions.RefreshError) as excinfo:
                _ = credentials.retrieve_subject_token(None)

            assert excinfo.match(
                r"Executable returned unsuccessful response: code: 401, message: Permission denied. Caller not authorized."
            )
            os.remove(ACTUAL_CREDENTIAL_SOURCE_EXECUTABLE_OUTPUT_FILE)

    @mock.patch.dict(os.environ, {"GOOGLE_EXTERNAL_ACCOUNT_ALLOW_EXECUTABLES": "0"})
    def test_retrieve_subject_token_not_allowd(self):
        with mock.patch(
            "subprocess.run",
            return_value=subprocess.CompletedProcess(
                args=[],
                stdout=json.dumps(
                    self.EXECUTABLE_SUCCESSFUL_OIDC_RESPONSE_ID_TOKEN
                ).encode("UTF-8"),
                returncode=0,
            ),
        ):
            credentials = self.make_pluggable(credential_source=self.CREDENTIAL_SOURCE)

            with pytest.raises(ValueError) as excinfo:
                _ = credentials.retrieve_subject_token(None)

            assert excinfo.match(r"Executables need to be explicitly allowed")

    @mock.patch.dict(os.environ, {"GOOGLE_EXTERNAL_ACCOUNT_ALLOW_EXECUTABLES": "1"})
    def test_retrieve_subject_token_invalid_version(self):
        EXECUTABLE_SUCCESSFUL_OIDC_RESPONSE_VERSION_2 = {
            "version": 2,
            "success": True,
            "token_type": "urn:ietf:params:oauth:token-type:id_token",
            "id_token": self.EXECUTABLE_OIDC_TOKEN,
            "expiration_time": 9999999999,
        }

        with mock.patch(
            "subprocess.run",
            return_value=subprocess.CompletedProcess(
                args=[],
                stdout=json.dumps(EXECUTABLE_SUCCESSFUL_OIDC_RESPONSE_VERSION_2).encode(
                    "UTF-8"
                ),
                returncode=0,
            ),
        ):
            credentials = self.make_pluggable(credential_source=self.CREDENTIAL_SOURCE)

            with pytest.raises(exceptions.RefreshError) as excinfo:
                _ = credentials.retrieve_subject_token(None)

            assert excinfo.match(r"Executable returned unsupported version.")

    @mock.patch.dict(os.environ, {"GOOGLE_EXTERNAL_ACCOUNT_ALLOW_EXECUTABLES": "1"})
    def test_retrieve_subject_token_expired_token(self):
        EXECUTABLE_SUCCESSFUL_OIDC_RESPONSE_EXPIRED = {
            "version": 1,
            "success": True,
            "token_type": "urn:ietf:params:oauth:token-type:id_token",
            "id_token": self.EXECUTABLE_OIDC_TOKEN,
            "expiration_time": 0,
        }

        with mock.patch(
            "subprocess.run",
            return_value=subprocess.CompletedProcess(
                args=[],
                stdout=json.dumps(EXECUTABLE_SUCCESSFUL_OIDC_RESPONSE_EXPIRED).encode(
                    "UTF-8"
                ),
                returncode=0,
            ),
        ):
            credentials = self.make_pluggable(credential_source=self.CREDENTIAL_SOURCE)

            with pytest.raises(exceptions.RefreshError) as excinfo:
                _ = credentials.retrieve_subject_token(None)

            assert excinfo.match(r"The token returned by the executable is expired.")

    @mock.patch.dict(os.environ, {"GOOGLE_EXTERNAL_ACCOUNT_ALLOW_EXECUTABLES": "1"})
    def test_retrieve_subject_token_file_cache(self, tmpdir):
        ACTUAL_CREDENTIAL_SOURCE_EXECUTABLE_OUTPUT_FILE = tmpdir.join(
            "actual_output_file"
        )
        ACTUAL_CREDENTIAL_SOURCE_EXECUTABLE = {
            "command": "command",
            "timeout_millis": 30000,
            "output_file": ACTUAL_CREDENTIAL_SOURCE_EXECUTABLE_OUTPUT_FILE,
        }
        ACTUAL_CREDENTIAL_SOURCE = {"executable": ACTUAL_CREDENTIAL_SOURCE_EXECUTABLE}
        with open(ACTUAL_CREDENTIAL_SOURCE_EXECUTABLE_OUTPUT_FILE, "w") as output_file:
            json.dump(self.EXECUTABLE_SUCCESSFUL_OIDC_RESPONSE_ID_TOKEN, output_file)

        credentials = self.make_pluggable(credential_source=ACTUAL_CREDENTIAL_SOURCE)

        subject_token = credentials.retrieve_subject_token(None)
        assert subject_token == self.EXECUTABLE_OIDC_TOKEN

        os.remove(ACTUAL_CREDENTIAL_SOURCE_EXECUTABLE_OUTPUT_FILE)

    @mock.patch.dict(os.environ, {"GOOGLE_EXTERNAL_ACCOUNT_ALLOW_EXECUTABLES": "1"})
    def test_retrieve_subject_token_no_file_cache(self):
        ACTUAL_CREDENTIAL_SOURCE_EXECUTABLE = {
            "command": "command",
            "timeout_millis": 30000,
        }
        ACTUAL_CREDENTIAL_SOURCE = {"executable": ACTUAL_CREDENTIAL_SOURCE_EXECUTABLE}

        with mock.patch(
            "subprocess.run",
            return_value=subprocess.CompletedProcess(
                args=[],
                stdout=json.dumps(
                    self.EXECUTABLE_SUCCESSFUL_OIDC_RESPONSE_ID_TOKEN
                ).encode("UTF-8"),
                returncode=0,
            ),
        ):
            credentials = self.make_pluggable(
                credential_source=ACTUAL_CREDENTIAL_SOURCE
            )

            subject_token = credentials.retrieve_subject_token(None)

            assert subject_token == self.EXECUTABLE_OIDC_TOKEN

    @mock.patch.dict(os.environ, {"GOOGLE_EXTERNAL_ACCOUNT_ALLOW_EXECUTABLES": "1"})
    def test_retrieve_subject_token_file_cache_value_error_report(self, tmpdir):
        ACTUAL_CREDENTIAL_SOURCE_EXECUTABLE_OUTPUT_FILE = tmpdir.join(
            "actual_output_file"
        )
        ACTUAL_CREDENTIAL_SOURCE_EXECUTABLE = {
            "command": "command",
            "timeout_millis": 30000,
            "output_file": ACTUAL_CREDENTIAL_SOURCE_EXECUTABLE_OUTPUT_FILE,
        }
        ACTUAL_CREDENTIAL_SOURCE = {"executable": ACTUAL_CREDENTIAL_SOURCE_EXECUTABLE}
        ACTUAL_EXECUTABLE_RESPONSE = {
            "success": True,
            "token_type": "urn:ietf:params:oauth:token-type:id_token",
            "id_token": self.EXECUTABLE_OIDC_TOKEN,
            "expiration_time": 9999999999,
        }
        with open(ACTUAL_CREDENTIAL_SOURCE_EXECUTABLE_OUTPUT_FILE, "w") as output_file:
            json.dump(ACTUAL_EXECUTABLE_RESPONSE, output_file)

        credentials = self.make_pluggable(credential_source=ACTUAL_CREDENTIAL_SOURCE)

        with pytest.raises(ValueError) as excinfo:
            _ = credentials.retrieve_subject_token(None)

        assert excinfo.match(r"The executable response is missing the version field.")

        os.remove(ACTUAL_CREDENTIAL_SOURCE_EXECUTABLE_OUTPUT_FILE)

    @mock.patch.dict(os.environ, {"GOOGLE_EXTERNAL_ACCOUNT_ALLOW_EXECUTABLES": "1"})
    def test_retrieve_subject_token_file_cache_refresh_error_retry(self, tmpdir):
        ACTUAL_CREDENTIAL_SOURCE_EXECUTABLE_OUTPUT_FILE = tmpdir.join(
            "actual_output_file"
        )
        ACTUAL_CREDENTIAL_SOURCE_EXECUTABLE = {
            "command": "command",
            "timeout_millis": 30000,
            "output_file": ACTUAL_CREDENTIAL_SOURCE_EXECUTABLE_OUTPUT_FILE,
        }
        ACTUAL_CREDENTIAL_SOURCE = {"executable": ACTUAL_CREDENTIAL_SOURCE_EXECUTABLE}
        ACTUAL_EXECUTABLE_RESPONSE = {
            "version": 2,
            "success": True,
            "token_type": "urn:ietf:params:oauth:token-type:id_token",
            "id_token": self.EXECUTABLE_OIDC_TOKEN,
            "expiration_time": 9999999999,
        }
        with open(ACTUAL_CREDENTIAL_SOURCE_EXECUTABLE_OUTPUT_FILE, "w") as output_file:
            json.dump(ACTUAL_EXECUTABLE_RESPONSE, output_file)

        with mock.patch(
            "subprocess.run",
            return_value=subprocess.CompletedProcess(
                args=[],
                stdout=json.dumps(
                    self.EXECUTABLE_SUCCESSFUL_OIDC_RESPONSE_ID_TOKEN
                ).encode("UTF-8"),
                returncode=0,
            ),
        ):
            credentials = self.make_pluggable(
                credential_source=ACTUAL_CREDENTIAL_SOURCE
            )

            subject_token = credentials.retrieve_subject_token(None)

            assert subject_token == self.EXECUTABLE_OIDC_TOKEN

        os.remove(ACTUAL_CREDENTIAL_SOURCE_EXECUTABLE_OUTPUT_FILE)

    @mock.patch.dict(os.environ, {"GOOGLE_EXTERNAL_ACCOUNT_ALLOW_EXECUTABLES": "1"})
    def test_retrieve_subject_token_unsupported_token_type(self):
        EXECUTABLE_SUCCESSFUL_OIDC_RESPONSE = {
            "version": 1,
            "success": True,
            "token_type": "unsupported_token_type",
            "id_token": self.EXECUTABLE_OIDC_TOKEN,
            "expiration_time": 9999999999,
        }

        with mock.patch(
            "subprocess.run",
            return_value=subprocess.CompletedProcess(
                args=[],
                stdout=json.dumps(EXECUTABLE_SUCCESSFUL_OIDC_RESPONSE).encode("UTF-8"),
                returncode=0,
            ),
        ):
            credentials = self.make_pluggable(credential_source=self.CREDENTIAL_SOURCE)

            with pytest.raises(exceptions.RefreshError) as excinfo:
                _ = credentials.retrieve_subject_token(None)

            assert excinfo.match(r"Executable returned unsupported token type.")

    @mock.patch.dict(os.environ, {"GOOGLE_EXTERNAL_ACCOUNT_ALLOW_EXECUTABLES": "1"})
    def test_retrieve_subject_token_missing_version(self):
        EXECUTABLE_SUCCESSFUL_OIDC_RESPONSE = {
            "success": True,
            "token_type": "urn:ietf:params:oauth:token-type:id_token",
            "id_token": self.EXECUTABLE_OIDC_TOKEN,
            "expiration_time": 9999999999,
        }

        with mock.patch(
            "subprocess.run",
            return_value=subprocess.CompletedProcess(
                args=[],
                stdout=json.dumps(EXECUTABLE_SUCCESSFUL_OIDC_RESPONSE).encode("UTF-8"),
                returncode=0,
            ),
        ):
            credentials = self.make_pluggable(credential_source=self.CREDENTIAL_SOURCE)

            with pytest.raises(ValueError) as excinfo:
                _ = credentials.retrieve_subject_token(None)

            assert excinfo.match(
                r"The executable response is missing the version field."
            )

    @mock.patch.dict(os.environ, {"GOOGLE_EXTERNAL_ACCOUNT_ALLOW_EXECUTABLES": "1"})
    def test_retrieve_subject_token_missing_success(self):
        EXECUTABLE_SUCCESSFUL_OIDC_RESPONSE = {
            "version": 1,
            "token_type": "urn:ietf:params:oauth:token-type:id_token",
            "id_token": self.EXECUTABLE_OIDC_TOKEN,
            "expiration_time": 9999999999,
        }

        with mock.patch(
            "subprocess.run",
            return_value=subprocess.CompletedProcess(
                args=[],
                stdout=json.dumps(EXECUTABLE_SUCCESSFUL_OIDC_RESPONSE).encode("UTF-8"),
                returncode=0,
            ),
        ):
            credentials = self.make_pluggable(credential_source=self.CREDENTIAL_SOURCE)

            with pytest.raises(ValueError) as excinfo:
                _ = credentials.retrieve_subject_token(None)

            assert excinfo.match(
                r"The executable response is missing the success field."
            )

    @mock.patch.dict(os.environ, {"GOOGLE_EXTERNAL_ACCOUNT_ALLOW_EXECUTABLES": "1"})
    def test_retrieve_subject_token_missing_error_code_message(self):
        EXECUTABLE_SUCCESSFUL_OIDC_RESPONSE = {"version": 1, "success": False}

        with mock.patch(
            "subprocess.run",
            return_value=subprocess.CompletedProcess(
                args=[],
                stdout=json.dumps(EXECUTABLE_SUCCESSFUL_OIDC_RESPONSE).encode("UTF-8"),
                returncode=0,
            ),
        ):
            credentials = self.make_pluggable(credential_source=self.CREDENTIAL_SOURCE)

            with pytest.raises(ValueError) as excinfo:
                _ = credentials.retrieve_subject_token(None)

            assert excinfo.match(
                r"Error code and message fields are required in the response."
            )

    @mock.patch.dict(os.environ, {"GOOGLE_EXTERNAL_ACCOUNT_ALLOW_EXECUTABLES": "1"})
    def test_retrieve_subject_token_without_expiration_time_should_pass_when_output_file_not_specified(
        self,
    ):
        EXECUTABLE_SUCCESSFUL_OIDC_RESPONSE = {
            "version": 1,
            "success": True,
            "token_type": "urn:ietf:params:oauth:token-type:id_token",
            "id_token": self.EXECUTABLE_OIDC_TOKEN,
        }

        CREDENTIAL_SOURCE = {
            "executable": {"command": "command", "timeout_millis": 30000}
        }

        with mock.patch(
            "subprocess.run",
            return_value=subprocess.CompletedProcess(
                args=[],
                stdout=json.dumps(EXECUTABLE_SUCCESSFUL_OIDC_RESPONSE).encode("UTF-8"),
                returncode=0,
            ),
        ):
            credentials = self.make_pluggable(credential_source=CREDENTIAL_SOURCE)
            subject_token = credentials.retrieve_subject_token(None)

            assert subject_token == self.EXECUTABLE_OIDC_TOKEN

    @mock.patch.dict(os.environ, {"GOOGLE_EXTERNAL_ACCOUNT_ALLOW_EXECUTABLES": "1"})
    def test_retrieve_subject_token_missing_token_type(self):
        EXECUTABLE_SUCCESSFUL_OIDC_RESPONSE = {
            "version": 1,
            "success": True,
            "id_token": self.EXECUTABLE_OIDC_TOKEN,
            "expiration_time": 9999999999,
        }

        with mock.patch(
            "subprocess.run",
            return_value=subprocess.CompletedProcess(
                args=[],
                stdout=json.dumps(EXECUTABLE_SUCCESSFUL_OIDC_RESPONSE).encode("UTF-8"),
                returncode=0,
            ),
        ):
            credentials = self.make_pluggable(credential_source=self.CREDENTIAL_SOURCE)

            with pytest.raises(ValueError) as excinfo:
                _ = credentials.retrieve_subject_token(None)

            assert excinfo.match(
                r"The executable response is missing the token_type field."
            )

    @mock.patch.dict(os.environ, {"GOOGLE_EXTERNAL_ACCOUNT_ALLOW_EXECUTABLES": "1"})
    def test_credential_source_missing_command(self):
        with pytest.raises(ValueError) as excinfo:
            CREDENTIAL_SOURCE = {
                "executable": {
                    "timeout_millis": 30000,
                    "output_file": self.CREDENTIAL_SOURCE_EXECUTABLE_OUTPUT_FILE,
                }
            }
            _ = self.make_pluggable(credential_source=CREDENTIAL_SOURCE)

        assert excinfo.match(
            r"Missing command field. Executable command must be provided."
        )

    @mock.patch.dict(os.environ, {"GOOGLE_EXTERNAL_ACCOUNT_ALLOW_EXECUTABLES": "1"})
    def test_credential_source_missing_output_interactive_mode(self):
        CREDENTIAL_SOURCE = {
            "executable": {"command": self.CREDENTIAL_SOURCE_EXECUTABLE_COMMAND}
        }
        credentials = self.make_pluggable(
            credential_source=CREDENTIAL_SOURCE, interactive=True
        )
        with pytest.raises(ValueError) as excinfo:
            _ = credentials.retrieve_subject_token(None)

        assert excinfo.match(
            r"An output_file must be specified in the credential configuration for interactive mode."
        )

    @mock.patch.dict(os.environ, {"GOOGLE_EXTERNAL_ACCOUNT_ALLOW_EXECUTABLES": "1"})
    def test_credential_source_timeout_missing_will_use_default_timeout_value(self):
        CREDENTIAL_SOURCE = {
            "executable": {
                "command": self.CREDENTIAL_SOURCE_EXECUTABLE_COMMAND,
                "output_file": self.CREDENTIAL_SOURCE_EXECUTABLE_OUTPUT_FILE,
            }
        }
        credentials = self.make_pluggable(credential_source=CREDENTIAL_SOURCE)

        assert (
            credentials._credential_source_executable_timeout_millis
            == pluggable.EXECUTABLE_TIMEOUT_MILLIS_DEFAULT
        )

    @mock.patch.dict(os.environ, {"GOOGLE_EXTERNAL_ACCOUNT_ALLOW_EXECUTABLES": "1"})
    def test_credential_source_timeout_small(self):
        with pytest.raises(ValueError) as excinfo:
            CREDENTIAL_SOURCE = {
                "executable": {
                    "command": self.CREDENTIAL_SOURCE_EXECUTABLE_COMMAND,
                    "timeout_millis": 5000 - 1,
                    "output_file": self.CREDENTIAL_SOURCE_EXECUTABLE_OUTPUT_FILE,
                }
            }
            _ = self.make_pluggable(credential_source=CREDENTIAL_SOURCE)

        assert excinfo.match(r"Timeout must be between 5 and 120 seconds.")

    @mock.patch.dict(os.environ, {"GOOGLE_EXTERNAL_ACCOUNT_ALLOW_EXECUTABLES": "1"})
    def test_credential_source_timeout_large(self):
        with pytest.raises(ValueError) as excinfo:
            CREDENTIAL_SOURCE = {
                "executable": {
                    "command": self.CREDENTIAL_SOURCE_EXECUTABLE_COMMAND,
                    "timeout_millis": 120000 + 1,
                    "output_file": self.CREDENTIAL_SOURCE_EXECUTABLE_OUTPUT_FILE,
                }
            }
            _ = self.make_pluggable(credential_source=CREDENTIAL_SOURCE)

        assert excinfo.match(r"Timeout must be between 5 and 120 seconds.")

    @mock.patch.dict(os.environ, {"GOOGLE_EXTERNAL_ACCOUNT_ALLOW_EXECUTABLES": "1"})
    def test_credential_source_interactive_timeout_small(self):
        with pytest.raises(ValueError) as excinfo:
            CREDENTIAL_SOURCE = {
                "executable": {
                    "command": self.CREDENTIAL_SOURCE_EXECUTABLE_COMMAND,
                    "interactive_timeout_millis": 30000 - 1,
                    "output_file": self.CREDENTIAL_SOURCE_EXECUTABLE_OUTPUT_FILE,
                }
            }
            _ = self.make_pluggable(credential_source=CREDENTIAL_SOURCE)

        assert excinfo.match(
            r"Interactive timeout must be between 30 seconds and 30 minutes."
        )

    @mock.patch.dict(os.environ, {"GOOGLE_EXTERNAL_ACCOUNT_ALLOW_EXECUTABLES": "1"})
    def test_credential_source_interactive_timeout_large(self):
        with pytest.raises(ValueError) as excinfo:
            CREDENTIAL_SOURCE = {
                "executable": {
                    "command": self.CREDENTIAL_SOURCE_EXECUTABLE_COMMAND,
                    "interactive_timeout_millis": 1800000 + 1,
                    "output_file": self.CREDENTIAL_SOURCE_EXECUTABLE_OUTPUT_FILE,
                }
            }
            _ = self.make_pluggable(credential_source=CREDENTIAL_SOURCE)

        assert excinfo.match(
            r"Interactive timeout must be between 30 seconds and 30 minutes."
        )

    @mock.patch.dict(os.environ, {"GOOGLE_EXTERNAL_ACCOUNT_ALLOW_EXECUTABLES": "1"})
    def test_retrieve_subject_token_executable_fail(self):
        with mock.patch(
            "subprocess.run",
            return_value=subprocess.CompletedProcess(
                args=[], stdout=None, returncode=1
            ),
        ):
            credentials = self.make_pluggable(credential_source=self.CREDENTIAL_SOURCE)

            with pytest.raises(exceptions.RefreshError) as excinfo:
                _ = credentials.retrieve_subject_token(None)

            assert excinfo.match(
                r"Executable exited with non-zero return code 1. Error: None"
            )

    @mock.patch.dict(os.environ, {"GOOGLE_EXTERNAL_ACCOUNT_ALLOW_EXECUTABLES": "1"})
    def test_retrieve_subject_token_non_workforce_fail_interactive_mode(self):
        credentials = self.make_pluggable(
            credential_source=self.CREDENTIAL_SOURCE, interactive=True
        )
        with pytest.raises(ValueError) as excinfo:
            _ = credentials.retrieve_subject_token(None)

        assert excinfo.match(r"Interactive mode is only enabled for workforce pool.")

    @mock.patch.dict(os.environ, {"GOOGLE_EXTERNAL_ACCOUNT_ALLOW_EXECUTABLES": "1"})
    def test_retrieve_subject_token_fail_on_validation_missing_interactive_timeout(
        self
    ):
        CREDENTIAL_SOURCE_EXECUTABLE = {
            "command": self.CREDENTIAL_SOURCE_EXECUTABLE_COMMAND,
            "output_file": self.CREDENTIAL_SOURCE_EXECUTABLE_OUTPUT_FILE,
        }
        CREDENTIAL_SOURCE = {"executable": CREDENTIAL_SOURCE_EXECUTABLE}
        credentials = self.make_pluggable(
            credential_source=CREDENTIAL_SOURCE, interactive=True
        )
        with pytest.raises(ValueError) as excinfo:
            _ = credentials.retrieve_subject_token(None)

        assert excinfo.match(
            r"Interactive mode cannot run without an interactive timeout."
        )

    @mock.patch.dict(os.environ, {"GOOGLE_EXTERNAL_ACCOUNT_ALLOW_EXECUTABLES": "1"})
    def test_retrieve_subject_token_executable_fail_interactive_mode(self):
        with mock.patch(
            "subprocess.run",
            return_value=subprocess.CompletedProcess(
                args=[], stdout=None, returncode=1
            ),
        ):
            credentials = self.make_pluggable(
                audience=WORKFORCE_AUDIENCE,
                credential_source=self.CREDENTIAL_SOURCE,
                interactive=True,
            )

            with pytest.raises(exceptions.RefreshError) as excinfo:
                _ = credentials.retrieve_subject_token(None)

            assert excinfo.match(
                r"Executable exited with non-zero return code 1. Error: None"
            )

    @mock.patch.dict(os.environ, {"GOOGLE_EXTERNAL_ACCOUNT_ALLOW_EXECUTABLES": "0"})
    def test_revoke_failed_executable_not_allowed(self):
        credentials = self.make_pluggable(
            credential_source=self.CREDENTIAL_SOURCE, interactive=True
        )
        with pytest.raises(ValueError) as excinfo:
            _ = credentials.revoke(None)

        assert excinfo.match(r"Executables need to be explicitly allowed")

    @mock.patch.dict(os.environ, {"GOOGLE_EXTERNAL_ACCOUNT_ALLOW_EXECUTABLES": "1"})
    def test_revoke_failed(self):
        testData = {
            "non_interactive_mode": {
                "interactive": False,
                "expectErrType": ValueError,
                "expectErrPattern": r"Revoke is only enabled under interactive mode.",
            },
            "executable_failed": {
                "returncode": 1,
                "expectErrType": exceptions.RefreshError,
                "expectErrPattern": r"Auth revoke failed on executable.",
            },
            "response_validation_missing_version": {
                "response": {},
                "expectErrType": ValueError,
                "expectErrPattern": r"The executable response is missing the version field.",
            },
            "response_validation_invalid_version": {
                "response": {"version": 2},
                "expectErrType": exceptions.RefreshError,
                "expectErrPattern": r"Executable returned unsupported version.",
            },
            "response_validation_missing_success": {
                "response": {"version": 1},
                "expectErrType": ValueError,
                "expectErrPattern": r"The executable response is missing the success field.",
            },
            "response_validation_failed_with_success_field_is_false": {
                "response": {"version": 1, "success": False},
                "expectErrType": exceptions.RefreshError,
                "expectErrPattern": r"Revoke failed with unsuccessful response.",
            },
        }
        for data in testData.values():
            with mock.patch(
                "subprocess.run",
                return_value=subprocess.CompletedProcess(
                    args=[],
                    stdout=json.dumps(data.get("response")).encode("UTF-8"),
                    returncode=data.get("returncode", 0),
                ),
            ):
                credentials = self.make_pluggable(
                    audience=WORKFORCE_AUDIENCE,
                    service_account_impersonation_url=SERVICE_ACCOUNT_IMPERSONATION_URL,
                    credential_source=self.CREDENTIAL_SOURCE,
                    interactive=data.get("interactive", True),
                )

                with pytest.raises(data.get("expectErrType")) as excinfo:
                    _ = credentials.revoke(None)

                assert excinfo.match(data.get("expectErrPattern"))

    @mock.patch.dict(os.environ, {"GOOGLE_EXTERNAL_ACCOUNT_ALLOW_EXECUTABLES": "1"})
    def test_revoke_successfully(self):
        ACTUAL_RESPONSE = {"version": 1, "success": True}
        with mock.patch(
            "subprocess.run",
            return_value=subprocess.CompletedProcess(
                args=[],
                stdout=json.dumps(ACTUAL_RESPONSE).encode("utf-8"),
                returncode=0,
            ),
        ):
            credentials = self.make_pluggable(
                audience=WORKFORCE_AUDIENCE,
                credential_source=self.CREDENTIAL_SOURCE,
                interactive=True,
            )
            _ = credentials.revoke(None)

    @mock.patch.dict(os.environ, {"GOOGLE_EXTERNAL_ACCOUNT_ALLOW_EXECUTABLES": "1"})
    def test_retrieve_subject_token_python_2(self):
        with mock.patch("sys.version_info", (2, 7)):
            credentials = self.make_pluggable(credential_source=self.CREDENTIAL_SOURCE)

            with pytest.raises(exceptions.RefreshError) as excinfo:
                _ = credentials.retrieve_subject_token(None)

            assert excinfo.match(r"Pluggable auth is only supported for python 3.7+")

    @mock.patch.dict(os.environ, {"GOOGLE_EXTERNAL_ACCOUNT_ALLOW_EXECUTABLES": "1"})
    def test_revoke_subject_token_python_2(self):
        with mock.patch("sys.version_info", (2, 7)):
            credentials = self.make_pluggable(
                audience=WORKFORCE_AUDIENCE,
                credential_source=self.CREDENTIAL_SOURCE,
                interactive=True,
            )

            with pytest.raises(exceptions.RefreshError) as excinfo:
                _ = credentials.revoke(None)

            assert excinfo.match(r"Pluggable auth is only supported for python 3.7+")