fix: honor custom retry in job.result() (#2302)

* fix(job): honor custom retry in job.result()
The `_AsyncJob.result()` method was not correctly passing the `retry`
argument to the superclass's `result()` method when the `retry` object
was the same as the default retry object. This caused the default
retry settings to be ignored in some cases.
This change modifies the `result()` method to always pass the `retry`
argument to the superclass, ensuring that the provided retry settings
are always honored.
A new test case is added to verify that `job.result()` correctly
handles both the default retry and a custom retry object.
* Update tests/unit/test_job_retry.py
* Update tests/unit/test_job_retry.py
* blacken and lint
* udpates retry handling and testing of retry handling
* Update tests/unit/test_job_retry.py
* Update tests/unit/test_job_retry.py
* Update tests/unit/test_job_retry.py
---------
Co-authored-by: google-labs-jules[bot] <161369871+google-labs-jules[bot]@users.noreply.github.com>

Author: chalmerlowe

google/cloud/bigquery/job/base.py

@@ -1044,8 +1044,7 @@ def result( # type: ignore # (incompatible with supertype)
 ifself.stateisNone:
 self._begin(retry=retry, timeout=timeout)
 
-kwargs= {} ifretryisDEFAULT_RETRYelse {"retry": retry}
-returnsuper(_AsyncJob, self).result(timeout=timeout, **kwargs)
+returnsuper(_AsyncJob, self).result(timeout=timeout, retry=retry)
 
 defcancelled(self):
 """Check if the job has been cancelled.

tests/unit/test_job_retry.py

@@ -615,3 +615,80 @@ def test_query_and_wait_retries_job_for_DDL_queries(global_time_lock):
 _, kwargs=calls[3]
 assertkwargs["method"] =="POST"
 assertkwargs["path"] ==query_request_path
+
+
+@pytest.mark.parametrize(
+"result_retry_param",
+ [
+pytest.param(
+ {},
+id="default retry {}",
+ ),
+pytest.param(
+ {
+"retry": google.cloud.bigquery.retry.DEFAULT_RETRY.with_timeout(
+timeout=10.0
+ )
+ },
+id="custom retry object with timeout 10.0",
+ ),
+ ],
+)
+deftest_retry_load_job_result(result_retry_param, PROJECT, DS_ID):
+fromgoogle.cloud.bigquery.datasetimportDatasetReference
+fromgoogle.cloud.bigquery.job.loadimportLoadJob
+importgoogle.cloud.bigquery.retry
+
+client=make_client()
+conn=client._connection=make_connection(
+dict(
+status=dict(state="RUNNING"),
+jobReference={"jobId": "id_1"},
+ ),
+google.api_core.exceptions.ServiceUnavailable("retry me"),
+dict(
+status=dict(state="DONE"),
+jobReference={"jobId": "id_1"},
+statistics={"load": {"outputRows": 1}},
+ ),
+ )
+
+table_ref=DatasetReference(project=PROJECT, dataset_id=DS_ID).table("new_table")
+job=LoadJob("id_1", source_uris=None, destination=table_ref, client=client)
+
+withmock.patch.object(
+client, "_call_api", wraps=client._call_api
+ ) aswrapped_call_api:
+result=job.result(**result_retry_param)
+
+assertjob.state=="DONE"
+assertresult.output_rows==1
+
+# Check that _call_api was called multiple times due to retry
+assertwrapped_call_api.call_count>1
+
+# Verify the retry object used in the calls to _call_api
+expected_retry=result_retry_param.get(
+"retry", google.cloud.bigquery.retry.DEFAULT_RETRY
+ )
+
+forcallinwrapped_call_api.mock_calls:
+name, args, kwargs=call
+# The retry object is the first positional argument to _call_api
+called_retry=args[0]
+
+# We only care about the calls made during the job.result() polling
+ifkwargs.get("method") =="GET"and"jobs/id_1"inkwargs.get("path", ""):
+assertcalled_retry._predicate==expected_retry._predicate
+assertcalled_retry._initial==expected_retry._initial
+assertcalled_retry._maximum==expected_retry._maximum
+assertcalled_retry._multiplier==expected_retry._multiplier
+assertcalled_retry._deadline==expected_retry._deadline
+if"retry"inresult_retry_param:
+# Specifically check the timeout for the custom retry case
+assertcalled_retry._timeout==10.0
+else:
+assertcalled_retry._timeout==expected_retry._timeout
+
+# The number of api_request calls should still be 3
+assertconn.api_request.call_count==3