feat: cleanup and some additional tests

Author: Test User

packages/google-api-core/google/api_core/resumable_media/requests_upload.py

@@ -27,11 +27,10 @@
 _DEFAULT_TIMEOUT = 60.0  # seconds
 
 
-
 class _RecoveryRetransmit(Exception):
     """Internal exception to trigger retransmission after a state resync."""
-    pass
 
+    pass
 
 
 class ResumableUploadStatus:
@@ -223,7 +222,9 @@ def do_initiate():
 
         try:
             response = retry(do_initiate)()
-            self._machine.process_initiate_response(response.status_code, response.headers)
+            self._machine.process_initiate_response(
+                response.status_code, response.headers
+            )
         except Exception as e:
             self._machine.process_initiate_error(e)
             raise
@@ -260,9 +261,7 @@ def do_transmit():
                 transport, data, data_len
             )
             if need_recovery:
-                _LOGGER.info(
-                    "Transparent recovery triggered for chunk transmission."
-                )
+                _LOGGER.info("Transparent recovery triggered for chunk transmission.")
                 self._recover(transport)
                 # Raise to signal the outer Retry loop to resubmit
                 raise _RecoveryRetransmit()
@@ -391,9 +390,9 @@ def do_query():
             return response
 
         retry_for_recovery = google.api_core.retry.Retry(
-                    predicate=self._get_retry_predicate(),
-                    deadline=self._check_deadline(),
-                )
+            predicate=self._get_retry_predicate(),
+            deadline=self._check_deadline(),
+        )
 
         try:
             response = retry_for_recovery(do_query)()
@@ -406,7 +405,7 @@ def do_query():
 
         # Seek stream to correct position
         _LOGGER.info("Server reported %d bytes received. Seeking stream.", received)
-        
+
         if hasattr(self.stream, "seekable") and not self.stream.seekable():
             raise ValueError(
                 f"Stream is not seekable. Cannot recover upload to offset {received}."

packages/google-api-core/tests/unit/resumable_media/test_upload.py

@@ -13,6 +13,7 @@
 # limitations under the License.
 
 
+from google.api_core import exceptions
 from google.api_core.resumable_media import _common
 from google.api_core.resumable_media import _upload
 
@@ -35,7 +36,7 @@ def test_chunk_size_granularity(self):
 
     def test_initiate_request(self):
         upload = _upload.ResumableUpload("http://test.local/upload")
-        method, url, headers, body = upload.initiate_request(
+        method, url, headers, body = upload.build_initiate_request(
             stream_metadata={"x-custom-meta": "value"},
             content_type="text/plain",
             size=100,
@@ -51,7 +52,7 @@ def test_initiate_request(self):
 
     def test_initiate_request_conflicting_metadata(self):
         upload = _upload.ResumableUpload("http://test.local/upload")
-        method, url, headers, body = upload.initiate_request(
+        method, url, headers, body = upload.build_initiate_request(
             stream_metadata={
                 "x-custom-meta": "value",
                 _common.UPLOAD_PROTOCOL_HEADER: "malicious-protocol",
@@ -209,3 +210,29 @@ def test_process_recovery_response_final(self):
         upload.process_recovery_response(200, headers)
 
         assert upload.finished
+
+    def test_process_initiate_error(self):
+        upload = _upload.ResumableUpload("http://test.local/upload")
+        upload.process_initiate_error(Exception("Test error"))
+        assert upload.invalid
+
+    def test_process_chunk_error_recoverable(self):
+        upload = _upload.ResumableUpload("http://test.local/upload")
+        exc = exceptions.GoogleAPICallError("Recoverable")
+        exc.code = 400
+
+        assert upload.process_chunk_error(exc)
+        assert not upload.invalid
+
+    def test_process_chunk_error_non_recoverable(self):
+        upload = _upload.ResumableUpload("http://test.local/upload")
+        exc = exceptions.GoogleAPICallError("Non-recoverable")
+        exc.code = 500
+
+        assert not upload.process_chunk_error(exc)
+        assert upload.invalid
+
+    def test_process_recovery_error(self):
+        upload = _upload.ResumableUpload("http://test.local/upload")
+        upload.process_recovery_error(Exception("Test error"))
+        assert upload.invalid