added test for check_and_mutate

Author: daniel-sanche

google/cloud/bigtable/data/_async/client.py

@@ -1608,7 +1608,7 @@ async def check_and_mutate_row(
             false_case_mutations = [false_case_mutations]
         false_case_list = [m._to_pb() for m in false_case_mutations or []]
 
-        with self._metrics.create_operation(OperationType.CHECK_AND_MUTATE):
+        with self._metrics.create_operation(OperationType.CHECK_AND_MUTATE) as op:
             result = await self.client._gapic_client.check_and_mutate_row(
                 request=CheckAndMutateRowRequest(
                     true_mutations=true_case_list,
@@ -1624,6 +1624,7 @@ async def check_and_mutate_row(
                 ),
                 timeout=operation_timeout,
                 retry=None,
+                metadata=[op.interceptor_metadata],
             )
             return result.predicate_matched
 

tests/system/data/test_metrics_async.py

@@ -134,3 +134,57 @@ async def test_read_modify_write(self, target, temp_rows, handler, cluster_confi
         assert attempt.gfe_latency_ns > 0 and attempt.gfe_latency_ns < attempt.duration_ns
         assert attempt.application_blocking_time_ns == 0
         assert attempt.grpc_throttling_time_ns == 0  # TODO: confirm
+
+    @CrossSync.pytest
+    async def test_check_and_mutate_row(self, target, temp_rows, handler, cluster_config):
+        from google.cloud.bigtable.data.mutations import SetCell
+        from google.cloud.bigtable.data.row_filters import ValueRangeFilter
+
+        row_key = b"test-row-key"
+        family = TEST_FAMILY
+        qualifier = b"test-qualifier"
+        await temp_rows.add_row(
+            row_key, value=1, family=family, qualifier=qualifier
+        )
+
+
+        false_mutation_value = b"false-mutation-value"
+        false_mutation = SetCell(
+            family=TEST_FAMILY, qualifier=qualifier, new_value=false_mutation_value
+        )
+        true_mutation_value = b"true-mutation-value"
+        true_mutation = SetCell(
+            family=TEST_FAMILY, qualifier=qualifier, new_value=true_mutation_value
+        )
+        predicate = ValueRangeFilter(0, 2)
+        await target.check_and_mutate_row(
+            row_key,
+            predicate,
+            true_case_mutations=true_mutation,
+            false_case_mutations=false_mutation,
+        )
+        # 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.value[0] == 0
+        assert operation.is_streaming is False
+        assert operation.op_type.value == "CheckAndMutateRow"
+        assert len(operation.completed_attempts) == 1
+        assert operation.completed_attempts[0] == handler.completed_attempts[0]
+        assert operation.cluster_id == next(iter(cluster_config.keys()))
+        assert operation.zone == cluster_config[operation.cluster_id].location.split("/")[-1]
+        assert operation.duration_ns > 0 and operation.duration_ns < 1e9
+        assert operation.first_response_latency_ns is None  # populated for read_rows only
+        assert operation.flow_throttling_time_ns == 0
+        # validate attempt
+        attempt = handler.completed_attempts[0]
+        assert attempt.duration_ns > 0 and attempt.duration_ns < operation.duration_ns
+        assert attempt.end_status.value[0] == 0
+        assert attempt.backoff_before_attempt_ns == 0
+        assert attempt.gfe_latency_ns > 0 and attempt.gfe_latency_ns < attempt.duration_ns
+        assert attempt.application_blocking_time_ns == 0
+        assert attempt.grpc_throttling_time_ns == 0  # TODO: confirm