fix: check response code in batch.finish (#609)

* add top-level response status code check to batch.finish

* test exception is raised when batch request fails

google/cloud/storage/batch.py

@@ -276,6 +276,11 @@ def finish(self):
         response = self._client._base_connection._make_request(
             "POST", url, data=body, headers=headers, timeout=timeout
         )
+
+        # Raise exception if the top-level batch request fails
+        if not 200 <= response.status_code < 300:
+            raise exceptions.from_http_response(response)
+
         responses = list(_unpack_batch_response(response))
         self._finish_futures(responses)
         return responses

tests/unit/test_batch.py

@@ -447,6 +447,24 @@ def test_finish_nonempty_non_multipart_response(self):
         with self.assertRaises(ValueError):
             batch.finish()
 
+    def test_finish_multipart_response_with_status_failure(self):
+        from google.cloud.exceptions import ServiceUnavailable
+
+        url = "http://api.example.com/other_api"
+        expected_response = _make_response(
+            status=http_client.SERVICE_UNAVAILABLE,
+            headers={"content-type": 'multipart/mixed; boundary="DEADBEEF="'},
+        )
+        http = _make_requests_session([expected_response])
+        connection = _Connection(http=http)
+        client = _Client(connection)
+        batch = self._make_one(client)
+        batch.API_BASE_URL = "http://api.example.com"
+        batch._requests.append(("POST", url, {}, {"foo": 1, "bar": 2}, None))
+
+        with self.assertRaises(ServiceUnavailable):
+            batch.finish()
+
     def test_as_context_mgr_wo_error(self):
         from google.cloud.storage.client import Client