combined channel refresh tests

Author: daniel-sanche

tests/system/data/test_system_async.py

@@ -266,75 +266,49 @@ async def test_ping_and_warm(self, client, target):
     @CrossSync.pytest
     async def test_channel_refresh(self, table_id, instance_id, temp_rows):
         """
-        change grpc channel to refresh quickly, then schedule a read_rows call after refresh
-        to ensure new channel is in place and works
-        """
-        await temp_rows.add_row(b"row_key_1")
-        await temp_rows.add_row(b"row_key_2")
-        async with self._make_client() as client:
-            # start custom refresh task
-            client._channel_refresh_task.cancel()
-            client._channel_refresh_task = CrossSync.create_task(
-                client._manage_channel,
-                refresh_interval_min=0.25,
-                refresh_interval_max=0.25,
-                sync_executor=client._executor,
-            )
-            # let task run
-            await CrossSync.yield_to_event_loop()
-            async with client.get_table(instance_id, table_id) as table:
-                rows = await table.read_rows({})
-                channel_wrapper = client.transport.grpc_channel
-                first_channel = channel_wrapper._channel
-                assert len(rows) == 2
-                await CrossSync.sleep(0.5)
-                rows_after_refresh = await table.read_rows({})
-                assert len(rows_after_refresh) == 2
-                assert client.transport.grpc_channel is channel_wrapper
-                updated_channel = channel_wrapper._channel
-                assert updated_channel is not first_channel
-                # ensure interceptors are kept (gapic's logging interceptor, and metric interceptor)
-                if CrossSync.is_async:
-                    unary_interceptors = updated_channel._unary_unary_interceptors
-                    assert len(unary_interceptors) == 2
-                    assert GapicInterceptor in [type(i) for i in unary_interceptors]
-                    assert client._metrics_interceptor in unary_interceptors
-                    stream_interceptors = updated_channel._unary_stream_interceptors
-                    assert len(stream_interceptors) == 1
-                    assert client._metrics_interceptor in stream_interceptors
-                else:
-                    assert isinstance(
-                        client.transport._logged_channel._interceptor, GapicInterceptor
-                    )
-                    assert updated_channel._interceptor == client._metrics_interceptor
-
-    @CrossSync.pytest
-    async def test_channel_refresh_stress_test(self, table_id, instance_id, temp_rows):
-        """
-        While swapping channels, consistently hit it with reads. Make sure no failures are found
+        perform requests while swapping out the grpc channel. Requests should continue without error
         """
         import time
 
         await temp_rows.add_row(b"test_row")
         async with self._make_client() as client:
             client._channel_refresh_task.cancel()
+            channel_wrapper = client.transport.grpc_channel
+            first_channel = channel_wrapper._channel
             # swap channels frequently, with large grace windows
             client._channel_refresh_task = CrossSync.create_task(
                 client._manage_channel,
                 refresh_interval_min=0.1,
                 refresh_interval_max=0.1,
-                grace_period=0.2,
+                grace_period=1,
                 sync_executor=client._executor,
             )
 
             # hit channels with frequent requests
-            end_time = time.monotonic() + 1
+            end_time = time.monotonic() + 3
             async with client.get_table(instance_id, table_id) as table:
                 while time.monotonic() < end_time:
                     # we expect a CancelledError if a channel is closed before completion
                     rows = await table.read_rows({})
                     assert len(rows) == 1
                     await CrossSync.yield_to_event_loop()
+            # ensure channel was updated
+            updated_channel = channel_wrapper._channel
+            assert updated_channel is not first_channel
+            # ensure interceptors are kept (gapic's logging interceptor, and metric interceptor)
+            if CrossSync.is_async:
+                unary_interceptors = updated_channel._unary_unary_interceptors
+                assert len(unary_interceptors) == 2
+                assert GapicInterceptor in [type(i) for i in unary_interceptors]
+                assert client._metrics_interceptor in unary_interceptors
+                stream_interceptors = updated_channel._unary_stream_interceptors
+                assert len(stream_interceptors) == 1
+                assert client._metrics_interceptor in stream_interceptors
+            else:
+                assert isinstance(
+                    client.transport._logged_channel._interceptor, GapicInterceptor
+                )
+                assert updated_channel._interceptor == client._metrics_interceptor
 
     @CrossSync.pytest
     @pytest.mark.usefixtures("target")

tests/system/data/test_system_autogen.py

@@ -221,55 +221,33 @@ def test_ping_and_warm(self, client, target):
         reason="emulator mode doesn't refresh channel",
     )
     def test_channel_refresh(self, table_id, instance_id, temp_rows):
-        """change grpc channel to refresh quickly, then schedule a read_rows call after refresh
-        to ensure new channel is in place and works"""
-        temp_rows.add_row(b"row_key_1")
-        temp_rows.add_row(b"row_key_2")
-        with self._make_client() as client:
-            client._channel_refresh_task.cancel()
-            client._channel_refresh_task = CrossSync._Sync_Impl.create_task(
-                client._manage_channel,
-                refresh_interval_min=0.25,
-                refresh_interval_max=0.25,
-                sync_executor=client._executor,
-            )
-            CrossSync._Sync_Impl.yield_to_event_loop()
-            with client.get_table(instance_id, table_id) as table:
-                rows = table.read_rows({})
-                channel_wrapper = client.transport.grpc_channel
-                first_channel = channel_wrapper._channel
-                assert len(rows) == 2
-                CrossSync._Sync_Impl.sleep(0.5)
-                rows_after_refresh = table.read_rows({})
-                assert len(rows_after_refresh) == 2
-                assert client.transport.grpc_channel is channel_wrapper
-                updated_channel = channel_wrapper._channel
-                assert updated_channel is not first_channel
-                assert isinstance(
-                    client.transport._logged_channel._interceptor, GapicInterceptor
-                )
-                assert updated_channel._interceptor == client._metrics_interceptor
-
-    def test_channel_refresh_stress_test(self, table_id, instance_id, temp_rows):
-        """While swapping channels, consistently hit it with reads. Make sure no failures are found"""
+        """perform requests while swapping out the grpc channel. Requests should continue without error"""
         import time
 
         temp_rows.add_row(b"test_row")
         with self._make_client() as client:
             client._channel_refresh_task.cancel()
+            channel_wrapper = client.transport.grpc_channel
+            first_channel = channel_wrapper._channel
             client._channel_refresh_task = CrossSync._Sync_Impl.create_task(
                 client._manage_channel,
                 refresh_interval_min=0.1,
                 refresh_interval_max=0.1,
-                grace_period=0.2,
+                grace_period=1,
                 sync_executor=client._executor,
             )
-            end_time = time.monotonic() + 1
+            end_time = time.monotonic() + 3
             with client.get_table(instance_id, table_id) as table:
                 while time.monotonic() < end_time:
                     rows = table.read_rows({})
                     assert len(rows) == 1
                     CrossSync._Sync_Impl.yield_to_event_loop()
+            updated_channel = channel_wrapper._channel
+            assert updated_channel is not first_channel
+            assert isinstance(
+                client.transport._logged_channel._interceptor, GapicInterceptor
+            )
+            assert updated_channel._interceptor == client._metrics_interceptor
 
     @pytest.mark.usefixtures("target")
     @CrossSync._Sync_Impl.Retry(