"""Test that URIs are the same, up to reordering of query parameters."""

expected = urllib.parse.urlparse(expected)

actual = urllib.parse.urlparse(actual)

testcase.assertEqual(expected.scheme, actual.scheme)

testcase.assertEqual(expected.netloc, actual.netloc)

testcase.assertEqual(expected.path, actual.path)

testcase.assertEqual(expected.params, actual.params)

testcase.assertEqual(expected.fragment, actual.fragment)

expected_query = urllib.parse.parse_qs(expected.query)

actual_query = urllib.parse.parse_qs(actual.query)

for name in list(expected_query.keys()):

testcase.assertEqual(expected_query[name], actual_query[name])

for name in list(actual_query.keys()):

"""Assert that discovery URI used was the one that was expected

params = {"api": service_name, "apiVersion": version}

expanded_requested_uri = uritemplate.expand(discovery, params)

"""Validates that there have > 0 calls to Http Discovery

testcase.assertTrue(len(http_mock.request_sequence) > 0)

if len(http_mock.request_sequence) > 0:

actual_uri = http_mock.request_sequence[-1][0]

"""Safely converts a semantic version string to a comparable tuple of integers.

parts = []

for part in version_string.split("."):

try:

parts.append(int(part))

except ValueError:

# If it's a non-numeric part (e.g., '1.0.0b1' -> 'b1'), stop here.

# This is a simplification compared to 'packaging.parse_version', but sufficient

# for comparing strictly numeric semantic versions.

break

def setUp(self):

self.zoo_root_desc = json.loads(read_datafile("zoo.json", "r"))

self.zoo_get_method_desc = self.zoo_root_desc["methods"]["query"]

self.zoo_animals_resource = self.zoo_root_desc["resources"]["animals"]

self.zoo_insert_method_desc = self.zoo_animals_resource["methods"]["insert"]

self.zoo_schema = Schemas(self.zoo_root_desc)

def test_key2param(self):

self.assertEqual("max_results", key2param("max-results"))

self.assertEqual("x007_bond", key2param("007-bond"))

def _base_fix_up_parameters_test(self, method_desc, http_method, root_desc, schema):

self.assertEqual(method_desc["httpMethod"], http_method)

method_desc_copy = copy.deepcopy(method_desc)

self.assertEqual(method_desc, method_desc_copy)

parameters = _fix_up_parameters(

method_desc_copy, root_desc, http_method, schema

)

self.assertNotEqual(method_desc, method_desc_copy)

for param_name in STACK_QUERY_PARAMETERS:

self.assertEqual(

STACK_QUERY_PARAMETER_DEFAULT_VALUE, parameters[param_name]

)

for param_name, value in root_desc.get("parameters", {}).items():

self.assertEqual(value, parameters[param_name])

return parameters

def test_fix_up_parameters_get(self):

parameters = self._base_fix_up_parameters_test(

self.zoo_get_method_desc, "GET", self.zoo_root_desc, self.zoo_schema

)

# Since http_method is 'GET'

self.assertFalse("body" in parameters)

def test_fix_up_parameters_insert(self):

parameters = self._base_fix_up_parameters_test(

self.zoo_insert_method_desc, "POST", self.zoo_root_desc, self.zoo_schema

)

body = {"description": "The request body.", "type": "object", "$ref": "Animal"}

self.assertEqual(parameters["body"], body)

def test_fix_up_parameters_check_body(self):

dummy_root_desc = {}

dummy_schema = {

"Request": {

"properties": {

"description": "Required. Dummy parameter.",

"type": "string",

}

}

}

no_payload_http_method = "DELETE"

with_payload_http_method = "PUT"

invalid_method_desc = {"response": "Who cares"}

valid_method_desc = {

"request": {"key1": "value1", "key2": "value2", "$ref": "Request"}

}

parameters = _fix_up_parameters(

invalid_method_desc, dummy_root_desc, no_payload_http_method, dummy_schema

)

self.assertFalse("body" in parameters)

parameters = _fix_up_parameters(

valid_method_desc, dummy_root_desc, no_payload_http_method, dummy_schema

)

self.assertFalse("body" in parameters)

parameters = _fix_up_parameters(

invalid_method_desc, dummy_root_desc, with_payload_http_method, dummy_schema

)

self.assertFalse("body" in parameters)

parameters = _fix_up_parameters(

valid_method_desc, dummy_root_desc, with_payload_http_method, dummy_schema

)

body = {

"description": "The request body.",

"type": "object",

"$ref": "Request",

"key1": "value1",

"key2": "value2",

}

self.assertEqual(parameters["body"], body)

def test_fix_up_parameters_optional_body(self):

# Request with no parameters

dummy_schema = {"Request": {"properties": {}}}

method_desc = {"request": {"$ref": "Request"}}

parameters = _fix_up_parameters(method_desc, {}, "POST", dummy_schema)

def _base_fix_up_method_description_test(

self,

method_desc,

initial_parameters,

final_parameters,

final_accept,

final_max_size,

final_media_path_url,

):

fake_root_desc = {

"rootUrl": "http://root/",

"servicePath": "fake/",

"mtlsRootUrl": "http://root/",

}

fake_path_url = "fake-path/"

accept, max_size, media_path_url = _fix_up_media_upload(

method_desc, fake_root_desc, fake_path_url, initial_parameters

)

self.assertEqual(accept, final_accept)

self.assertEqual(max_size, final_max_size)

self.assertEqual(media_path_url, final_media_path_url)

self.assertEqual(initial_parameters, final_parameters)

def test_fix_up_media_upload_no_initial_invalid(self):

invalid_method_desc = {"response": "Who cares"}

self._base_fix_up_method_description_test(

invalid_method_desc, {}, {}, [], 0, None

)

def test_fix_up_media_upload_no_initial_valid_minimal(self):

valid_method_desc = {"mediaUpload": {"accept": []}}

final_parameters = {

"media_body": MEDIA_BODY_PARAMETER_DEFAULT_VALUE,

"media_mime_type": MEDIA_MIME_TYPE_PARAMETER_DEFAULT_VALUE,

}

self._base_fix_up_method_description_test(

valid_method_desc,

{},

final_parameters,

[],

0,

"http://root/upload/fake/fake-path/",

)

def test_fix_up_media_upload_no_initial_valid_full(self):

valid_method_desc = {"mediaUpload": {"accept": ["*/*"], "maxSize": "10GB"}}

final_parameters = {

"media_body": MEDIA_BODY_PARAMETER_DEFAULT_VALUE,

"media_mime_type": MEDIA_MIME_TYPE_PARAMETER_DEFAULT_VALUE,

}

ten_gb = 10 * 2**30

self._base_fix_up_method_description_test(

valid_method_desc,

{},

final_parameters,

["*/*"],

ten_gb,

"http://root/upload/fake/fake-path/",

)

def test_fix_up_media_upload_with_initial_invalid(self):

invalid_method_desc = {"response": "Who cares"}

initial_parameters = {"body": {}}

self._base_fix_up_method_description_test(

invalid_method_desc, initial_parameters, initial_parameters, [], 0, None

)

def test_fix_up_media_upload_with_initial_valid_minimal(self):

valid_method_desc = {"mediaUpload": {"accept": []}}

initial_parameters = {"body": {}}

final_parameters = {

"body": {},

"media_body": MEDIA_BODY_PARAMETER_DEFAULT_VALUE,

"media_mime_type": MEDIA_MIME_TYPE_PARAMETER_DEFAULT_VALUE,

}

self._base_fix_up_method_description_test(

valid_method_desc,

initial_parameters,

final_parameters,

[],

0,

"http://root/upload/fake/fake-path/",

)

def test_fix_up_media_upload_with_initial_valid_full(self):

valid_method_desc = {"mediaUpload": {"accept": ["*/*"], "maxSize": "10GB"}}

initial_parameters = {"body": {}}

final_parameters = {

"body": {},

"media_body": MEDIA_BODY_PARAMETER_DEFAULT_VALUE,

"media_mime_type": MEDIA_MIME_TYPE_PARAMETER_DEFAULT_VALUE,

}

ten_gb = 10 * 2**30

self._base_fix_up_method_description_test(

valid_method_desc,

initial_parameters,

final_parameters,

["*/*"],

ten_gb,

"http://root/upload/fake/fake-path/",

)

def test_fix_up_method_description_get(self):

result = _fix_up_method_description(

self.zoo_get_method_desc, self.zoo_root_desc, self.zoo_schema

)

path_url = "query"

http_method = "GET"

method_id = "bigquery.query"

accept = []

max_size = 0

media_path_url = None

self.assertEqual(

result, (path_url, http_method, method_id, accept, max_size, media_path_url)

)

def test_fix_up_method_description_insert(self):

result = _fix_up_method_description(

self.zoo_insert_method_desc, self.zoo_root_desc, self.zoo_schema

)

path_url = "animals"

http_method = "POST"

method_id = "zoo.animals.insert"

accept = ["image/png"]

max_size = 1024

media_path_url = "https://www.googleapis.com/upload/zoo/v1/animals"

self.assertEqual(

result, (path_url, http_method, method_id, accept, max_size, media_path_url)

)

def test_fix_up_media_path_base_url_same_netloc(self):

result = _fix_up_media_path_base_url(

"https://www.googleapis.com/upload/foo",

"https://www.googleapis.com/upload/bar",

)

self.assertEqual(result, "https://www.googleapis.com/upload/foo")

def test_fix_up_media_path_base_url_different_netloc(self):

result = _fix_up_media_path_base_url(

"https://www.googleapis.com/upload/foo",

"https://www.example.com/upload/bar",

)

self.assertEqual(result, "https://www.example.com/upload/foo")

def test_urljoin(self):

# We want to exhaustively test various URL combinations.

simple_bases = ["https://www.googleapis.com", "https://www.googleapis.com/"]

long_urls = ["foo/v1/bar:custom?alt=json", "/foo/v1/bar:custom?alt=json"]

long_bases = [

"https://www.googleapis.com/foo/v1",

"https://www.googleapis.com/foo/v1/",

]

simple_urls = ["bar:custom?alt=json", "/bar:custom?alt=json"]

final_url = "https://www.googleapis.com/foo/v1/bar:custom?alt=json"

for base, url in itertools.product(simple_bases, long_urls):

self.assertEqual(final_url, _urljoin(base, url))

for base, url in itertools.product(long_bases, simple_urls):

self.assertEqual(final_url, _urljoin(base, url))

def test_ResourceMethodParameters_zoo_get(self):

parameters = ResourceMethodParameters(self.zoo_get_method_desc)

param_types = {

"a": "any",

"b": "boolean",

"e": "string",

"er": "string",

"i": "integer",

"n": "number",

"o": "object",

"q": "string",

"rr": "string",

}

keys = list(param_types.keys())

self.assertEqual(parameters.argmap, dict((key, key) for key in keys))

self.assertEqual(parameters.required_params, [])

self.assertEqual(sorted(parameters.repeated_params), ["er", "rr"])

self.assertEqual(parameters.pattern_params, {"rr": "[a-z]+"})

self.assertEqual(

sorted(parameters.query_params),

["a", "b", "e", "er", "i", "n", "o", "q", "rr"],

)

self.assertEqual(parameters.path_params, set())

self.assertEqual(parameters.param_types, param_types)

enum_params = {"e": ["foo", "bar"], "er": ["one", "two", "three"]}

self.assertEqual(parameters.enum_params, enum_params)

def test_ResourceMethodParameters_zoo_animals_patch(self):

method_desc = self.zoo_animals_resource["methods"]["patch"]

parameters = ResourceMethodParameters(method_desc)

param_types = {"name": "string"}

keys = list(param_types.keys())

self.assertEqual(parameters.argmap, dict((key, key) for key in keys))

self.assertEqual(parameters.required_params, ["name"])

self.assertEqual(parameters.repeated_params, [])

self.assertEqual(parameters.pattern_params, {})

self.assertEqual(parameters.query_params, [])

self.assertEqual(parameters.path_params, set(["name"]))

self.assertEqual(parameters.param_types, param_types)

def test_discovery_http_is_closed(self):

http = HttpMock(datafile("malformed.json"), {"status": "200"})

service = build("plus", "v1", credentials=mock.sentinel.credentials)

def test_tests_should_be_run_with_strict_positional_enforcement(self):

try:

plus = build("plus", "v1", None, static_discovery=False)

self.fail("should have raised a TypeError exception over missing http=.")

except TypeError:

pass

def test_failed_to_parse_discovery_json(self):

self.http = HttpMock(datafile("malformed.json"), {"status": "200"})

try:

plus = build(

"plus",

"v1",

http=self.http,

cache_discovery=False,

static_discovery=False,

)

self.fail("should have raised an exception over malformed JSON.")

except InvalidJsonError:

pass

def test_unknown_api_name_or_version(self):

http = HttpMockSequence(

[

({"status": "404"}, read_datafile("zoo.json", "rb")),

({"status": "404"}, read_datafile("zoo.json", "rb")),

]

)

with self.assertRaises(UnknownApiNameOrVersion):

plus = build("plus", "v1", http=http, cache_discovery=False)

def test_credentials_and_http_mutually_exclusive(self):

http = HttpMock(datafile("plus.json"), {"status": "200"})

with self.assertRaises(ValueError):

build(

"plus",

"v1",

http=http,

credentials=mock.sentinel.credentials,

static_discovery=False,

)

def test_credentials_file_and_http_mutually_exclusive(self):

http = HttpMock(datafile("plus.json"), {"status": "200"})

with self.assertRaises(ValueError):

build(

"plus",

"v1",

http=http,

client_options=google.api_core.client_options.ClientOptions(

credentials_file="credentials.json"

),

static_discovery=False,

)

def test_credentials_and_credentials_file_mutually_exclusive(self):

with self.assertRaises(google.api_core.exceptions.DuplicateCredentialArgs):

build(

"plus",

"v1",

credentials=mock.sentinel.credentials,

client_options=google.api_core.client_options.ClientOptions(

credentials_file="credentials.json"

),

static_discovery=False,

MOCK_CREDENTIALS = mock.Mock(spec=google.auth.credentials.Credentials)

_reset_universe_domain(MOCK_CREDENTIALS)

def test_can_build_from_local_document(self):

discovery = read_datafile("plus.json")

plus = build_from_document(

discovery,

base="https://www.googleapis.com/",

credentials=self.MOCK_CREDENTIALS,

)

self.assertIsNotNone(plus)

self.assertTrue(hasattr(plus, "activities"))

def test_can_build_from_local_deserialized_document(self):

discovery = read_datafile("plus.json")

discovery = json.loads(discovery)

plus = build_from_document(

discovery,

base="https://www.googleapis.com/",

credentials=self.MOCK_CREDENTIALS,

)

self.assertIsNotNone(plus)

self.assertTrue(hasattr(plus, "activities"))

def test_building_with_base_remembers_base(self):

discovery = read_datafile("plus.json")

base = "https://www.example.com/"

plus = build_from_document(

discovery, base=base, credentials=self.MOCK_CREDENTIALS

)

self.assertEqual("https://www.googleapis.com/plus/v1/", plus._baseUrl)

def test_building_with_optional_http_with_authorization(self):

discovery = read_datafile("plus.json")

plus = build_from_document(

discovery,

base="https://www.googleapis.com/",

credentials=self.MOCK_CREDENTIALS,

)

# plus service requires Authorization, hence we expect to see AuthorizedHttp object here

self.assertIsInstance(plus._http, google_auth_httplib2.AuthorizedHttp)

self.assertIsInstance(plus._http.http, httplib2.Http)

self.assertIsInstance(plus._http.http.timeout, int)

self.assertGreater(plus._http.http.timeout, 0)

def test_building_with_optional_http_with_no_authorization(self):

discovery = read_datafile("plus.json")

# Cleanup auth field, so we would use plain http client

discovery = json.loads(discovery)

discovery["auth"] = {}

discovery = json.dumps(discovery)

plus = build_from_document(

discovery, base="https://www.googleapis.com/", credentials=None

)

# plus service requires Authorization

self.assertIsInstance(plus._http, httplib2.Http)

self.assertIsInstance(plus._http.timeout, int)

self.assertGreater(plus._http.timeout, 0)

def test_building_with_explicit_http(self):

http = HttpMock()

discovery = read_datafile("plus.json")

plus = build_from_document(

discovery, base="https://www.googleapis.com/", http=http

)

self.assertEqual(plus._http, http)

def test_building_with_developer_key_skips_adc(self):

discovery = read_datafile("plus.json")

plus = build_from_document(

discovery, base="https://www.googleapis.com/", developerKey="123"

)

self.assertIsInstance(plus._http, httplib2.Http)

# It should not be an AuthorizedHttp, because that would indicate that

# application default credentials were used.

self.assertNotIsInstance(plus._http, google_auth_httplib2.AuthorizedHttp)

def test_building_with_context_manager(self):

discovery = read_datafile("plus.json")

with mock.patch("httplib2.Http") as http:

with build_from_document(

discovery,

base="https://www.googleapis.com/",

credentials=self.MOCK_CREDENTIALS,

) as plus:

self.assertIsNotNone(plus)

self.assertTrue(hasattr(plus, "activities"))

plus._http.http.close.assert_called_once()

def test_resource_close(self):

discovery = read_datafile("plus.json")

with mock.patch("httplib2.Http", autospec=True) as httplib2_http:

http = httplib2_http()

plus = build_from_document(

discovery,

base="https://www.googleapis.com/",

http=http,

)

plus.close()

http.close.assert_called_once()

def test_resource_close_authorized_http(self):

discovery = read_datafile("plus.json")

with mock.patch("google_auth_httplib2.AuthorizedHttp", autospec=True):

plus = build_from_document(

discovery,

base="https://www.googleapis.com/",

credentials=self.MOCK_CREDENTIALS,

)

plus.close()

plus._http.close.assert_called_once()

def test_api_endpoint_override_from_client_options(self):

discovery = read_datafile("plus.json")

api_endpoint = "https://foo.googleapis.com/"

options = google.api_core.client_options.ClientOptions(

api_endpoint=api_endpoint

)

plus = build_from_document(

discovery, client_options=options, credentials=self.MOCK_CREDENTIALS

)

self.assertEqual(plus._baseUrl, api_endpoint)

def test_api_endpoint_override_from_client_options_mapping_object(self):

discovery = read_datafile("plus.json")

api_endpoint = "https://foo.googleapis.com/"

mapping_object = defaultdict(str)

mapping_object["api_endpoint"] = api_endpoint

plus = build_from_document(

discovery, client_options=mapping_object, credentials=self.MOCK_CREDENTIALS

)

self.assertEqual(plus._baseUrl, api_endpoint)

def test_api_endpoint_override_from_client_options_dict(self):

discovery = read_datafile("plus.json")

api_endpoint = "https://foo.googleapis.com/"

plus = build_from_document(

discovery,

client_options={"api_endpoint": api_endpoint},

credentials=self.MOCK_CREDENTIALS,

)

self.assertEqual(plus._baseUrl, api_endpoint)

def test_scopes_from_client_options(self):

discovery = read_datafile("plus.json")

with mock.patch("googleapiclient._auth.default_credentials") as default:

_reset_universe_domain(default.return_value)

plus = build_from_document(

discovery,

client_options={"scopes": ["1", "2"]},

)

default.assert_called_once_with(scopes=["1", "2"], quota_project_id=None)

def test_quota_project_from_client_options(self):

discovery = read_datafile("plus.json")

with mock.patch("googleapiclient._auth.default_credentials") as default:

_reset_universe_domain(default.return_value)

plus = build_from_document(

discovery,

client_options=google.api_core.client_options.ClientOptions(

quota_project_id="my-project"

),

)

default.assert_called_once_with(scopes=None, quota_project_id="my-project")

def test_credentials_file_from_client_options(self):

discovery = read_datafile("plus.json")

with mock.patch("googleapiclient._auth.credentials_from_file") as default:

_reset_universe_domain(default.return_value)

plus = build_from_document(

discovery,

client_options=google.api_core.client_options.ClientOptions(

credentials_file="credentials.json"

),

)

default.assert_called_once_with(

"credentials.json", scopes=None, quota_project_id=None

)

def test_self_signed_jwt_enabled(self):

service_account_file_path = os.path.join(DATA_DIR, "service_account.json")

creds = google.oauth2.service_account.Credentials.from_service_account_file(

service_account_file_path

)

discovery = read_datafile("logging.json")

with mock.patch(

"google.oauth2.service_account.Credentials._create_self_signed_jwt"

) as _create_self_signed_jwt:

build_from_document(

discovery,

credentials=creds,

always_use_jwt_access=True,

)

_create_self_signed_jwt.assert_called_with(

"https://logging.googleapis.com/"

)

def test_self_signed_jwt_disabled(self):

service_account_file_path = os.path.join(DATA_DIR, "service_account.json")

creds = google.oauth2.service_account.Credentials.from_service_account_file(

service_account_file_path

)

discovery = read_datafile("logging.json")

with mock.patch(

"google.oauth2.service_account.Credentials._create_self_signed_jwt"

) as _create_self_signed_jwt:

build_from_document(

discovery,

credentials=creds,

)

"version": 1,

"cert_configs": {

"workload": {

"cert_path": "path/to/cert/file",

"key_path": "path/to/key/file",

}

},

MOCK_CREDENTIALS = mock.Mock(spec=google.auth.credentials.Credentials)

_reset_universe_domain(MOCK_CREDENTIALS)

ADC_CERT_PATH = "adc_cert_path"

ADC_KEY_PATH = "adc_key_path"

ADC_PASSPHRASE = "adc_passphrase"

def check_http_client_cert(self, resource, has_client_cert="false"):

if isinstance(resource._http, google_auth_httplib2.AuthorizedHttp):

certs = list(resource._http.http.certificates.iter(""))

else:

certs = list(resource._http.certificates.iter(""))

if has_client_cert == "true":

self.assertEqual(len(certs), 1)

self.assertEqual(

certs[0], (self.ADC_KEY_PATH, self.ADC_CERT_PATH, self.ADC_PASSPHRASE)

)

else:

self.assertEqual(len(certs), 0)

def client_encrypted_cert_source(self):

return self.ADC_CERT_PATH, self.ADC_KEY_PATH, self.ADC_PASSPHRASE

@parameterized.expand(

def test_mtls_not_trigger_if_http_provided(self, use_mtls_env, use_client_cert):

discovery = read_datafile("plus.json")

with mock.patch.dict(

"os.environ", {"GOOGLE_API_USE_MTLS_ENDPOINT": use_mtls_env}

):

with mock.patch.dict(

"os.environ", {"GOOGLE_API_USE_CLIENT_CERTIFICATE": use_client_cert}

):

plus = build_from_document(discovery, http=httplib2.Http())

self.assertIsNotNone(plus)

self.assertEqual(plus._baseUrl, REGULAR_ENDPOINT)

self.check_http_client_cert(plus, has_client_cert="false")

@parameterized.expand(

def test_exception_with_client_cert_source(self, use_mtls_env, use_client_cert):

discovery = read_datafile("plus.json")

with mock.patch.dict(

"os.environ", {"GOOGLE_API_USE_MTLS_ENDPOINT": use_mtls_env}

):

with mock.patch.dict(

"os.environ", {"GOOGLE_API_USE_CLIENT_CERTIFICATE": use_client_cert}

):

with self.assertRaises(MutualTLSChannelError):

build_from_document(

discovery,

credentials=self.MOCK_CREDENTIALS,

client_options={"client_cert_source": mock.Mock()},

)

@parameterized.expand(

def test_mtls_with_provided_client_cert(

self, use_mtls_env, use_client_cert, base_url

):

discovery = read_datafile("plus.json")

with mock.patch.dict(

"os.environ", {"GOOGLE_API_USE_MTLS_ENDPOINT": use_mtls_env}

):

with mock.patch.dict(

"os.environ", {"GOOGLE_API_USE_CLIENT_CERTIFICATE": use_client_cert}

):

plus = build_from_document(

discovery,

credentials=self.MOCK_CREDENTIALS,

client_options={

"client_encrypted_cert_source": self.client_encrypted_cert_source

},

)

self.assertIsNotNone(plus)

self.check_http_client_cert(plus, has_client_cert=use_client_cert)

self.assertEqual(plus._baseUrl, base_url)

@parameterized.expand(

@pytest.mark.skipif(

def test_mtls_with_provided_client_cert_unset_environment_variable(

self, use_mtls_env, use_client_cert, config_data, base_url

):

"""Tests that mTLS is correctly handled when a client certificate is provided.

