diff --git a/packages/google-cloud-build/google/cloud/devtools/cloudbuild_v1/services/cloud_build/transports/base.py b/packages/google-cloud-build/google/cloud/devtools/cloudbuild_v1/services/cloud_build/transports/base.py index a7862c8d0b10..5f436d463eb7 100644 --- a/packages/google-cloud-build/google/cloud/devtools/cloudbuild_v1/services/cloud_build/transports/base.py +++ b/packages/google-cloud-build/google/cloud/devtools/cloudbuild_v1/services/cloud_build/transports/base.py @@ -102,7 +102,6 @@ def __init__( credentials, _ = google.auth.load_credentials_from_file( credentials_file, **scopes_kwargs, quota_project_id=quota_project_id ) - elif credentials is None: credentials, _ = google.auth.default( **scopes_kwargs, quota_project_id=quota_project_id diff --git a/packages/google-cloud-build/tests/unit/gapic/cloudbuild_v1/test_cloud_build.py b/packages/google-cloud-build/tests/unit/gapic/cloudbuild_v1/test_cloud_build.py index cf9dad2d2d81..7c013d6f9da5 100644 --- a/packages/google-cloud-build/tests/unit/gapic/cloudbuild_v1/test_cloud_build.py +++ b/packages/google-cloud-build/tests/unit/gapic/cloudbuild_v1/test_cloud_build.py @@ -248,20 +248,20 @@ def test_cloud_build_client_client_options( # unsupported value. with mock.patch.dict(os.environ, {"GOOGLE_API_USE_MTLS_ENDPOINT": "Unsupported"}): with pytest.raises(MutualTLSChannelError): - client = client_class() + client = client_class(transport=transport_name) # 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() + client = client_class(transport=transport_name) # 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(transport=transport_name, client_options=options) + client = client_class(client_options=options, transport=transport_name) patched.assert_called_once_with( credentials=None, credentials_file=None, @@ -318,7 +318,7 @@ def test_cloud_build_client_mtls_env_auto( ) with mock.patch.object(transport_class, "__init__") as patched: patched.return_value = None - client = client_class(transport=transport_name, client_options=options) + client = client_class(client_options=options, transport=transport_name) if use_client_cert_env == "false": expected_client_cert_source = None @@ -413,7 +413,7 @@ def test_cloud_build_client_client_options_scopes( options = client_options.ClientOptions(scopes=["1", "2"],) with mock.patch.object(transport_class, "__init__") as patched: patched.return_value = None - client = client_class(transport=transport_name, client_options=options) + client = client_class(client_options=options, transport=transport_name) patched.assert_called_once_with( credentials=None, credentials_file=None, @@ -444,7 +444,7 @@ def test_cloud_build_client_client_options_credentials_file( options = client_options.ClientOptions(credentials_file="credentials.json") with mock.patch.object(transport_class, "__init__") as patched: patched.return_value = None - client = client_class(transport=transport_name, client_options=options) + client = client_class(client_options=options, transport=transport_name) patched.assert_called_once_with( credentials=None, credentials_file="credentials.json", @@ -475,9 +475,8 @@ def test_cloud_build_client_client_options_from_dict(): ) -def test_create_build( - transport: str = "grpc", request_type=cloudbuild.CreateBuildRequest -): +@pytest.mark.parametrize("request_type", [cloudbuild.CreateBuildRequest, dict,]) +def test_create_build(request_type, transport: str = "grpc"): client = CloudBuildClient( credentials=ga_credentials.AnonymousCredentials(), transport=transport, ) @@ -501,10 +500,6 @@ def test_create_build( assert isinstance(response, future.Future) -def test_create_build_from_dict(): - test_create_build(request_type=dict) - - def test_create_build_empty_call(): # This test is a coverage failsafe to make sure that totally empty calls, # i.e. request == None and no flattened fields passed, work. @@ -636,7 +631,8 @@ async def test_create_build_flattened_error_async(): ) -def test_get_build(transport: str = "grpc", request_type=cloudbuild.GetBuildRequest): +@pytest.mark.parametrize("request_type", [cloudbuild.GetBuildRequest, dict,]) +def test_get_build(request_type, transport: str = "grpc"): client = CloudBuildClient( credentials=ga_credentials.AnonymousCredentials(), transport=transport, ) @@ -683,10 +679,6 @@ def test_get_build(transport: str = "grpc", request_type=cloudbuild.GetBuildRequ assert response.service_account == "service_account_value" -def test_get_build_from_dict(): - test_get_build(request_type=dict) - - def test_get_build_empty_call(): # This test is a coverage failsafe to make sure that totally empty calls, # i.e. request == None and no flattened fields passed, work. @@ -833,9 +825,8 @@ async def test_get_build_flattened_error_async(): ) -def test_list_builds( - transport: str = "grpc", request_type=cloudbuild.ListBuildsRequest -): +@pytest.mark.parametrize("request_type", [cloudbuild.ListBuildsRequest, dict,]) +def test_list_builds(request_type, transport: str = "grpc"): client = CloudBuildClient( credentials=ga_credentials.AnonymousCredentials(), transport=transport, ) @@ -862,10 +853,6 @@ def test_list_builds( assert response.next_page_token == "next_page_token_value" -def test_list_builds_from_dict(): - test_list_builds(request_type=dict) - - def test_list_builds_empty_call(): # This test is a coverage failsafe to make sure that totally empty calls, # i.e. request == None and no flattened fields passed, work. @@ -998,8 +985,10 @@ async def test_list_builds_flattened_error_async(): ) -def test_list_builds_pager(): - client = CloudBuildClient(credentials=ga_credentials.AnonymousCredentials,) +def test_list_builds_pager(transport_name: str = "grpc"): + client = CloudBuildClient( + credentials=ga_credentials.AnonymousCredentials, transport=transport_name, + ) # Mock the actual call within the gRPC stub, and fake the request. with mock.patch.object(type(client.transport.list_builds), "__call__") as call: @@ -1029,8 +1018,10 @@ def test_list_builds_pager(): assert all(isinstance(i, cloudbuild.Build) for i in results) -def test_list_builds_pages(): - client = CloudBuildClient(credentials=ga_credentials.AnonymousCredentials,) +def test_list_builds_pages(transport_name: str = "grpc"): + client = CloudBuildClient( + credentials=ga_credentials.AnonymousCredentials, transport=transport_name, + ) # Mock the actual call within the gRPC stub, and fake the request. with mock.patch.object(type(client.transport.list_builds), "__call__") as call: @@ -1117,9 +1108,8 @@ async def test_list_builds_async_pages(): assert page_.raw_page.next_page_token == token -def test_cancel_build( - transport: str = "grpc", request_type=cloudbuild.CancelBuildRequest -): +@pytest.mark.parametrize("request_type", [cloudbuild.CancelBuildRequest, dict,]) +def test_cancel_build(request_type, transport: str = "grpc"): client = CloudBuildClient( credentials=ga_credentials.AnonymousCredentials(), transport=transport, ) @@ -1166,10 +1156,6 @@ def test_cancel_build( assert response.service_account == "service_account_value" -def test_cancel_build_from_dict(): - test_cancel_build(request_type=dict) - - def test_cancel_build_empty_call(): # This test is a coverage failsafe to make sure that totally empty calls, # i.e. request == None and no flattened fields passed, work. @@ -1322,9 +1308,8 @@ async def test_cancel_build_flattened_error_async(): ) -def test_retry_build( - transport: str = "grpc", request_type=cloudbuild.RetryBuildRequest -): +@pytest.mark.parametrize("request_type", [cloudbuild.RetryBuildRequest, dict,]) +def test_retry_build(request_type, transport: str = "grpc"): client = CloudBuildClient( credentials=ga_credentials.AnonymousCredentials(), transport=transport, ) @@ -1348,10 +1333,6 @@ def test_retry_build( assert isinstance(response, future.Future) -def test_retry_build_from_dict(): - test_retry_build(request_type=dict) - - def test_retry_build_empty_call(): # This test is a coverage failsafe to make sure that totally empty calls, # i.e. request == None and no flattened fields passed, work. @@ -1483,9 +1464,8 @@ async def test_retry_build_flattened_error_async(): ) -def test_approve_build( - transport: str = "grpc", request_type=cloudbuild.ApproveBuildRequest -): +@pytest.mark.parametrize("request_type", [cloudbuild.ApproveBuildRequest, dict,]) +def test_approve_build(request_type, transport: str = "grpc"): client = CloudBuildClient( credentials=ga_credentials.AnonymousCredentials(), transport=transport, ) @@ -1509,10 +1489,6 @@ def test_approve_build( assert isinstance(response, future.Future) -def test_approve_build_from_dict(): - test_approve_build(request_type=dict) - - def test_approve_build_empty_call(): # This test is a coverage failsafe to make sure that totally empty calls, # i.e. request == None and no flattened fields passed, work. @@ -1705,9 +1681,8 @@ async def test_approve_build_flattened_error_async(): ) -def test_create_build_trigger( - transport: str = "grpc", request_type=cloudbuild.CreateBuildTriggerRequest -): +@pytest.mark.parametrize("request_type", [cloudbuild.CreateBuildTriggerRequest, dict,]) +def test_create_build_trigger(request_type, transport: str = "grpc"): client = CloudBuildClient( credentials=ga_credentials.AnonymousCredentials(), transport=transport, ) @@ -1755,10 +1730,6 @@ def test_create_build_trigger( assert response.service_account == "service_account_value" -def test_create_build_trigger_from_dict(): - test_create_build_trigger(request_type=dict) - - def test_create_build_trigger_empty_call(): # This test is a coverage failsafe to make sure that totally empty calls, # i.e. request == None and no flattened fields passed, work. @@ -1921,9 +1892,8 @@ async def test_create_build_trigger_flattened_error_async(): ) -def test_get_build_trigger( - transport: str = "grpc", request_type=cloudbuild.GetBuildTriggerRequest -): +@pytest.mark.parametrize("request_type", [cloudbuild.GetBuildTriggerRequest, dict,]) +def test_get_build_trigger(request_type, transport: str = "grpc"): client = CloudBuildClient( credentials=ga_credentials.AnonymousCredentials(), transport=transport, ) @@ -1971,10 +1941,6 @@ def test_get_build_trigger( assert response.service_account == "service_account_value" -def test_get_build_trigger_from_dict(): - test_get_build_trigger(request_type=dict) - - def test_get_build_trigger_empty_call(): # This test is a coverage failsafe to make sure that totally empty calls, # i.e. request == None and no flattened fields passed, work. @@ -2135,9 +2101,8 @@ async def test_get_build_trigger_flattened_error_async(): ) -def test_list_build_triggers( - transport: str = "grpc", request_type=cloudbuild.ListBuildTriggersRequest -): +@pytest.mark.parametrize("request_type", [cloudbuild.ListBuildTriggersRequest, dict,]) +def test_list_build_triggers(request_type, transport: str = "grpc"): client = CloudBuildClient( credentials=ga_credentials.AnonymousCredentials(), transport=transport, ) @@ -2166,10 +2131,6 @@ def test_list_build_triggers( assert response.next_page_token == "next_page_token_value" -def test_list_build_triggers_from_dict(): - test_list_build_triggers(request_type=dict) - - def test_list_build_triggers_empty_call(): # This test is a coverage failsafe to make sure that totally empty calls, # i.e. request == None and no flattened fields passed, work. @@ -2298,8 +2259,10 @@ async def test_list_build_triggers_flattened_error_async(): ) -def test_list_build_triggers_pager(): - client = CloudBuildClient(credentials=ga_credentials.AnonymousCredentials,) +def test_list_build_triggers_pager(transport_name: str = "grpc"): + client = CloudBuildClient( + credentials=ga_credentials.AnonymousCredentials, transport=transport_name, + ) # Mock the actual call within the gRPC stub, and fake the request. with mock.patch.object( @@ -2335,8 +2298,10 @@ def test_list_build_triggers_pager(): assert all(isinstance(i, cloudbuild.BuildTrigger) for i in results) -def test_list_build_triggers_pages(): - client = CloudBuildClient(credentials=ga_credentials.AnonymousCredentials,) +def test_list_build_triggers_pages(transport_name: str = "grpc"): + client = CloudBuildClient( + credentials=ga_credentials.AnonymousCredentials, transport=transport_name, + ) # Mock the actual call within the gRPC stub, and fake the request. with mock.patch.object( @@ -2441,9 +2406,8 @@ async def test_list_build_triggers_async_pages(): assert page_.raw_page.next_page_token == token -def test_delete_build_trigger( - transport: str = "grpc", request_type=cloudbuild.DeleteBuildTriggerRequest -): +@pytest.mark.parametrize("request_type", [cloudbuild.DeleteBuildTriggerRequest, dict,]) +def test_delete_build_trigger(request_type, transport: str = "grpc"): client = CloudBuildClient( credentials=ga_credentials.AnonymousCredentials(), transport=transport, ) @@ -2469,10 +2433,6 @@ def test_delete_build_trigger( assert response is None -def test_delete_build_trigger_from_dict(): - test_delete_build_trigger(request_type=dict) - - def test_delete_build_trigger_empty_call(): # This test is a coverage failsafe to make sure that totally empty calls, # i.e. request == None and no flattened fields passed, work. @@ -2608,9 +2568,8 @@ async def test_delete_build_trigger_flattened_error_async(): ) -def test_update_build_trigger( - transport: str = "grpc", request_type=cloudbuild.UpdateBuildTriggerRequest -): +@pytest.mark.parametrize("request_type", [cloudbuild.UpdateBuildTriggerRequest, dict,]) +def test_update_build_trigger(request_type, transport: str = "grpc"): client = CloudBuildClient( credentials=ga_credentials.AnonymousCredentials(), transport=transport, ) @@ -2658,10 +2617,6 @@ def test_update_build_trigger( assert response.service_account == "service_account_value" -def test_update_build_trigger_from_dict(): - test_update_build_trigger(request_type=dict) - - def test_update_build_trigger_empty_call(): # This test is a coverage failsafe to make sure that totally empty calls, # i.e. request == None and no flattened fields passed, work. @@ -2834,9 +2789,8 @@ async def test_update_build_trigger_flattened_error_async(): ) -def test_run_build_trigger( - transport: str = "grpc", request_type=cloudbuild.RunBuildTriggerRequest -): +@pytest.mark.parametrize("request_type", [cloudbuild.RunBuildTriggerRequest, dict,]) +def test_run_build_trigger(request_type, transport: str = "grpc"): client = CloudBuildClient( credentials=ga_credentials.AnonymousCredentials(), transport=transport, ) @@ -2862,10 +2816,6 @@ def test_run_build_trigger( assert isinstance(response, future.Future) -def test_run_build_trigger_from_dict(): - test_run_build_trigger(request_type=dict) - - def test_run_build_trigger_empty_call(): # This test is a coverage failsafe to make sure that totally empty calls, # i.e. request == None and no flattened fields passed, work. @@ -3017,9 +2967,10 @@ async def test_run_build_trigger_flattened_error_async(): ) -def test_receive_trigger_webhook( - transport: str = "grpc", request_type=cloudbuild.ReceiveTriggerWebhookRequest -): +@pytest.mark.parametrize( + "request_type", [cloudbuild.ReceiveTriggerWebhookRequest, dict,] +) +def test_receive_trigger_webhook(request_type, transport: str = "grpc"): client = CloudBuildClient( credentials=ga_credentials.AnonymousCredentials(), transport=transport, ) @@ -3045,10 +2996,6 @@ def test_receive_trigger_webhook( assert isinstance(response, cloudbuild.ReceiveTriggerWebhookResponse) -def test_receive_trigger_webhook_from_dict(): - test_receive_trigger_webhook(request_type=dict) - - def test_receive_trigger_webhook_empty_call(): # This test is a coverage failsafe to make sure that totally empty calls, # i.e. request == None and no flattened fields passed, work. @@ -3103,9 +3050,8 @@ async def test_receive_trigger_webhook_async_from_dict(): await test_receive_trigger_webhook_async(request_type=dict) -def test_create_worker_pool( - transport: str = "grpc", request_type=cloudbuild.CreateWorkerPoolRequest -): +@pytest.mark.parametrize("request_type", [cloudbuild.CreateWorkerPoolRequest, dict,]) +def test_create_worker_pool(request_type, transport: str = "grpc"): client = CloudBuildClient( credentials=ga_credentials.AnonymousCredentials(), transport=transport, ) @@ -3131,10 +3077,6 @@ def test_create_worker_pool( assert isinstance(response, future.Future) -def test_create_worker_pool_from_dict(): - test_create_worker_pool(request_type=dict) - - def test_create_worker_pool_empty_call(): # This test is a coverage failsafe to make sure that totally empty calls, # i.e. request == None and no flattened fields passed, work. @@ -3341,9 +3283,8 @@ async def test_create_worker_pool_flattened_error_async(): ) -def test_get_worker_pool( - transport: str = "grpc", request_type=cloudbuild.GetWorkerPoolRequest -): +@pytest.mark.parametrize("request_type", [cloudbuild.GetWorkerPoolRequest, dict,]) +def test_get_worker_pool(request_type, transport: str = "grpc"): client = CloudBuildClient( credentials=ga_credentials.AnonymousCredentials(), transport=transport, ) @@ -3383,10 +3324,6 @@ def test_get_worker_pool( assert response.etag == "etag_value" -def test_get_worker_pool_from_dict(): - test_get_worker_pool(request_type=dict) - - def test_get_worker_pool_empty_call(): # This test is a coverage failsafe to make sure that totally empty calls, # i.e. request == None and no flattened fields passed, work. @@ -3566,9 +3503,8 @@ async def test_get_worker_pool_flattened_error_async(): ) -def test_delete_worker_pool( - transport: str = "grpc", request_type=cloudbuild.DeleteWorkerPoolRequest -): +@pytest.mark.parametrize("request_type", [cloudbuild.DeleteWorkerPoolRequest, dict,]) +def test_delete_worker_pool(request_type, transport: str = "grpc"): client = CloudBuildClient( credentials=ga_credentials.AnonymousCredentials(), transport=transport, ) @@ -3594,10 +3530,6 @@ def test_delete_worker_pool( assert isinstance(response, future.Future) -def test_delete_worker_pool_from_dict(): - test_delete_worker_pool(request_type=dict) - - def test_delete_worker_pool_empty_call(): # This test is a coverage failsafe to make sure that totally empty calls, # i.e. request == None and no flattened fields passed, work. @@ -3778,9 +3710,8 @@ async def test_delete_worker_pool_flattened_error_async(): ) -def test_update_worker_pool( - transport: str = "grpc", request_type=cloudbuild.UpdateWorkerPoolRequest -): +@pytest.mark.parametrize("request_type", [cloudbuild.UpdateWorkerPoolRequest, dict,]) +def test_update_worker_pool(request_type, transport: str = "grpc"): client = CloudBuildClient( credentials=ga_credentials.AnonymousCredentials(), transport=transport, ) @@ -3806,10 +3737,6 @@ def test_update_worker_pool( assert isinstance(response, future.Future) -def test_update_worker_pool_from_dict(): - test_update_worker_pool(request_type=dict) - - def test_update_worker_pool_empty_call(): # This test is a coverage failsafe to make sure that totally empty calls, # i.e. request == None and no flattened fields passed, work. @@ -4010,9 +3937,8 @@ async def test_update_worker_pool_flattened_error_async(): ) -def test_list_worker_pools( - transport: str = "grpc", request_type=cloudbuild.ListWorkerPoolsRequest -): +@pytest.mark.parametrize("request_type", [cloudbuild.ListWorkerPoolsRequest, dict,]) +def test_list_worker_pools(request_type, transport: str = "grpc"): client = CloudBuildClient( credentials=ga_credentials.AnonymousCredentials(), transport=transport, ) @@ -4041,10 +3967,6 @@ def test_list_worker_pools( assert response.next_page_token == "next_page_token_value" -def test_list_worker_pools_from_dict(): - test_list_worker_pools(request_type=dict) - - def test_list_worker_pools_empty_call(): # This test is a coverage failsafe to make sure that totally empty calls, # i.e. request == None and no flattened fields passed, work. @@ -4226,8 +4148,10 @@ async def test_list_worker_pools_flattened_error_async(): ) -def test_list_worker_pools_pager(): - client = CloudBuildClient(credentials=ga_credentials.AnonymousCredentials,) +def test_list_worker_pools_pager(transport_name: str = "grpc"): + client = CloudBuildClient( + credentials=ga_credentials.AnonymousCredentials, transport=transport_name, + ) # Mock the actual call within the gRPC stub, and fake the request. with mock.patch.object( @@ -4266,8 +4190,10 @@ def test_list_worker_pools_pager(): assert all(isinstance(i, cloudbuild.WorkerPool) for i in results) -def test_list_worker_pools_pages(): - client = CloudBuildClient(credentials=ga_credentials.AnonymousCredentials,) +def test_list_worker_pools_pages(transport_name: str = "grpc"): + client = CloudBuildClient( + credentials=ga_credentials.AnonymousCredentials, transport=transport_name, + ) # Mock the actual call within the gRPC stub, and fake the request. with mock.patch.object( @@ -5095,7 +5021,7 @@ def test_parse_common_location_path(): assert expected == actual -def test_client_withDEFAULT_CLIENT_INFO(): +def test_client_with_default_client_info(): client_info = gapic_v1.client_info.ClientInfo() with mock.patch.object(