# -*- coding: utf-8 -*- # Copyright 2020 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 os import mock import pytest try: import grpc # noqa: F401 except ImportError: pytest.skip("No GRPC", allow_module_level=True) from requests import Response # noqa I201 from requests.sessions import Session from google.api_core import client_options from google.api_core import exceptions as core_exceptions from google.api_core import gapic_v1 from google.api_core.operations_v1 import AbstractOperationsClient from google.api_core.operations_v1 import pagers from google.api_core.operations_v1 import transports import google.auth from google.auth import credentials as ga_credentials from google.auth.exceptions import MutualTLSChannelError from google.longrunning import operations_pb2 from google.oauth2 import service_account from google.protobuf import json_format # type: ignore from google.rpc import status_pb2 # type: ignore HTTP_OPTIONS = { "google.longrunning.Operations.CancelOperation": [ {"method": "post", "uri": "/v3/{name=operations/*}:cancel", "body": "*"}, ], "google.longrunning.Operations.DeleteOperation": [ {"method": "delete", "uri": "/v3/{name=operations/*}"}, ], "google.longrunning.Operations.GetOperation": [ {"method": "get", "uri": "/v3/{name=operations/*}"}, ], "google.longrunning.Operations.ListOperations": [ {"method": "get", "uri": "/v3/{name=operations}"}, ], } def client_cert_source_callback(): return b"cert bytes", b"key bytes" def _get_operations_client(http_options=HTTP_OPTIONS): transport = transports.rest.OperationsRestTransport( credentials=ga_credentials.AnonymousCredentials(), http_options=http_options ) return AbstractOperationsClient(transport=transport) # If default endpoint is localhost, then default mtls endpoint will be the same. # This method modifies the default endpoint so the client can produce a different # mtls endpoint for endpoint testing purposes. def modify_default_endpoint(client): return ( "foo.googleapis.com" if ("localhost" in client.DEFAULT_ENDPOINT) else client.DEFAULT_ENDPOINT ) def test__get_default_mtls_endpoint(): api_endpoint = "example.googleapis.com" api_mtls_endpoint = "example.mtls.googleapis.com" sandbox_endpoint = "example.sandbox.googleapis.com" sandbox_mtls_endpoint = "example.mtls.sandbox.googleapis.com" non_googleapi = "api.example.com" assert AbstractOperationsClient._get_default_mtls_endpoint(None) is None assert ( AbstractOperationsClient._get_default_mtls_endpoint(api_endpoint) == api_mtls_endpoint ) assert ( AbstractOperationsClient._get_default_mtls_endpoint(api_mtls_endpoint) == api_mtls_endpoint ) assert ( AbstractOperationsClient._get_default_mtls_endpoint(sandbox_endpoint) == sandbox_mtls_endpoint ) assert ( AbstractOperationsClient._get_default_mtls_endpoint(sandbox_mtls_endpoint) == sandbox_mtls_endpoint ) assert ( AbstractOperationsClient._get_default_mtls_endpoint(non_googleapi) == non_googleapi ) @pytest.mark.parametrize("client_class", [AbstractOperationsClient]) def test_operations_client_from_service_account_info(client_class): creds = ga_credentials.AnonymousCredentials() with mock.patch.object( service_account.Credentials, "from_service_account_info" ) as factory: factory.return_value = creds info = {"valid": True} client = client_class.from_service_account_info(info) assert client.transport._credentials == creds assert isinstance(client, client_class) assert client.transport._host == "longrunning.googleapis.com:443" @pytest.mark.parametrize( "transport_class,transport_name", [(transports.OperationsRestTransport, "rest")] ) def test_operations_client_service_account_always_use_jwt( transport_class, transport_name ): with mock.patch.object( service_account.Credentials, "with_always_use_jwt_access", create=True ) as use_jwt: creds = service_account.Credentials(None, None, None) transport_class(credentials=creds, always_use_jwt_access=True) use_jwt.assert_called_once_with(True) with mock.patch.object( service_account.Credentials, "with_always_use_jwt_access", create=True ) as use_jwt: creds = service_account.Credentials(None, None, None) transport_class(credentials=creds, always_use_jwt_access=False) use_jwt.assert_not_called() @pytest.mark.parametrize("client_class", [AbstractOperationsClient]) def test_operations_client_from_service_account_file(client_class): creds = ga_credentials.AnonymousCredentials() with mock.patch.object( service_account.Credentials, "from_service_account_file" ) as factory: factory.return_value = creds client = client_class.from_service_account_file("dummy/file/path.json") assert client.transport._credentials == creds assert isinstance(client, client_class) client = client_class.from_service_account_json("dummy/file/path.json") assert client.transport._credentials == creds assert isinstance(client, client_class) assert client.transport._host == "longrunning.googleapis.com:443" def test_operations_client_get_transport_class(): transport = AbstractOperationsClient.get_transport_class() available_transports = [ transports.OperationsRestTransport, ] assert transport in available_transports transport = AbstractOperationsClient.get_transport_class("rest") assert transport == transports.OperationsRestTransport @pytest.mark.parametrize( "client_class,transport_class,transport_name", [(AbstractOperationsClient, transports.OperationsRestTransport, "rest")], ) @mock.patch.object( AbstractOperationsClient, "DEFAULT_ENDPOINT", modify_default_endpoint(AbstractOperationsClient), ) def test_operations_client_client_options( client_class, transport_class, transport_name ): # Check that if channel is provided we won't create a new one. with mock.patch.object(AbstractOperationsClient, "get_transport_class") as gtc: transport = transport_class(credentials=ga_credentials.AnonymousCredentials()) client = client_class(transport=transport) gtc.assert_not_called() # Check that if channel is provided via str we will create a new one. with mock.patch.object(AbstractOperationsClient, "get_transport_class") as gtc: client = client_class(transport=transport_name) gtc.assert_called() # Check the case api_endpoint is provided. options = client_options.ClientOptions(api_endpoint="squid.clam.whelk") with mock.patch.object(transport_class, "__init__") as patched: patched.return_value = None client = client_class(client_options=options) patched.assert_called_once_with( credentials=None, credentials_file=None, host="squid.clam.whelk", scopes=None, client_cert_source_for_mtls=None, quota_project_id=None, client_info=transports.base.DEFAULT_CLIENT_INFO, always_use_jwt_access=True, ) # Check the case api_endpoint is not provided and GOOGLE_API_USE_MTLS_ENDPOINT is # "never". with mock.patch.dict(os.environ, {"GOOGLE_API_USE_MTLS_ENDPOINT": "never"}): with mock.patch.object(transport_class, "__init__") as patched: patched.return_value = None client = client_class() patched.assert_called_once_with( credentials=None, credentials_file=None, host=client.DEFAULT_ENDPOINT, scopes=None, client_cert_source_for_mtls=None, quota_project_id=None, client_info=transports.base.DEFAULT_CLIENT_INFO, always_use_jwt_access=True, ) # Check the case api_endpoint is not provided and GOOGLE_API_USE_MTLS_ENDPOINT is # "always". with mock.patch.dict(os.environ, {"GOOGLE_API_USE_MTLS_ENDPOINT": "always"}): with mock.patch.object(transport_class, "__init__") as patched: patched.return_value = None client = client_class() patched.assert_called_once_with( credentials=None, credentials_file=None, host=client.DEFAULT_MTLS_ENDPOINT, scopes=None, client_cert_source_for_mtls=None, quota_project_id=None, client_info=transports.base.DEFAULT_CLIENT_INFO, always_use_jwt_access=True, ) # Check the case api_endpoint is not provided and GOOGLE_API_USE_MTLS_ENDPOINT has # unsupported value. with mock.patch.dict(os.environ, {"GOOGLE_API_USE_MTLS_ENDPOINT": "Unsupported"}): with pytest.raises(MutualTLSChannelError): client = client_class() # Check the case GOOGLE_API_USE_CLIENT_CERTIFICATE has unsupported value. with mock.patch.dict( os.environ, {"GOOGLE_API_USE_CLIENT_CERTIFICATE": "Unsupported"} ): with pytest.raises(ValueError): client = client_class() # Check the case quota_project_id is provided options = client_options.ClientOptions(quota_project_id="octopus") with mock.patch.object(transport_class, "__init__") as patched: patched.return_value = None client = client_class(client_options=options) patched.assert_called_once_with( credentials=None, credentials_file=None, host=client.DEFAULT_ENDPOINT, scopes=None, client_cert_source_for_mtls=None, quota_project_id="octopus", client_info=transports.base.DEFAULT_CLIENT_INFO, always_use_jwt_access=True, ) @pytest.mark.parametrize( "client_class,transport_class,transport_name,use_client_cert_env", [ (AbstractOperationsClient, transports.OperationsRestTransport, "rest", "true"), (AbstractOperationsClient, transports.OperationsRestTransport, "rest", "false"), ], ) @mock.patch.object( AbstractOperationsClient, "DEFAULT_ENDPOINT", modify_default_endpoint(AbstractOperationsClient), ) @mock.patch.dict(os.environ, {"GOOGLE_API_USE_MTLS_ENDPOINT": "auto"}) def test_operations_client_mtls_env_auto( client_class, transport_class, transport_name, use_client_cert_env ): # This tests the endpoint autoswitch behavior. Endpoint is autoswitched to the default # mtls endpoint, if GOOGLE_API_USE_CLIENT_CERTIFICATE is "true" and client cert exists. # Check the case client_cert_source is provided. Whether client cert is used depends on # GOOGLE_API_USE_CLIENT_CERTIFICATE value. with mock.patch.dict( os.environ, {"GOOGLE_API_USE_CLIENT_CERTIFICATE": use_client_cert_env} ): options = client_options.ClientOptions( client_cert_source=client_cert_source_callback ) def fake_init(client_cert_source_for_mtls=None, **kwargs): """Invoke client_cert source if provided.""" if client_cert_source_for_mtls: client_cert_source_for_mtls() return None with mock.patch.object(transport_class, "__init__") as patched: patched.side_effect = fake_init client = client_class(client_options=options) if use_client_cert_env == "false": expected_client_cert_source = None expected_host = client.DEFAULT_ENDPOINT else: expected_client_cert_source = client_cert_source_callback expected_host = client.DEFAULT_MTLS_ENDPOINT patched.assert_called_once_with( credentials=None, credentials_file=None, host=expected_host, scopes=None, client_cert_source_for_mtls=expected_client_cert_source, quota_project_id=None, client_info=transports.base.DEFAULT_CLIENT_INFO, always_use_jwt_access=True, ) # Check the case ADC client cert is provided. Whether client cert is used depends on # GOOGLE_API_USE_CLIENT_CERTIFICATE value. with mock.patch.dict( os.environ, {"GOOGLE_API_USE_CLIENT_CERTIFICATE": use_client_cert_env} ): with mock.patch.object(transport_class, "__init__") as patched: with mock.patch( "google.auth.transport.mtls.has_default_client_cert_source", return_value=True, ): with mock.patch( "google.auth.transport.mtls.default_client_cert_source", return_value=client_cert_source_callback, ): if use_client_cert_env == "false": expected_host = client.DEFAULT_ENDPOINT expected_client_cert_source = None else: expected_host = client.DEFAULT_MTLS_ENDPOINT expected_client_cert_source = client_cert_source_callback patched.return_value = None client = client_class() patched.assert_called_once_with( credentials=None, credentials_file=None, host=expected_host, scopes=None, client_cert_source_for_mtls=expected_client_cert_source, quota_project_id=None, client_info=transports.base.DEFAULT_CLIENT_INFO, always_use_jwt_access=True, ) # Check the case client_cert_source and ADC client cert are not provided. with mock.patch.dict( os.environ, {"GOOGLE_API_USE_CLIENT_CERTIFICATE": use_client_cert_env} ): with mock.patch.object(transport_class, "__init__") as patched: with mock.patch( "google.auth.transport.mtls.has_default_client_cert_source", return_value=False, ): patched.return_value = None client = client_class() patched.assert_called_once_with( credentials=None, credentials_file=None, host=client.DEFAULT_ENDPOINT, scopes=None, client_cert_source_for_mtls=None, quota_project_id=None, client_info=transports.base.DEFAULT_CLIENT_INFO, always_use_jwt_access=True, ) @pytest.mark.parametrize( "client_class,transport_class,transport_name", [(AbstractOperationsClient, transports.OperationsRestTransport, "rest")], ) def test_operations_client_client_options_scopes( client_class, transport_class, transport_name ): # Check the case scopes are provided. options = client_options.ClientOptions(scopes=["1", "2"],) with mock.patch.object(transport_class, "__init__") as patched: patched.return_value = None client = client_class(client_options=options) patched.assert_called_once_with( credentials=None, credentials_file=None, host=client.DEFAULT_ENDPOINT, scopes=["1", "2"], client_cert_source_for_mtls=None, quota_project_id=None, client_info=transports.base.DEFAULT_CLIENT_INFO, always_use_jwt_access=True, ) @pytest.mark.parametrize( "client_class,transport_class,transport_name", [(AbstractOperationsClient, transports.OperationsRestTransport, "rest")], ) def test_operations_client_client_options_credentials_file( client_class, transport_class, transport_name ): # Check the case credentials file is provided. options = client_options.ClientOptions(credentials_file="credentials.json") with mock.patch.object(transport_class, "__init__") as patched: patched.return_value = None client = client_class(client_options=options) patched.assert_called_once_with( credentials=None, credentials_file="credentials.json", host=client.DEFAULT_ENDPOINT, scopes=None, client_cert_source_for_mtls=None, quota_project_id=None, client_info=transports.base.DEFAULT_CLIENT_INFO, always_use_jwt_access=True, ) def test_list_operations_rest( transport: str = "rest", request_type=operations_pb2.ListOperationsRequest ): client = _get_operations_client() # Mock the http request call within the method and fake a response. with mock.patch.object(Session, "request") as req: # Designate an appropriate value for the returned response. return_value = operations_pb2.ListOperationsResponse( next_page_token="next_page_token_value", ) # Wrap the value into a proper Response obj response_value = Response() response_value.status_code = 200 json_return_value = json_format.MessageToJson(return_value) response_value._content = json_return_value.encode("UTF-8") req.return_value = response_value response = client.list_operations( name="operations", filter_="my_filter", page_size=10, page_token="abc" ) actual_args = req.call_args assert actual_args.args[0] == "GET" assert ( actual_args.args[1] == "https://longrunning.googleapis.com:443/v3/operations" ) assert actual_args.kwargs["params"] == [ ("filter", "my_filter"), ("pageSize", 10), ("pageToken", "abc"), ] # Establish that the response is the type that we expect. assert isinstance(response, pagers.ListOperationsPager) assert response.next_page_token == "next_page_token_value" def test_list_operations_rest_failure(): client = _get_operations_client(http_options=None) with mock.patch.object(Session, "request") as req: response_value = Response() response_value.status_code = 400 mock_request = mock.MagicMock() mock_request.method = "GET" mock_request.url = "https://longrunning.googleapis.com:443/v1/operations" response_value.request = mock_request req.return_value = response_value with pytest.raises(core_exceptions.GoogleAPIError): client.list_operations(name="operations") def test_list_operations_rest_pager(): client = AbstractOperationsClient( credentials=ga_credentials.AnonymousCredentials(), ) # Mock the http request call within the method and fake a response. with mock.patch.object(Session, "request") as req: # TODO(kbandes): remove this mock unless there's a good reason for it. # with mock.patch.object(path_template, 'transcode') as transcode: # Set the response as a series of pages response = ( operations_pb2.ListOperationsResponse( operations=[ operations_pb2.Operation(), operations_pb2.Operation(), operations_pb2.Operation(), ], next_page_token="abc", ), operations_pb2.ListOperationsResponse( operations=[], next_page_token="def", ), operations_pb2.ListOperationsResponse( operations=[operations_pb2.Operation()], next_page_token="ghi", ), operations_pb2.ListOperationsResponse( operations=[operations_pb2.Operation(), operations_pb2.Operation()], ), ) # Two responses for two calls response = response + response # Wrap the values into proper Response objs response = tuple(json_format.MessageToJson(x) for x in response) return_values = tuple(Response() for i in response) for return_val, response_val in zip(return_values, response): return_val._content = response_val.encode("UTF-8") return_val.status_code = 200 req.side_effect = return_values pager = client.list_operations(name="operations") results = list(pager) assert len(results) == 6 assert all(isinstance(i, operations_pb2.Operation) for i in results) pages = list(client.list_operations(name="operations").pages) for page_, token in zip(pages, ["abc", "def", "ghi", ""]): assert page_.next_page_token == token def test_get_operation_rest( transport: str = "rest", request_type=operations_pb2.GetOperationRequest ): client = _get_operations_client() # Mock the http request call within the method and fake a response. with mock.patch.object(Session, "request") as req: # Designate an appropriate value for the returned response. return_value = operations_pb2.Operation( name="operations/sample1", done=True, error=status_pb2.Status(code=411), ) # Wrap the value into a proper Response obj response_value = Response() response_value.status_code = 200 json_return_value = json_format.MessageToJson(return_value) response_value._content = json_return_value.encode("UTF-8") req.return_value = response_value response = client.get_operation("operations/sample1") actual_args = req.call_args assert actual_args.args[0] == "GET" assert ( actual_args.args[1] == "https://longrunning.googleapis.com:443/v3/operations/sample1" ) # Establish that the response is the type that we expect. assert isinstance(response, operations_pb2.Operation) assert response.name == "operations/sample1" assert response.done is True def test_get_operation_rest_failure(): client = _get_operations_client(http_options=None) with mock.patch.object(Session, "request") as req: response_value = Response() response_value.status_code = 400 mock_request = mock.MagicMock() mock_request.method = "GET" mock_request.url = ( "https://longrunning.googleapis.com:443/v1/operations/sample1" ) response_value.request = mock_request req.return_value = response_value with pytest.raises(core_exceptions.GoogleAPIError): client.get_operation("operations/sample1") def test_delete_operation_rest( transport: str = "rest", request_type=operations_pb2.DeleteOperationRequest ): client = _get_operations_client() # Mock the http request call within the method and fake a response. with mock.patch.object(Session, "request") as req: # Wrap the value into a proper Response obj response_value = Response() response_value.status_code = 200 json_return_value = "" response_value._content = json_return_value.encode("UTF-8") req.return_value = response_value client.delete_operation(name="operations/sample1") assert req.call_count == 1 actual_args = req.call_args assert actual_args.args[0] == "DELETE" assert ( actual_args.args[1] == "https://longrunning.googleapis.com:443/v3/operations/sample1" ) def test_delete_operation_rest_failure(): client = _get_operations_client(http_options=None) with mock.patch.object(Session, "request") as req: response_value = Response() response_value.status_code = 400 mock_request = mock.MagicMock() mock_request.method = "DELETE" mock_request.url = ( "https://longrunning.googleapis.com:443/v1/operations/sample1" ) response_value.request = mock_request req.return_value = response_value with pytest.raises(core_exceptions.GoogleAPIError): client.delete_operation(name="operations/sample1") def test_cancel_operation_rest(transport: str = "rest"): client = _get_operations_client() # Mock the http request call within the method and fake a response. with mock.patch.object(Session, "request") as req: # Wrap the value into a proper Response obj response_value = Response() response_value.status_code = 200 json_return_value = "" response_value._content = json_return_value.encode("UTF-8") req.return_value = response_value client.cancel_operation(name="operations/sample1") assert req.call_count == 1 actual_args = req.call_args assert actual_args.args[0] == "POST" assert ( actual_args.args[1] == "https://longrunning.googleapis.com:443/v3/operations/sample1:cancel" ) def test_cancel_operation_rest_failure(): client = _get_operations_client(http_options=None) with mock.patch.object(Session, "request") as req: response_value = Response() response_value.status_code = 400 mock_request = mock.MagicMock() mock_request.method = "POST" mock_request.url = ( "https://longrunning.googleapis.com:443/v1/operations/sample1:cancel" ) response_value.request = mock_request req.return_value = response_value with pytest.raises(core_exceptions.GoogleAPIError): client.cancel_operation(name="operations/sample1") def test_credentials_transport_error(): # It is an error to provide credentials and a transport instance. transport = transports.OperationsRestTransport( credentials=ga_credentials.AnonymousCredentials(), ) with pytest.raises(ValueError): AbstractOperationsClient( credentials=ga_credentials.AnonymousCredentials(), transport=transport, ) # It is an error to provide a credentials file and a transport instance. transport = transports.OperationsRestTransport( credentials=ga_credentials.AnonymousCredentials(), ) with pytest.raises(ValueError): AbstractOperationsClient( client_options={"credentials_file": "credentials.json"}, transport=transport, ) # It is an error to provide scopes and a transport instance. transport = transports.OperationsRestTransport( credentials=ga_credentials.AnonymousCredentials(), ) with pytest.raises(ValueError): AbstractOperationsClient( client_options={"scopes": ["1", "2"]}, transport=transport, ) def test_transport_instance(): # A client may be instantiated with a custom transport instance. transport = transports.OperationsRestTransport( credentials=ga_credentials.AnonymousCredentials(), ) client = AbstractOperationsClient(transport=transport) assert client.transport is transport @pytest.mark.parametrize("transport_class", [transports.OperationsRestTransport]) def test_transport_adc(transport_class): # Test default credentials are used if not provided. with mock.patch.object(google.auth, "default") as adc: adc.return_value = (ga_credentials.AnonymousCredentials(), None) transport_class() adc.assert_called_once() def test_operations_base_transport_error(): # Passing both a credentials object and credentials_file should raise an error with pytest.raises(core_exceptions.DuplicateCredentialArgs): transports.OperationsTransport( credentials=ga_credentials.AnonymousCredentials(), credentials_file="credentials.json", ) def test_operations_base_transport(): # Instantiate the base transport. with mock.patch( "google.api_core.operations_v1.transports.OperationsTransport.__init__" ) as Transport: Transport.return_value = None transport = transports.OperationsTransport( credentials=ga_credentials.AnonymousCredentials(), ) # Every method on the transport should just blindly # raise NotImplementedError. methods = ( "list_operations", "get_operation", "delete_operation", "cancel_operation", ) for method in methods: with pytest.raises(NotImplementedError): getattr(transport, method)(request=object()) with pytest.raises(NotImplementedError): transport.close() def test_operations_base_transport_with_credentials_file(): # Instantiate the base transport with a credentials file with mock.patch.object( google.auth, "load_credentials_from_file", autospec=True ) as load_creds, mock.patch( "google.api_core.operations_v1.transports.OperationsTransport._prep_wrapped_messages" ) as Transport: Transport.return_value = None load_creds.return_value = (ga_credentials.AnonymousCredentials(), None) transports.OperationsTransport( credentials_file="credentials.json", quota_project_id="octopus", ) load_creds.assert_called_once_with( "credentials.json", scopes=None, default_scopes=(), quota_project_id="octopus", ) def test_operations_base_transport_with_adc(): # Test the default credentials are used if credentials and credentials_file are None. with mock.patch.object(google.auth, "default", autospec=True) as adc, mock.patch( "google.api_core.operations_v1.transports.OperationsTransport._prep_wrapped_messages" ) as Transport: Transport.return_value = None adc.return_value = (ga_credentials.AnonymousCredentials(), None) transports.OperationsTransport() adc.assert_called_once() def test_operations_auth_adc(): # If no credentials are provided, we should use ADC credentials. with mock.patch.object(google.auth, "default", autospec=True) as adc: adc.return_value = (ga_credentials.AnonymousCredentials(), None) AbstractOperationsClient() adc.assert_called_once_with( scopes=None, default_scopes=(), quota_project_id=None, ) def test_operations_http_transport_client_cert_source_for_mtls(): cred = ga_credentials.AnonymousCredentials() with mock.patch( "google.auth.transport.requests.AuthorizedSession.configure_mtls_channel" ) as mock_configure_mtls_channel: transports.OperationsRestTransport( credentials=cred, client_cert_source_for_mtls=client_cert_source_callback ) mock_configure_mtls_channel.assert_called_once_with(client_cert_source_callback) def test_operations_host_no_port(): client = AbstractOperationsClient( credentials=ga_credentials.AnonymousCredentials(), client_options=client_options.ClientOptions( api_endpoint="longrunning.googleapis.com" ), ) assert client.transport._host == "longrunning.googleapis.com:443" def test_operations_host_with_port(): client = AbstractOperationsClient( credentials=ga_credentials.AnonymousCredentials(), client_options=client_options.ClientOptions( api_endpoint="longrunning.googleapis.com:8000" ), ) assert client.transport._host == "longrunning.googleapis.com:8000" def test_common_billing_account_path(): billing_account = "squid" expected = "billingAccounts/{billing_account}".format( billing_account=billing_account, ) actual = AbstractOperationsClient.common_billing_account_path(billing_account) assert expected == actual def test_parse_common_billing_account_path(): expected = { "billing_account": "clam", } path = AbstractOperationsClient.common_billing_account_path(**expected) # Check that the path construction is reversible. actual = AbstractOperationsClient.parse_common_billing_account_path(path) assert expected == actual def test_common_folder_path(): folder = "whelk" expected = "folders/{folder}".format(folder=folder,) actual = AbstractOperationsClient.common_folder_path(folder) assert expected == actual def test_parse_common_folder_path(): expected = { "folder": "octopus", } path = AbstractOperationsClient.common_folder_path(**expected) # Check that the path construction is reversible. actual = AbstractOperationsClient.parse_common_folder_path(path) assert expected == actual def test_common_organization_path(): organization = "oyster" expected = "organizations/{organization}".format(organization=organization,) actual = AbstractOperationsClient.common_organization_path(organization) assert expected == actual def test_parse_common_organization_path(): expected = { "organization": "nudibranch", } path = AbstractOperationsClient.common_organization_path(**expected) # Check that the path construction is reversible. actual = AbstractOperationsClient.parse_common_organization_path(path) assert expected == actual def test_common_project_path(): project = "cuttlefish" expected = "projects/{project}".format(project=project,) actual = AbstractOperationsClient.common_project_path(project) assert expected == actual def test_parse_common_project_path(): expected = { "project": "mussel", } path = AbstractOperationsClient.common_project_path(**expected) # Check that the path construction is reversible. actual = AbstractOperationsClient.parse_common_project_path(path) assert expected == actual def test_common_location_path(): project = "winkle" location = "nautilus" expected = "projects/{project}/locations/{location}".format( project=project, location=location, ) actual = AbstractOperationsClient.common_location_path(project, location) assert expected == actual def test_parse_common_location_path(): expected = { "project": "scallop", "location": "abalone", } path = AbstractOperationsClient.common_location_path(**expected) # Check that the path construction is reversible. actual = AbstractOperationsClient.parse_common_location_path(path) assert expected == actual def test_client_withDEFAULT_CLIENT_INFO(): client_info = gapic_v1.client_info.ClientInfo() with mock.patch.object( transports.OperationsTransport, "_prep_wrapped_messages" ) as prep: AbstractOperationsClient( credentials=ga_credentials.AnonymousCredentials(), client_info=client_info, ) prep.assert_called_once_with(client_info) with mock.patch.object( transports.OperationsTransport, "_prep_wrapped_messages" ) as prep: transport_class = AbstractOperationsClient.get_transport_class() transport_class( credentials=ga_credentials.AnonymousCredentials(), client_info=client_info, ) prep.assert_called_once_with(client_info)