temp changes

Author: Pulkit0110

google/cloud/storage/_experimental/asyncio/retry/bidi_stream_retry_manager.py

@@ -69,15 +69,22 @@ async def execute(self, initial_state: Any, retry_policy):
                     self._strategy.update_state_from_response(response, state)
                 return
             except Exception as e:
-                if not retry_policy._predicate(e):
-                    raise
+                # AsyncRetry may expose either 'on_error' (public) or the private
+                # '_on_error' depending on google.api_core version. Call whichever
+                # exists so the retry policy can decide to raise (non-retriable /
+                # deadline exceeded) or allow a retry.
+                on_error_callable = getattr(retry_policy, "on_error", None)
+                if on_error_callable is None:
+                    on_error_callable = getattr(retry_policy, "_on_error", None)
 
-                await self._strategy.recover_state_on_failure(e, state)
+                if on_error_callable is None:
+                    # No hook available on the policy; re-raise the error.
+                    raise
 
-                sleep = next(sleep_generator)
-                if deadline is not None and time.monotonic() + sleep > deadline:
-                    raise exceptions.RetryError(
-                        f"Deadline of {retry_policy._deadline}s exceeded", cause=e
-                    ) from e
+                # Let the retry policy handle the error (may raise RetryError).
+                await on_error_callable(e)
 
-                await asyncio.sleep(sleep)
+                # If the retry policy did not raise, allow the strategy to recover
+                # and then sleep per policy before next attempt.
+                await self._strategy.recover_state_on_failure(e, state)
+                await retry_policy.sleep()

tests/conftest.py

@@ -0,0 +1,9 @@
+import asyncio
+
+# Ensure there's an event loop in the main thread so grpc.aio.create_channel
+# and other code that expects a loop at import/initialization time don't fail.
+try:
+    asyncio.get_running_loop()
+except RuntimeError:
+    loop = asyncio.new_event_loop()
+    asyncio.set_event_loop(loop)

tests/unit/asyncio/retry/test_bidi_stream_retry_manager.py

@@ -59,6 +59,8 @@ async def mock_stream_opener(*args, **kwargs):
             strategy=mock_strategy, stream_opener=mock_stream_opener
         )
         retry_policy = AsyncRetry(predicate=_is_retriable, initial=0.01)
+        retry_policy.sleep = mock.AsyncMock()
+
         await retry_manager.execute(initial_state={}, retry_policy=retry_policy)
 
         self.assertEqual(attempt_count, 2)