if hasattr(google.auth.transport.mtls, "should_use_client_cert"):

discovery = read_datafile("plus.json")

config_filename = "mock_certificate_config.json"

config_file_content = json.dumps(config_data)

m = mock.mock_open(read_data=config_file_content)

with mock.patch.dict(

"os.environ", {"GOOGLE_API_USE_MTLS_ENDPOINT": use_mtls_env}

):

# Clear CLOUDSDK_CONTEXT_AWARE_USE_CLIENT_CERTIFICATE so its fallback in

# google-auth (often "true" in Cloud SDK environments) does not bypass

# certificate_config.json auto-discovery.

with mock.patch.dict(

"os.environ",

{

"GOOGLE_API_USE_CLIENT_CERTIFICATE": use_client_cert,

"CLOUDSDK_CONTEXT_AWARE_USE_CLIENT_CERTIFICATE": "",

},

):

with mock.patch("builtins.open", m):

with mock.patch("os.path.exists", return_value=True):

with mock.patch.dict(

"os.environ",

{"GOOGLE_API_CERTIFICATE_CONFIG": config_filename},

):

plus = build_from_document(

discovery,

credentials=self.MOCK_CREDENTIALS,

client_options={

"client_encrypted_cert_source": self.client_encrypted_cert_source

},

)

self.assertIsNotNone(plus)

self.assertEqual(plus._baseUrl, base_url)

@parameterized.expand(

def test_endpoint_not_switch(self, use_mtls_env, use_client_cert):

# Test endpoint is not switched if user provided api endpoint

discovery = read_datafile("plus.json")

with mock.patch.dict(