#!/usr/bin/env python

# -*- coding: utf-8 -*-

#

# Copyright 2014 Google Inc. All Rights Reserved.

#

# 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.

"""Discovery document tests

Unit tests for objects created from discovery documents.

"""

from __future__ import absolute_import

__author__ = "jcgregorio@google.com (Joe Gregorio)"

from collections import defaultdict

import copy

import datetime

import io

import itertools

import json

import os

import pickle

import re

import sys

import unittest

from unittest import mock

import urllib

import google.api_core.exceptions

import google.auth.credentials

from google.auth.exceptions import MutualTLSChannelError

import google_auth_httplib2

import httplib2

from parameterized import parameterized

import uritemplate

try:

from oauth2client import GOOGLE_TOKEN_URI

from oauth2client.client import GoogleCredentials, OAuth2Credentials

HAS_OAUTH2CLIENT = True

except ImportError:

HAS_OAUTH2CLIENT = False

from googleapiclient import _helpers as util

from googleapiclient.discovery import (

DISCOVERY_URI,

MEDIA_BODY_PARAMETER_DEFAULT_VALUE,

MEDIA_MIME_TYPE_PARAMETER_DEFAULT_VALUE,

STACK_QUERY_PARAMETER_DEFAULT_VALUE,

STACK_QUERY_PARAMETERS,

V1_DISCOVERY_URI,

V2_DISCOVERY_URI,

ResourceMethodParameters,

_fix_up_media_path_base_url,

_fix_up_media_upload,

_fix_up_method_description,

_fix_up_parameters,

_urljoin,

build,

build_from_document,

key2param,

)

from googleapiclient.discovery_cache import DISCOVERY_DOC_MAX_AGE

from googleapiclient.discovery_cache.base import Cache

from googleapiclient.errors import (

HttpError,

InvalidJsonError,

MediaUploadSizeError,

ResumableUploadError,

UnacceptableMimeTypeError,

UnknownApiNameOrVersion,

UnknownFileType,

)

from googleapiclient.http import (

HttpMock,

HttpMockSequence,

MediaFileUpload,

MediaIoBaseUpload,

MediaUpload,

MediaUploadProgress,

build_http,

tunnel_patch,

)

from googleapiclient.model import JsonModel

from googleapiclient.schema import Schemas

DATA_DIR = os.path.join(os.path.dirname(__file__), "data")

def assertUrisEqual(testcase, expected, actual):

"""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()):

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

def assert_discovery_uri(testcase, actual, service_name, version, discovery):

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

for a given service and version."""

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

expanded_requested_uri = uritemplate.expand(discovery, params)

assertUrisEqual(testcase, expanded_requested_uri, actual)

def validate_discovery_requests(testcase, http_mock, service_name, version, discovery):

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

and that LAST discovery URI used was the one that was expected

for a given service and version."""

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

if len(http_mock.request_sequence) > 0:

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

assert_discovery_uri(testcase, actual_uri, service_name, version, discovery)

def datafile(filename):

return os.path.join(DATA_DIR, filename)

def read_datafile(filename, mode="r"):

with open(datafile(filename), mode=mode) as f:

return f.read()

class SetupHttplib2(unittest.TestCase):

def test_retries(self):

# Merely loading googleapiclient.discovery should set the RETRIES to 1.

self.assertEqual(1, httplib2.RETRIES)

class Utilities(unittest.TestCase):

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)

self.assertEqual(parameters.enum_params, {})

class Discovery(unittest.TestCase):

def test_discovery_http_is_closed(self):

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

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

http.close.assert_called_once()

class DiscoveryErrors(unittest.TestCase):

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,

)

class DiscoveryFromDocument(unittest.TestCase):

MOCK_CREDENTIALS = mock.Mock(spec=google.auth.credentials.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:

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:

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:

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,

)

_create_self_signed_jwt.assert_not_called()

REGULAR_ENDPOINT = "https://www.googleapis.com/plus/v1/"

MTLS_ENDPOINT = "https://www.mtls.googleapis.com/plus/v1/"

class DiscoveryFromDocumentMutualTLS(unittest.TestCase):

MOCK_CREDENTIALS = mock.Mock(spec=google.auth.credentials.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(

[

("never", "true"),

("auto", "true"),

("always", "true"),

("never", "false"),

("auto", "false"),

("always", "false"),

]

)

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(

[

("never", "true"),

("auto", "true"),

("always", "true"),

("never", "false"),

("auto", "false"),

("always", "false"),

]

)

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(

[

("never", "true", REGULAR_ENDPOINT),

("auto", "true", MTLS_ENDPOINT),

("always", "true", MTLS_ENDPOINT),

("never", "false", REGULAR_ENDPOINT),

("auto", "false", REGULAR_ENDPOINT),

("always", "false", MTLS_ENDPOINT),

]

)

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(

[

("never", "true"),

("auto", "true"),

("always", "true"),

("never", "false"),

("auto", "false"),

("always", "false"),

]

)

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(

"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={

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

"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, "https://foo.googleapis.com")

@parameterized.expand(

[

("never", "true", REGULAR_ENDPOINT),

("auto", "true", MTLS_ENDPOINT),

("always", "true", MTLS_ENDPOINT),

("never", "false", REGULAR_ENDPOINT),

("auto", "false", REGULAR_ENDPOINT),

("always", "false", MTLS_ENDPOINT),

]

)

@mock.patch(

"google.auth.transport.mtls.has_default_client_cert_source", autospec=True

)

@mock.patch(

"google.auth.transport.mtls.default_client_encrypted_cert_source", autospec=True

)

def test_mtls_with_default_client_cert(

self,

use_mtls_env,

use_client_cert,

base_url,

default_client_encrypted_cert_source,

has_default_client_cert_source,

):

has_default_client_cert_source.return_value = True

default_client_encrypted_cert_source.return_value = (

self.client_encrypted_cert_source

)

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,

adc_cert_path=self.ADC_CERT_PATH,

adc_key_path=self.ADC_KEY_PATH,

)

self.assertIsNotNone(plus)

self.check_http_client_cert(plus, has_client_cert=use_client_cert)

self.assertEqual(plus._baseUrl, base_url)

@parameterized.expand(

[

("never", "true", REGULAR_ENDPOINT),

("auto", "true", REGULAR_ENDPOINT),

("always", "true", MTLS_ENDPOINT),

("never", "false", REGULAR_ENDPOINT),

("auto", "false", REGULAR_ENDPOINT),

("always", "false", MTLS_ENDPOINT),

]

)

@mock.patch(

"google.auth.transport.mtls.has_default_client_cert_source", autospec=True

)

def test_mtls_with_no_client_cert(

self, use_mtls_env, use_client_cert, base_url, has_default_client_cert_source

):

has_default_client_cert_source.return_value = False

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,

adc_cert_path=self.ADC_CERT_PATH,

adc_key_path=self.ADC_KEY_PATH,

)

self.assertIsNotNone(plus)

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

self.assertEqual(plus._baseUrl, base_url)

class DiscoveryFromHttp(unittest.TestCase):

def setUp(self):

self.old_environ = os.environ.copy()

def tearDown(self):

os.environ = self.old_environ

def test_userip_is_added_to_discovery_uri(self):

# build() will raise an HttpError on a 400, use this to pick the request uri

# out of the raised exception.

os.environ["REMOTE_ADDR"] = "10.0.0.1"

try:

http = HttpMockSequence(

[({"status": "400"}, read_datafile("zoo.json", "rb"))]

)

zoo = build(

"zoo",