added test for streaming retries

Author: daniel-sanche

tests/system/data/test_metrics_async.py

@@ -592,6 +592,48 @@ async def test_read_rows_stream_failure_unauthorized(
             and attempt.gfe_latency_ns < operation.duration_ns
         )
 
+    @CrossSync.pytest
+    async def test_read_rows_stream_failure_unauthorized_with_retries(
+        self, handler, authorized_view, cluster_config
+    ):
+        """
+        retry unauthorized request multiple times before timing out
+        """
+        from google.cloud.bigtable.data.row_filters import FamilyNameRegexFilter
+
+        with pytest.raises(GoogleAPICallError) as e:
+            generator = await authorized_view.read_rows_stream(
+                ReadRowsQuery(row_filter=FamilyNameRegexFilter("unauthorized")),
+                retryable_errors=[PermissionDenied],
+                operation_timeout=0.1,
+            )
+            [_ async for _ in generator]
+        assert e.value.grpc_status_code.name == "DEADLINE_EXCEEDED"
+        # validate counts
+        assert len(handler.completed_operations) == 1
+        assert len(handler.completed_attempts) > 1
+        assert len(handler.cancelled_operations) == 0
+        # validate operation
+        operation = handler.completed_operations[0]
+        assert isinstance(operation, CompletedOperationMetric)
+        assert operation.final_status.name == "DEADLINE_EXCEEDED"
+        assert operation.op_type.value == "ReadRows"
+        assert operation.is_streaming is True
+        assert len(operation.completed_attempts) > 1
+        assert operation.cluster_id == next(iter(cluster_config.keys()))
+        assert (
+            operation.zone
+            == cluster_config[operation.cluster_id].location.split("/")[-1]
+        )
+        # validate attempts
+        for attempt in handler.completed_attempts:
+            assert isinstance(attempt, CompletedAttemptMetric)
+            assert attempt.end_status.name == "PERMISSION_DENIED"
+            assert (
+                attempt.gfe_latency_ns >= 0
+                and attempt.gfe_latency_ns < operation.duration_ns
+            )
+
     @CrossSync.pytest
     async def test_read_rows_stream_failure_mid_stream(
         self, table, temp_rows, handler, error_injector
@@ -1574,7 +1616,7 @@ async def test_mutate_row_failure_unauthorized_with_retries(
         mutation = SetCell("unauthorized", b"q", b"v")
 
         with pytest.raises(GoogleAPICallError) as e:
-            await authorized_view.mutate_row(row_key, [mutation], retryable_errors=[PermissionDenied], operation_timeout=30)
+            await authorized_view.mutate_row(row_key, [mutation], retryable_errors=[PermissionDenied], operation_timeout=0.1)
         assert e.value.grpc_status_code.name == "DEADLINE_EXCEEDED"
         # validate counts
         assert len(handler.completed_operations) == 1