diff --git a/packages/google-auth/google/auth/transport/_mtls_helper.py b/packages/google-auth/google/auth/transport/_mtls_helper.py
index eb0600740c0d..647829ef4f07 100644
--- a/packages/google-auth/google/auth/transport/_mtls_helper.py
+++ b/packages/google-auth/google/auth/transport/_mtls_helper.py
@@ -609,7 +609,10 @@ def get_client_ssl_credentials(
     """
 
     # 1.  Attempt to retrieve X.509 Workload cert and key.
-    cert, key = _get_workload_cert_and_key(certificate_config_path)
+    try:
+        cert, key = _get_workload_cert_and_key(certificate_config_path)
+    except exceptions.ClientCertError:
+        cert, key = None, None
     if cert and key:
         return True, cert, key, None
 
diff --git a/packages/google-auth/google/auth/transport/grpc.py b/packages/google-auth/google/auth/transport/grpc.py
index 7482038589a3..563495973db0 100644
--- a/packages/google-auth/google/auth/transport/grpc.py
+++ b/packages/google-auth/google/auth/transport/grpc.py
@@ -17,7 +17,13 @@
 from __future__ import absolute_import
 
 import logging
+import threading
+import collections.abc
+import time
+import random
+import concurrent.futures
 
+_LOGGER = logging.getLogger(__name__)
 from google.auth import exceptions
 from google.auth.transport import _mtls_helper
 from google.auth.transport import mtls
@@ -209,7 +215,7 @@ def my_client_cert_callback():
 
         channel = google.auth.transport.grpc.secure_authorized_channel(
             credentials, request, mtls_endpoint)
-
+      
     Args:
         credentials (google.auth.credentials.Credentials): The credentials to
             add to requests.
@@ -254,6 +260,7 @@ def my_client_cert_callback():
         )
 
     # If SSL credentials are not explicitly set, try client_cert_callback and ADC.
+    cached_cert = None
     if not ssl_credentials:
         use_client_cert = _mtls_helper.check_use_client_cert()
         if use_client_cert and client_cert_callback:
@@ -262,10 +269,12 @@ def my_client_cert_callback():
             ssl_credentials = grpc.ssl_channel_credentials(
                 certificate_chain=cert, private_key=key
             )
+            cached_cert = cert   
         elif use_client_cert:
             # Use application default SSL credentials.
-            adc_ssl_credentils = SslCredentials()
-            ssl_credentials = adc_ssl_credentils.ssl_credentials
+            adc_ssl_credentials = SslCredentials()
+            ssl_credentials = adc_ssl_credentials.ssl_credentials
+            cached_cert = adc_ssl_credentials._cached_cert
         else:
             ssl_credentials = grpc.ssl_channel_credentials()
 
@@ -273,9 +282,27 @@ def my_client_cert_callback():
     composite_credentials = grpc.composite_channel_credentials(
         ssl_credentials, google_auth_credentials
     )
-
-    return grpc.secure_channel(target, composite_credentials, **kwargs)
-
+    is_retry = kwargs.pop("_is_retry", False)
+    channel = grpc.secure_channel(target, composite_credentials, **kwargs)
+    # Check if we are already inside a retry to avoid infinite recursion
+    if cached_cert and not is_retry:
+        # Package arguments to recreate the channel if rotation occurs
+        factory_args = {
+            "credentials": credentials,
+            "request": request, 
+            "target": target,
+            "ssl_credentials": None,
+            "client_cert_callback": client_cert_callback,
+            "_is_retry": True, # Hidden flag to stop recursion
+            **kwargs
+        }
+        interceptor = _MTLSCallInterceptor()
+
+        wrapper = _MTLSRefreshingChannel(target, factory_args, channel, cached_cert)
+     
+        interceptor._wrapper = wrapper   
+        return grpc.intercept_channel(wrapper, interceptor)
+    return channel
 
 class SslCredentials:
     """Class for application default SSL credentials.
@@ -298,6 +325,7 @@ class SslCredentials:
 
     def __init__(self):
         use_client_cert = _mtls_helper.check_use_client_cert()
+        self._cached_cert = None
         if not use_client_cert:
             self._is_mtls = False
         else:
@@ -326,6 +354,7 @@ def ssl_credentials(self):
                     self._ssl_credentials = grpc.ssl_channel_credentials(
                         certificate_chain=cert, private_key=key
                     )
+                    self._cached_cert = cert
                 else:
                     self._ssl_credentials = grpc.ssl_channel_credentials()
                     self._is_mtls = False
@@ -341,3 +370,435 @@ def ssl_credentials(self):
     def is_mtls(self):
         """Indicates if the created SSL channel credentials is mutual TLS."""
         return self._is_mtls
+
+class _MTLSCallInterceptor(
+    grpc.UnaryUnaryClientInterceptor,
+    grpc.UnaryStreamClientInterceptor,
+    grpc.StreamUnaryClientInterceptor,
+    grpc.StreamStreamClientInterceptor,
+):
+    def __init__(self):
+        self._wrapper = None
+        self._max_retries = 2 # Set your desired limit here
+        self._executor = concurrent.futures.ThreadPoolExecutor(max_workers=5)
+
+    def _should_retry(self, code, retry_count, attempt_cert):
+        if code != grpc.StatusCode.UNAUTHENTICATED or not self._wrapper:
+            return False
+
+        if retry_count >= self._max_retries:
+            _LOGGER.debug("Max retries reached (%d/%d).", retry_count, self._max_retries)
+            return False
+
+        # If the wrapper has already rotated to a new cert, we can retry immediately
+        if attempt_cert != self._wrapper._cached_cert:
+            return True
+
+        # Fingerprint check logic
+        _, _, cached_fp, current_fp = _mtls_helper.check_parameters_for_unauthorized_response(attempt_cert)
+        return cached_fp != current_fp
+
+    def intercept_unary_unary(self, continuation, client_call_details, request):
+        retry_count = 0
+
+        while True:
+            attempt_cert = self._wrapper._cached_cert if self._wrapper else None
+            try:
+                # Every time we call continuation(), our Wrapper (which is the channel
+                # being intercepted) will point to its CURRENT active raw channel.
+                response = continuation(client_call_details, request)
+                status_code = response.code()
+            except grpc.RpcError as e:
+                status_code = e.code()
+                if not self._should_retry(status_code, retry_count, attempt_cert):
+                    raise e
+                # If we should retry, we fall through to the refresh logic below
+
+            if self._should_retry(status_code, retry_count, attempt_cert):
+                retry_count += 1
+                # Tell the wrapper to swap the channel.
+                # We don't need the wrapper to execute the retry; the loop does it!
+                self._wrapper.refresh_logic(retry_count)
+                continue # Jump back to the start of the while loop
+
+            return response
+
+    def intercept_unary_stream(self, continuation, client_call_details, request):
+        return _RetryableUnaryStreamCall(continuation, client_call_details, request, self)
+
+    def intercept_stream_unary(self, continuation, client_call_details, request_iterator):
+        return _RetryableStreamUnaryFuture(continuation, client_call_details, request_iterator, self)
+
+    def intercept_stream_stream(self, continuation, client_call_details, request_iterator):
+        return _RetryableStreamStreamCall(continuation, client_call_details, request_iterator, self)
+
+class _MTLSRefreshingChannel(grpc.Channel):
+    def __init__(self, target, factory_args, initial_channel, initial_cert):
+        self._target = target
+        self._factory_args = factory_args
+        self._channel = initial_channel
+        self._cached_cert = initial_cert
+        self._lock = threading.Lock()
+        self._subscribers = set()
+
+    def refresh_logic(self, count):
+        with self._lock:
+            # Re-check inside lock to prevent race conditions
+            _, _, cached_fp, current_fp = _mtls_helper.check_parameters_for_unauthorized_response(self._cached_cert)
+            if cached_fp != current_fp:
+                _LOGGER.debug("Wrapper: Refreshing mTLS channel. Retry count: %d", count)
+                old_channel = self._channel
+                client_cert_callback = self._factory_args.get("client_cert_callback")
+                if client_cert_callback:
+                    cert, _ = client_cert_callback()
+                    self._cached_cert = cert
+                else:
+                    try:
+                        creds = _mtls_helper.get_client_ssl_credentials()
+                        self._cached_cert = creds[1]
+                    except Exception:
+                        pass
+                        
+                self._channel = secure_authorized_channel(**self._factory_args)
+                
+                for callback in self._subscribers:
+                    try:
+                        old_channel.unsubscribe(callback)
+                    except Exception:
+                        pass
+                    self._channel.subscribe(callback)
+
+    def unary_unary(self, method, *args, **kwargs):
+        # Always return a callable from the CURRENT channel
+        return self._channel.unary_unary(method, *args, **kwargs)
+
+    # Mandatory passthroughs
+    def unary_stream(self, method, *args, **kwargs): return self._channel.unary_stream(method, *args, **kwargs)
+    def stream_unary(self, method, *args, **kwargs): return self._channel.stream_unary(method, *args, **kwargs)
+    def stream_stream(self, method, *args, **kwargs): return self._channel.stream_stream(method, *args, **kwargs)
+    
+    def subscribe(self, callback, try_to_connect=False):
+        with self._lock:
+            self._subscribers.add(callback)
+            return self._channel.subscribe(callback, try_to_connect=try_to_connect)
+
+    def unsubscribe(self, callback):
+        with self._lock:
+            self._subscribers.discard(callback)
+            return self._channel.unsubscribe(callback)
+            
+    def close(self): self._channel.close()
+
+
+class _RetryableUnaryStreamCall(grpc.Call, collections.abc.Iterator):
+    def __init__(self, continuation, client_call_details, request, interceptor):
+        self._continuation = continuation
+        self._client_call_details = client_call_details
+        self._request = request
+        self._interceptor = interceptor
+        self._retry_count = 0
+        self._call = None
+        self._iterator = None
+        self._yielded_any = False
+        self._start_call()
+
+    def _start_call(self):
+        self._attempt_cert = self._interceptor._wrapper._cached_cert if self._interceptor._wrapper else None
+        self._call = self._continuation(self._client_call_details, self._request)
+        self._iterator = iter(self._call)
+
+    def __iter__(self):
+        return self
+
+    def __next__(self):
+        while True:
+            try:
+                val = next(self._iterator)
+                self._yielded_any = True
+                return val
+            except grpc.RpcError as e:
+                status_code = e.code()
+                if not self._yielded_any and self._interceptor._should_retry(status_code, self._retry_count, self._attempt_cert):
+                    self._retry_count += 1
+                    self._interceptor._wrapper.refresh_logic(self._retry_count)
+                    _LOGGER.info("gRPC stream connection dropped due to cert rotation. Transparently re-fetching the stream...")
+                    time.sleep(random.uniform(0.1, 1.0))
+                    self._start_call()
+                    continue
+                
+                if getattr(self._interceptor, "_wrapper", None):
+                    if self._interceptor._should_retry(status_code, 0, self._attempt_cert):
+                        self._interceptor._wrapper.refresh_logic(1)
+                raise e
+
+    def cancel(self): self._call.cancel()
+    def is_active(self): return self._call.is_active()
+    def time_remaining(self): return self._call.time_remaining()
+    def add_callback(self, callback): self._call.add_callback(callback)
+    def initial_metadata(self): return self._call.initial_metadata()
+    def trailing_metadata(self): return self._call.trailing_metadata()
+    def code(self): return self._call.code()
+    def details(self): return self._call.details()
+
+
+class _RetryableStreamUnaryFuture(grpc.Call, grpc.Future):
+    def __init__(self, continuation, client_call_details, request_iterator, interceptor):
+        self._continuation = continuation
+        self._client_call_details = client_call_details
+        self._replayable_request_iterator = _ReplayableIterator(request_iterator)
+        self._interceptor = interceptor
+        self._retry_count = 0
+        self._done_callbacks = []
+        self._target_future = None
+        self._lock = threading.Lock()
+        self._start_call()
+
+    def _on_inner_future_done(self, inner_future):
+        with self._lock:
+            if inner_future is not self._target_future:
+                return
+                
+        exc = inner_future.exception()
+        if isinstance(exc, grpc.RpcError):
+            status_code = exc.code()
+            can_replay = self._replayable_request_iterator.can_replay()
+            
+            if can_replay and self._interceptor._should_retry(status_code, self._retry_count, getattr(self, "_attempt_cert", None)):
+                self._retry_count += 1
+                
+                def async_retry():
+                    self._interceptor._wrapper.refresh_logic(self._retry_count)
+                    time.sleep(random.uniform(0.1, 1.0))
+                    self._start_call()
+                    
+                self._interceptor._executor.submit(async_retry)
+                return
+                
+            if getattr(self._interceptor, "_wrapper", None):
+                if self._interceptor._should_retry(status_code, 0, getattr(self, "_attempt_cert", None)):
+                    self._interceptor._wrapper.refresh_logic(1)
+
+        with self._lock:
+            for cb in self._done_callbacks:
+                cb(self)
+
+    def _start_call(self):
+        self._attempt_cert = self._interceptor._wrapper._cached_cert if self._interceptor._wrapper else None
+        req_iter = iter(self._replayable_request_iterator)
+        with self._lock:
+            self._target_future = self._continuation(self._client_call_details, req_iter)
+            self._target_future.add_done_callback(self._on_inner_future_done)
+
+    def result(self, timeout=None):
+        deadline = time.time() + timeout if timeout else None
+        
+        while True:
+            with self._lock:
+                current_future = self._target_future
+                
+            try:
+                if deadline:
+                    remaining = deadline - time.time()
+                    if remaining <= 0:
+                        raise grpc.FutureTimeoutError()
+                    return current_future.result(timeout=remaining)
+                else:    
+                    return current_future.result()
+                    
+            except grpc.RpcError as e:
+                with self._lock:
+                    if current_future is not self._target_future:
+                        continue
+                raise e
+
+    def add_done_callback(self, fn):
+        with self._lock:
+            self._done_callbacks.append(fn)
+            if self._target_future.done():
+                exc = self._target_future.exception()
+                if not (isinstance(exc, grpc.RpcError) and self._interceptor._should_retry(exc.code(), self._retry_count, getattr(self, "_attempt_cert", None))):
+                    fn(self)
+
+    def exception(self, timeout=None):
+        try:
+            self.result(timeout)
+            return None
+        except Exception as e:
+            return e
+            
+    def traceback(self, timeout=None):
+        try:
+            self.result(timeout)
+            return None
+        except Exception:
+            with self._lock:
+                return self._target_future.traceback(timeout=timeout)
+
+    def cancel(self): 
+        with self._lock: return self._target_future.cancel()
+    def cancelled(self): 
+        with self._lock: return self._target_future.cancelled()
+    def running(self): 
+        with self._lock: return self._target_future.running()
+    def done(self): 
+        with self._lock: return self._target_future.done()
+    def code(self): 
+        with self._lock: return self._target_future.code()
+    def details(self): 
+        with self._lock: return self._target_future.details()
+    def is_active(self):
+        with self._lock: return self._target_future.is_active()
+    def time_remaining(self):
+        with self._lock: return self._target_future.time_remaining()
+    def initial_metadata(self):
+        with self._lock: return self._target_future.initial_metadata()
+    def trailing_metadata(self):
+        with self._lock: return self._target_future.trailing_metadata()
+    def add_callback(self, cb):
+        with self._lock: return self._target_future.add_callback(cb)
+
+
+class _ReplayableIterator(object):
+    def __init__(self, target_iterator, max_items=1000):
+        self._target_iterator = target_iterator
+        self._max_items = max_items
+        self._buffer = []
+        self._exhausted = False
+        self._can_replay = True
+        
+        self._lock = threading.Lock()
+        self._consumer_lock = threading.Lock()
+        self._active_reader = None
+
+    def __iter__(self):
+        reader = _ReplayableIteratorReader(self)
+        with self._lock:
+            self._active_reader = reader
+        return reader
+
+    def can_replay(self):
+        with self._lock:
+            return self._can_replay
+
+
+class _ReplayableIteratorReader(object):
+    def __init__(self, parent):
+        self._parent = parent
+        self._read_index = 0
+
+    def __next__(self):
+        while True:
+            with self._parent._lock:
+                if self._read_index < len(self._parent._buffer):
+                    val = self._parent._buffer[self._read_index]
+                    self._read_index += 1
+                    return val
+
+                if self._parent._exhausted:
+                    raise StopIteration()
+
+                if self._parent._active_reader is not self:
+                    raise StopIteration()
+
+            with self._parent._consumer_lock:
+                with self._parent._lock:
+                    if self._read_index < len(self._parent._buffer):
+                        continue
+                    if self._parent._active_reader is not self:
+                        raise StopIteration()
+
+                try:
+                    val = next(self._parent._target_iterator)
+                except StopIteration:
+                    with self._parent._lock:
+                        if self._parent._active_reader is self:
+                            self._parent._exhausted = True
+                    raise
+
+            with self._parent._lock:
+                if self._parent._active_reader is not self:
+                    if self._parent._can_replay:
+                        self._parent._buffer.append(val)
+                    raise StopIteration()
+
+                if self._parent._can_replay:
+                    self._parent._buffer.append(val)
+                    if len(self._parent._buffer) > self._parent._max_items:
+                        self._parent._buffer.clear()
+                        self._parent._can_replay = False
+
+                self._read_index += 1
+                return val
+
+
+class _RetryableStreamStreamCall(grpc.Call, collections.abc.Iterator):
+    def __init__(self, continuation, client_call_details, request_iterator, interceptor):
+        self._continuation = continuation
+        self._client_call_details = client_call_details
+        self._replayable_request_iterator = _ReplayableIterator(request_iterator)
+        self._interceptor = interceptor
+        self._retry_count = 0
+        self._done_callbacks = []
+        self._call = None
+        self._response_iterator = None
+        self._yielded_any_response = False
+        self._start_call()
+    def _on_inner_call_done(self, inner_call):
+        if inner_call is not self._call:
+            return
+            
+        status_code = inner_call.code()
+        if status_code == grpc.StatusCode.UNAUTHENTICATED:
+            can_replay = self._replayable_request_iterator.can_replay()
+            if not self._yielded_any_response and can_replay and self._interceptor._should_retry(status_code, self._retry_count, getattr(self, "_attempt_cert", None)):
+                # IMPORTANT: Swallow the callback so bidi.py does not tear down 
+                # the router tracking threads while we attempt to reconstruct the stream!
+                return
+
+        for cb in self._done_callbacks:
+            cb(self)
+    def _start_call(self):
+        self._attempt_cert = self._interceptor._wrapper._cached_cert if self._interceptor._wrapper else None
+        req_iter = iter(self._replayable_request_iterator)
+        self._call = self._continuation(self._client_call_details, req_iter)
+        self._response_iterator = iter(self._call)
+        self._call.add_done_callback(self._on_inner_call_done)
+
+    def add_done_callback(self, callback):
+        # Store requested callbacks natively instead of forwarding blindly
+        self._done_callbacks.append(callback)
+
+    def __iter__(self):
+        return self
+
+    def __next__(self):
+        while True:
+            try:
+                val = next(self._response_iterator)
+                self._yielded_any_response = True
+                return val
+            except grpc.RpcError as e:
+                status_code = e.code()
+                can_replay = self._replayable_request_iterator.can_replay()
+                if not self._yielded_any_response and can_replay and self._interceptor._should_retry(status_code, self._retry_count, getattr(self, "_attempt_cert", None)):
+                    self._retry_count += 1
+                    self._interceptor._wrapper.refresh_logic(self._retry_count)
+                    _LOGGER.info("gRPC stream connection dropped due to cert rotation. Transparently re-fetching the stream...")
+                    time.sleep(random.uniform(0.1, 1.0))
+                    self._start_call()
+                    continue
+                
+                if getattr(self._interceptor, "_wrapper", None):
+                    if self._interceptor._should_retry(status_code, 0, getattr(self, "_attempt_cert", None)):
+                        self._interceptor._wrapper.refresh_logic(1)
+                raise e
+
+    # Simple pass-throughs for the remaining gRPC methods
+    def cancel(self): return self._call.cancel()
+    def code(self): return self._call.code()
+    def details(self): return self._call.details()
+    def is_active(self): return self._call.is_active()
+    def time_remaining(self): return self._call.time_remaining()
+    def add_callback(self, callback): self._call.add_callback(callback)
+    def initial_metadata(self): return self._call.initial_metadata()
+    def trailing_metadata(self): return self._call.trailing_metadata()
diff --git a/packages/google-auth/python_grpc_401_stream_unary_test.py b/packages/google-auth/python_grpc_401_stream_unary_test.py
new file mode 100644
index 000000000000..61de4edab3b6
--- /dev/null
+++ b/packages/google-auth/python_grpc_401_stream_unary_test.py
@@ -0,0 +1,102 @@
+"""Python gRPC stream-unary example test for cert rotation resilience.
+
+This test validates Stream-Unary methodologies by targeting Cloud Storage via raw Channels.
+"""
+
+import concurrent.futures
+from unittest import mock
+
+import grpc
+import google.auth
+import google.auth.credentials
+import google.auth.transport.grpc
+import google.auth.transport.requests
+
+class RecoveringCredentials(google.auth.credentials.Credentials):
+    """Fails on attempt 1, but succeeds with real Google credentials on attempt 2."""
+    def __init__(self):
+        super().__init__()
+        self.attempts = 0
+        try:
+            self.real_creds, _ = google.auth.default()
+        except:
+            print("WARNING: Could not load default credentials. Some fallback authentication features may not work.")
+            self.real_creds = None
+
+    def refresh(self, request):
+        if self.real_creds:
+            self.real_creds.refresh(request)
+
+    def before_request(self, request, method, url, headers):
+        if self.attempts == 0:
+            print(f"> Attempt {self.attempts}: Sending INVALID token to force UNAUTHENTICATED error.")
+            headers["authorization"] = "Bearer simulated_invalid_token"
+        else:
+            print(f"> Attempt {self.attempts}: Sending REAL token to bypass AUTH check!")
+            if self.real_creds:
+                self.real_creds.before_request(request, method, url, headers)
+            else:
+                headers["authorization"] = "Bearer still_invalid_no_gcloud_auth_credentials_found"
+        self.attempts += 1
+
+
+def test_grpc_stream_unary_example():
+    """Run a Stream-Unary request to verify gRPC resilience."""
+
+    credentials = RecoveringCredentials()
+    auth_request = google.auth.transport.requests.Request()
+
+    # Hit the true mTLS endpoint. This requires GOOGLE_API_USE_CLIENT_CERTIFICATE=true
+    # to be set in your terminal so it automatically picks up your device certificate!
+    target = "storage.mtls.googleapis.com:443"
+
+    print(f"Attempting to create channel configuration for {target}...")
+    channel = google.auth.transport.grpc.secure_authorized_channel(
+        credentials,
+        auth_request,
+        target,
+        # Notice we removed client_cert_callback to let google.auth fetch the real device cert automatically
+    )
+
+    stream_unary_method = channel.stream_unary(
+        "/google.storage.v2.Storage/WriteObject",
+        request_serializer=lambda x: x.encode("utf-8"),
+        response_deserializer=lambda x: x,
+    )
+
+    def payload_generator():
+        yield "Chunk 1: Payload transmission"
+        yield "Chunk 2: Simulating broken stream logic"
+
+    # Mock `check_parameters` so the interceptor assumes the cert on disk changed during our 401 response
+    with mock.patch(
+        "google.auth.transport._mtls_helper.check_parameters_for_unauthorized_response",
+        return_value=("foo.pem", "foo.pem", "old_fp", "new_fp"),
+    ) as mock_check_params:
+
+        print("Firing Stream-Unary...")
+        future = stream_unary_method.future(payload_generator())
+        
+        def future_done_callback(completed_future):
+            try:
+                completed_future.result()
+            except grpc.RpcError:
+                # We expect the final call to be executed fully
+                pass
+
+        future.add_done_callback(future_done_callback)
+
+        try:
+            future.result(timeout=5)
+        except Exception as e:  
+            print(f"Final Execution Error Code: {e.code() if hasattr(e, 'code') else e}")
+            print(f"Total times rotation interceptor was triggered: {mock_check_params.call_count}")
+
+            if hasattr(e, 'code') and e.code() != grpc.StatusCode.UNAUTHENTICATED:
+                print("\n\033[92m>>> SUCCESS! The request bypassed the authentication layer natively and was processed by GCP! <<<\033[0m")
+                print(">>> (We received INVALID_ARGUMENT instead of UNAUTHENTICATED because we uploaded raw utf8 strings instead of a Protobuf format, but Auth passed!) <<<")
+
+if __name__ == "__main__":
+    print("Starting Stream-Unary streaming script...")
+    test_grpc_stream_unary_example()
+    print("Script finished.")
diff --git a/packages/google-auth/tests/transport/test_grpc_mtls_streaming.py b/packages/google-auth/tests/transport/test_grpc_mtls_streaming.py
new file mode 100644
index 000000000000..24c7755b1307
--- /dev/null
+++ b/packages/google-auth/tests/transport/test_grpc_mtls_streaming.py
@@ -0,0 +1,123 @@
+import pytest
+from unittest import mock
+import grpc
+import threading
+import time
+
+from google.auth.transport.grpc import (
+    _ReplayableIterator,
+    _MTLSRefreshingChannel,
+    _MTLSCallInterceptor,
+)
+from google.auth.transport import _mtls_helper
+
+class TestReplayableIterator:
+    def test_buffer_and_replay(self):
+        source = iter([1, 2, 3])
+        replayable = _ReplayableIterator(source, max_items=2)
+        
+        # Read two items
+        reader = iter(replayable)
+        assert next(reader) == 1
+        assert next(reader) == 2
+        
+        # Reader is preempted/dies, we should be able to start another reader
+        # since it fits in the buffer
+        assert replayable.can_replay()
+        
+        reader2 = iter(replayable)
+        assert next(reader2) == 1
+        assert next(reader2) == 2
+        assert next(reader2) == 3
+        
+        # Since it exceeded max_items=2 during reading 3, can_replay becomes False
+        assert not replayable.can_replay()
+
+    def test_concurrent_handoff(self):
+        def slow_source():
+            yield 1
+            yield 2
+            time.sleep(0.5)
+            yield 3
+
+        replayable = _ReplayableIterator(slow_source())
+        reader1 = iter(replayable)
+        
+        # start first reader in a thread
+        values1 = []
+        def read_thread():
+            try:
+                for v in reader1:
+                    values1.append(v)
+            except Exception:
+                pass
+                
+        t = threading.Thread(target=read_thread)
+        t.start()
+        
+        # let it read 1, 2
+        time.sleep(0.1)
+        
+        # Now start second reader. First reader should abort when it wakes up.
+        reader2 = iter(replayable)
+        values2 = [v for v in reader2]
+        
+        t.join()
+        
+        # Reader 1 should only have read 1, 2 before being aborted
+        assert values1 == [1, 2]
+        # Reader 2 should get everything
+        assert values2 == [1, 2, 3]
+
+
+class _MockCall(grpc.Call):
+    def __init__(self, code, should_fail=True):
+        self._code = code
+        self._should_fail = should_fail
+        self._count = 0
+
+    def code(self):
+        return self._code
+
+    def is_active(self):
+        return True
+
+    def __iter__(self):
+        return self
+
+    def __next__(self):
+        if self._count == 0 and self._should_fail:
+            self._count += 1
+            err = grpc.RpcError()
+            err.code = lambda: self._code
+            raise err
+        self._count += 1
+        return "success"
+
+
+class TestMTLSRefreshingChannel:
+    @mock.patch("google.auth.transport._mtls_helper.check_parameters_for_unauthorized_response")
+    @mock.patch("google.auth.transport.grpc.secure_authorized_channel")
+    def test_refresh_logic(self, mock_secure_channel, mock_check_params):
+        # mock fingerprint differences indicating rotation is needed
+        mock_check_params.return_value = (None, None, b"old", b"new")
+        mock_secure_channel.return_value = mock.Mock(spec=grpc.Channel)
+        
+        initial_channel = mock.Mock(spec=grpc.Channel)
+        wrapper = _MTLSRefreshingChannel(
+            target="target",
+            factory_args={},
+            initial_channel=initial_channel,
+            initial_cert=b"old_cert"
+        )
+        
+        # Subscribing adds to the initial channel
+        mock_callback = mock.Mock()
+        wrapper.subscribe(mock_callback)
+        initial_channel.subscribe.assert_called_with(mock_callback, try_to_connect=False)
+        
+        wrapper.refresh_logic(1)
+        
+        initial_channel.unsubscribe.assert_called_with(mock_callback)
+        mock_secure_channel.return_value.subscribe.assert_called_with(mock_callback)
+
diff --git a/packages/google-auth/tests_async/transport/test_aiohttp_requests.py b/packages/google-auth/tests_async/transport/test_aiohttp_requests.py
index 1dc5b0025edc..91827c996cef 100644
--- a/packages/google-auth/tests_async/transport/test_aiohttp_requests.py
+++ b/packages/google-auth/tests_async/transport/test_aiohttp_requests.py
@@ -128,12 +128,13 @@ def test_mock_session_unspecified_auto_decompress(self):
         request = aiohttp_requests.Request(http)
         assert request.session == http
 
-    def test_timeout(self):
+    @pytest.mark.asyncio
+    async def test_timeout(self):
         http = mock.create_autospec(
             aiohttp.ClientSession, instance=True, auto_decompress=False
         )
         request = aiohttp_requests.Request(http)
-        request(url="http://example.com", method="GET", timeout=5)
+        await request(url="http://example.com", method="GET", timeout=5)
 
     @pytest.mark.asyncio
     async def test__clone(self):