test_system_autogen.py

# Copyright 2024 Google LLC
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.


# This file is automatically generated by CrossSync. Do not edit manually.

import datetime
import os
import uuid

import pytest
from google.api_core import retry
from google.api_core.exceptions import ClientError, PermissionDenied
from google.cloud.environment_vars import BIGTABLE_EMULATOR
from google.type import date_pb2

from google.cloud.bigtable.data._cross_sync import CrossSync
from google.cloud.bigtable.data.execute_query.metadata import SqlType
from google.cloud.bigtable.data.read_modify_write_rules import _MAX_INCREMENT_VALUE
from google.cloud.bigtable_v2.services.bigtable.transports.grpc import (
    _LoggingClientInterceptor as GapicInterceptor,
)

from . import TEST_AGGREGATE_FAMILY, TEST_FAMILY, TEST_FAMILY_2, SystemTestRunner

TARGETS = ["table"]
if not os.environ.get(BIGTABLE_EMULATOR):
    TARGETS.append("authorized_view")


@CrossSync._Sync_Impl.add_mapping_decorator("TempRowBuilder")
class TempRowBuilder:
    """
    Used to add rows to a table for testing purposes.
    """

    def __init__(self, target):
        self.rows = []
        self.target = target

    def add_row(
        self, row_key, *, family=TEST_FAMILY, qualifier=b"q", value=b"test-value"
    ):
        if isinstance(value, str):
            value = value.encode("utf-8")
        elif isinstance(value, int):
            value = value.to_bytes(8, byteorder="big", signed=True)
        request = {
            "table_name": self.target.table_name,
            "row_key": row_key,
            "mutations": [
                {
                    "set_cell": {
                        "family_name": family,
                        "column_qualifier": qualifier,
                        "value": value,
                    }
                }
            ],
        }
        self.target.client._gapic_client.mutate_row(request)
        self.rows.append(row_key)

    def add_aggregate_row(
        self, row_key, *, family=TEST_AGGREGATE_FAMILY, qualifier=b"q", input=0
    ):
        request = {
            "table_name": self.target.table_name,
            "row_key": row_key,
            "mutations": [
                {
                    "add_to_cell": {
                        "family_name": family,
                        "column_qualifier": {"raw_value": qualifier},
                        "timestamp": {"raw_timestamp_micros": 0},
                        "input": {"int_value": input},
                    }
                }
            ],
        }
        self.target.client._gapic_client.mutate_row(request)
        self.rows.append(row_key)

    def delete_rows(self):
        if self.rows:
            chunk_size = 5000
            for i in range(0, len(self.rows), chunk_size):
                chunk = self.rows[i : i + chunk_size]
                request = {
                    **self.target._request_path,
                    "entries": [
                        {"row_key": row, "mutations": [{"delete_from_row": {}}]}
                        for row in chunk
                    ],
                }
                stream = self.target.client._gapic_client.mutate_rows(request)
                for response in stream:
                    pass

    def retrieve_cell_value(self, target, row_key):
        """Helper to read an individual row"""
        from google.cloud.bigtable.data import ReadRowsQuery

        row_list = target.read_rows(ReadRowsQuery(row_keys=row_key))
        assert len(row_list) == 1
        row = row_list[0]
        cell = row.cells[0]
        return cell.value

    def create_row_and_mutation(
        self, table, *, start_value=b"start", new_value=b"new_value"
    ):
        """Helper to create a new row, and a sample set_cell mutation to change its value"""
        from google.cloud.bigtable.data.mutations import SetCell

        row_key = uuid.uuid4().hex.encode()
        family = TEST_FAMILY
        qualifier = b"test-qualifier"
        self.add_row(row_key, family=family, qualifier=qualifier, value=start_value)
        assert self.retrieve_cell_value(table, row_key) == start_value
        mutation = SetCell(family=TEST_FAMILY, qualifier=qualifier, new_value=new_value)
        return (row_key, mutation)


class TestSystem(SystemTestRunner):
    def _make_client(self):
        project = os.getenv("GOOGLE_CLOUD_PROJECT") or None
        return CrossSync._Sync_Impl.DataClient(project=project)

    @pytest.fixture(scope="session")
    def client(self):
        with self._make_client() as client:
            yield client

    @pytest.fixture(scope="session", params=TARGETS)
    def target(self, client, table_id, authorized_view_id, instance_id, request):
        """This fixture runs twice: once for a standard table, and once with an authorized view

        Note: emulator doesn't support authorized views. Only use target"""
        if request.param == "table":
            with client.get_table(instance_id, table_id) as table:
                yield table
        elif request.param == "authorized_view":
            with client.get_authorized_view(
                instance_id, table_id, authorized_view_id
            ) as view:
                yield view
        else:
            raise ValueError(f"unknown target type: {request.param}")

    @pytest.fixture(scope="function")
    def temp_rows(self, target):
        builder = CrossSync._Sync_Impl.TempRowBuilder(target)
        yield builder
        builder.delete_rows()

    @pytest.mark.usefixtures("target")
    @pytest.mark.usefixtures("client")
    @CrossSync._Sync_Impl.Retry(
        predicate=retry.if_exception_type(ClientError), initial=1, maximum=10
    )
    def test_ping_and_warm_gapic(self, client, target):
        """Simple ping rpc test
        This test ensures channels are able to authenticate with backend"""
        request = {"name": target.instance_name}
        client._gapic_client.ping_and_warm(request)

    @pytest.mark.usefixtures("target")
    @pytest.mark.usefixtures("client")
    @CrossSync._Sync_Impl.Retry(
        predicate=retry.if_exception_type(ClientError), initial=1, maximum=5
    )
    def test_ping_and_warm(self, client, target):
        """Test ping and warm from handwritten client"""
        results = client._ping_and_warm_instances()
        assert len(results) == 1
        assert results[0] is None

    @pytest.mark.skipif(
        bool(os.environ.get(BIGTABLE_EMULATOR)),
        reason="emulator mode doesn't refresh channel",
    )
    def test_channel_refresh(self, table_id, instance_id, temp_rows):
        """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=1,
                sync_executor=client._executor,
            )
            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(
        predicate=retry.if_exception_type(ClientError), initial=1, maximum=5
    )
    def test_mutation_set_cell(self, target, temp_rows):
        """Ensure cells can be set properly"""
        row_key = b"bulk_mutate"
        new_value = uuid.uuid4().hex.encode()
        row_key, mutation = temp_rows.create_row_and_mutation(
            target, new_value=new_value
        )
        target.mutate_row(row_key, mutation)
        assert temp_rows.retrieve_cell_value(target, row_key) == new_value

    @pytest.mark.usefixtures("target")
    @CrossSync._Sync_Impl.Retry(
        predicate=retry.if_exception_type(ClientError), initial=1, maximum=5
    )
    def test_mutation_add_to_cell(self, target, temp_rows):
        """Test add to cell mutation"""
        from google.cloud.bigtable.data.mutations import AddToCell

        row_key = b"add_to_cell"
        family = TEST_AGGREGATE_FAMILY
        qualifier = b"test-qualifier"
        temp_rows.add_aggregate_row(row_key, family=family, qualifier=qualifier)
        target.mutate_row(row_key, AddToCell(family, qualifier, 1, timestamp_micros=0))
        encoded_result = temp_rows.retrieve_cell_value(target, row_key)
        int_result = int.from_bytes(encoded_result, byteorder="big")
        assert int_result == 1
        target.mutate_row(row_key, AddToCell(family, qualifier, 9, timestamp_micros=0))
        encoded_result = temp_rows.retrieve_cell_value(target, row_key)
        int_result = int.from_bytes(encoded_result, byteorder="big")
        assert int_result == 10

    @pytest.mark.skipif(
        bool(os.environ.get(BIGTABLE_EMULATOR)), reason="emulator doesn't use splits"
    )
    @pytest.mark.usefixtures("client")
    @pytest.mark.usefixtures("target")
    @CrossSync._Sync_Impl.Retry(
        predicate=retry.if_exception_type(ClientError), initial=1, maximum=5
    )
    def test_sample_row_keys(self, client, target, temp_rows, column_split_config):
        """Sample keys should return a single sample in small test targets"""
        temp_rows.add_row(b"row_key_1")
        temp_rows.add_row(b"row_key_2")
        results = target.sample_row_keys()
        assert len(results) == len(column_split_config) + 1
        for idx in range(len(column_split_config)):
            assert results[idx][0] == column_split_config[idx]
            assert isinstance(results[idx][1], int)
        assert results[-1][0] == b""
        assert isinstance(results[-1][1], int)

    @pytest.mark.skipif(
        bool(os.environ.get(BIGTABLE_EMULATOR)), reason="emulator doesn't use splits"
    )
    @pytest.mark.usefixtures("client")
    @pytest.mark.usefixtures("target")
    @CrossSync._Sync_Impl.Retry(
        predicate=retry.if_exception_type(ClientError), initial=1, maximum=5
    )
    def test_sample_row_keys_w_row_range(self, client, target, column_split_config):
        """Sample keys with row range should return samples within the range,
        with the last key matching the end of the range."""
        if len(column_split_config) < 4:
            pytest.skip("Not enough splits in column_split_config for this test")
        from google.cloud.bigtable.data import RowRange

        start_key = column_split_config[1]
        end_key = column_split_config[3]
        row_range = RowRange(start_key=start_key, end_key=end_key)
        results = target.sample_row_keys(row_range=row_range)
        assert len(results) == 2
        assert results[0][0] == column_split_config[2]
        assert results[1][0] == column_split_config[3]
        for _, offset in results:
            assert isinstance(offset, int)

    @pytest.mark.usefixtures("client")
    @pytest.mark.usefixtures("target")
    def test_bulk_mutations_set_cell(self, client, target, temp_rows):
        """Ensure cells can be set properly"""
        from google.cloud.bigtable.data.mutations import RowMutationEntry

        new_value = uuid.uuid4().hex.encode()
        row_key, mutation = temp_rows.create_row_and_mutation(
            target, new_value=new_value
        )
        bulk_mutation = RowMutationEntry(row_key, [mutation])
        target.bulk_mutate_rows([bulk_mutation])
        assert temp_rows.retrieve_cell_value(target, row_key) == new_value

    def test_bulk_mutations_raise_exception(self, client, target):
        """If an invalid mutation is passed, an exception should be raised"""
        from google.cloud.bigtable.data.exceptions import (
            FailedMutationEntryError,
            MutationsExceptionGroup,
        )
        from google.cloud.bigtable.data.mutations import RowMutationEntry, SetCell

        row_key = uuid.uuid4().hex.encode()
        mutation = SetCell(
            family="nonexistent", qualifier=b"test-qualifier", new_value=b""
        )
        bulk_mutation = RowMutationEntry(row_key, [mutation])
        with pytest.raises(MutationsExceptionGroup) as exc:
            target.bulk_mutate_rows([bulk_mutation])
        assert len(exc.value.exceptions) == 1
        entry_error = exc.value.exceptions[0]
        assert isinstance(entry_error, FailedMutationEntryError)
        assert entry_error.index == 0
        assert entry_error.entry == bulk_mutation

    @pytest.mark.usefixtures("client")
    @pytest.mark.usefixtures("target")
    @CrossSync._Sync_Impl.Retry(
        predicate=retry.if_exception_type(ClientError), initial=1, maximum=5
    )
    def test_mutations_batcher_context_manager(self, client, target, temp_rows):
        """test batcher with context manager. Should flush on exit"""
        from google.cloud.bigtable.data.mutations import RowMutationEntry

        new_value, new_value2 = [uuid.uuid4().hex.encode() for _ in range(2)]
        row_key, mutation = temp_rows.create_row_and_mutation(
            target, new_value=new_value
        )
        row_key2, mutation2 = temp_rows.create_row_and_mutation(
            target, new_value=new_value2
        )
        bulk_mutation = RowMutationEntry(row_key, [mutation])
        bulk_mutation2 = RowMutationEntry(row_key2, [mutation2])
        with target.mutations_batcher() as batcher:
            batcher.append(bulk_mutation)
            batcher.append(bulk_mutation2)
        assert temp_rows.retrieve_cell_value(target, row_key) == new_value
        assert len(batcher._staged_entries) == 0

    @pytest.mark.usefixtures("client")
    @pytest.mark.usefixtures("target")
    @CrossSync._Sync_Impl.Retry(
        predicate=retry.if_exception_type(ClientError), initial=1, maximum=5
    )
    def test_mutations_batcher_timer_flush(self, client, target, temp_rows):
        """batch should occur after flush_interval seconds"""
        from google.cloud.bigtable.data.mutations import RowMutationEntry

        new_value = uuid.uuid4().hex.encode()
        row_key, mutation = temp_rows.create_row_and_mutation(
            target, new_value=new_value
        )
        bulk_mutation = RowMutationEntry(row_key, [mutation])
        flush_interval = 0.1
        with target.mutations_batcher(flush_interval=flush_interval) as batcher:
            batcher.append(bulk_mutation)
            CrossSync._Sync_Impl.yield_to_event_loop()
            assert len(batcher._staged_entries) == 1
            CrossSync._Sync_Impl.sleep(flush_interval + 0.1)
            assert len(batcher._staged_entries) == 0
            assert temp_rows.retrieve_cell_value(target, row_key) == new_value

    @pytest.mark.usefixtures("client")
    @pytest.mark.usefixtures("target")
    @CrossSync._Sync_Impl.Retry(
        predicate=retry.if_exception_type(ClientError), initial=1, maximum=5
    )
    def test_mutations_batcher_count_flush(self, client, target, temp_rows):
        """batch should flush after flush_limit_mutation_count mutations"""
        from google.cloud.bigtable.data.mutations import RowMutationEntry

        new_value, new_value2 = [uuid.uuid4().hex.encode() for _ in range(2)]
        row_key, mutation = temp_rows.create_row_and_mutation(
            target, new_value=new_value
        )
        bulk_mutation = RowMutationEntry(row_key, [mutation])
        row_key2, mutation2 = temp_rows.create_row_and_mutation(
            target, new_value=new_value2
        )
        bulk_mutation2 = RowMutationEntry(row_key2, [mutation2])
        with target.mutations_batcher(flush_limit_mutation_count=2) as batcher:
            batcher.append(bulk_mutation)
            assert len(batcher._flush_jobs) == 0
            assert len(batcher._staged_entries) == 1
            batcher.append(bulk_mutation2)
            assert len(batcher._flush_jobs) == 1
            for future in list(batcher._flush_jobs):
                future
                future.result()
            assert len(batcher._staged_entries) == 0
            assert len(batcher._flush_jobs) == 0
            assert temp_rows.retrieve_cell_value(target, row_key) == new_value
            assert temp_rows.retrieve_cell_value(target, row_key2) == new_value2

    @pytest.mark.usefixtures("client")
    @pytest.mark.usefixtures("target")
    @CrossSync._Sync_Impl.Retry(
        predicate=retry.if_exception_type(ClientError), initial=1, maximum=5
    )
    def test_mutations_batcher_bytes_flush(self, client, target, temp_rows):
        """batch should flush after flush_limit_bytes bytes"""
        from google.cloud.bigtable.data.mutations import RowMutationEntry

        new_value, new_value2 = [uuid.uuid4().hex.encode() for _ in range(2)]
        row_key, mutation = temp_rows.create_row_and_mutation(
            target, new_value=new_value
        )
        bulk_mutation = RowMutationEntry(row_key, [mutation])
        row_key2, mutation2 = temp_rows.create_row_and_mutation(
            target, new_value=new_value2
        )
        bulk_mutation2 = RowMutationEntry(row_key2, [mutation2])
        flush_limit = bulk_mutation.size() + bulk_mutation2.size() - 1
        with target.mutations_batcher(flush_limit_bytes=flush_limit) as batcher:
            batcher.append(bulk_mutation)
            assert len(batcher._flush_jobs) == 0
            assert len(batcher._staged_entries) == 1
            batcher.append(bulk_mutation2)
            assert len(batcher._flush_jobs) == 1
            assert len(batcher._staged_entries) == 0
            for future in list(batcher._flush_jobs):
                future
                future.result()
            assert temp_rows.retrieve_cell_value(target, row_key) == new_value
            assert temp_rows.retrieve_cell_value(target, row_key2) == new_value2

    @pytest.mark.usefixtures("client")
    @pytest.mark.usefixtures("target")
    def test_mutations_batcher_no_flush(self, client, target, temp_rows):
        """test with no flush requirements met"""
        from google.cloud.bigtable.data.mutations import RowMutationEntry

        new_value = uuid.uuid4().hex.encode()
        start_value = b"unchanged"
        row_key, mutation = temp_rows.create_row_and_mutation(
            target, start_value=start_value, new_value=new_value
        )
        bulk_mutation = RowMutationEntry(row_key, [mutation])
        row_key2, mutation2 = temp_rows.create_row_and_mutation(
            target, start_value=start_value, new_value=new_value
        )
        bulk_mutation2 = RowMutationEntry(row_key2, [mutation2])
        size_limit = bulk_mutation.size() + bulk_mutation2.size() + 1
        with target.mutations_batcher(
            flush_limit_bytes=size_limit, flush_limit_mutation_count=3, flush_interval=1
        ) as batcher:
            batcher.append(bulk_mutation)
            assert len(batcher._staged_entries) == 1
            batcher.append(bulk_mutation2)
            assert len(batcher._flush_jobs) == 0
            CrossSync._Sync_Impl.yield_to_event_loop()
            assert len(batcher._staged_entries) == 2
            assert len(batcher._flush_jobs) == 0
            assert temp_rows.retrieve_cell_value(target, row_key) == start_value
            assert temp_rows.retrieve_cell_value(target, row_key2) == start_value

    @pytest.mark.usefixtures("client")
    @pytest.mark.usefixtures("target")
    @CrossSync._Sync_Impl.Retry(
        predicate=retry.if_exception_type(ClientError), initial=1, maximum=5
    )
    def test_mutations_batcher_large_batch(self, client, target, temp_rows):
        """test batcher with large batch of mutations"""
        from google.cloud.bigtable.data.mutations import RowMutationEntry, SetCell

        add_mutation = SetCell(
            family=TEST_FAMILY, qualifier=b"test-qualifier", new_value=b"a"
        )
        row_mutations = []
        for i in range(50000):
            row_key = uuid.uuid4().hex.encode()
            row_mutations.append(RowMutationEntry(row_key, [add_mutation]))
            temp_rows.rows.append(row_key)
        with target.mutations_batcher() as batcher:
            for mutation in row_mutations:
                batcher.append(mutation)
        assert len(batcher._staged_entries) == 0

    @pytest.mark.usefixtures("client")
    @pytest.mark.usefixtures("target")
    @pytest.mark.parametrize(
        "start,increment,expected",
        [
            (0, 0, 0),
            (0, 1, 1),
            (0, -1, -1),
            (1, 0, 1),
            (0, -100, -100),
            (0, 3000, 3000),
            (10, 4, 14),
            (_MAX_INCREMENT_VALUE, -_MAX_INCREMENT_VALUE, 0),
            (_MAX_INCREMENT_VALUE, 2, -_MAX_INCREMENT_VALUE),
            (-_MAX_INCREMENT_VALUE, -2, _MAX_INCREMENT_VALUE),
        ],
    )
    def test_read_modify_write_row_increment(
        self, client, target, temp_rows, start, increment, expected
    ):
        """test read_modify_write_row"""
        from google.cloud.bigtable.data.read_modify_write_rules import IncrementRule

        row_key = b"test-row-key"
        family = TEST_FAMILY
        qualifier = b"test-qualifier"
        temp_rows.add_row(row_key, value=start, family=family, qualifier=qualifier)
        rule = IncrementRule(family, qualifier, increment)
        result = target.read_modify_write_row(row_key, rule)
        assert result.row_key == row_key
        assert len(result) == 1
        assert result[0].family == family
        assert result[0].qualifier == qualifier
        assert int(result[0]) == expected
        assert temp_rows.retrieve_cell_value(target, row_key) == result[0].value

    @pytest.mark.usefixtures("client")
    @pytest.mark.usefixtures("target")
    @pytest.mark.parametrize(
        "start,append,expected",
        [
            (b"", b"", b""),
            ("", "", b""),
            (b"abc", b"123", b"abc123"),
            (b"abc", "123", b"abc123"),
            ("", b"1", b"1"),
            (b"abc", "", b"abc"),
            (b"hello", b"world", b"helloworld"),
        ],
    )
    def test_read_modify_write_row_append(
        self, client, target, temp_rows, start, append, expected
    ):
        """test read_modify_write_row"""
        from google.cloud.bigtable.data.read_modify_write_rules import AppendValueRule

        row_key = b"test-row-key"
        family = TEST_FAMILY
        qualifier = b"test-qualifier"
        temp_rows.add_row(row_key, value=start, family=family, qualifier=qualifier)
        rule = AppendValueRule(family, qualifier, append)
        result = target.read_modify_write_row(row_key, rule)
        assert result.row_key == row_key
        assert len(result) == 1
        assert result[0].family == family
        assert result[0].qualifier == qualifier
        assert result[0].value == expected
        assert temp_rows.retrieve_cell_value(target, row_key) == result[0].value

    @pytest.mark.usefixtures("client")
    @pytest.mark.usefixtures("target")
    def test_read_modify_write_row_chained(self, client, target, temp_rows):
        """test read_modify_write_row with multiple rules"""
        from google.cloud.bigtable.data.read_modify_write_rules import (
            AppendValueRule,
            IncrementRule,
        )

        row_key = b"test-row-key"
        family = TEST_FAMILY
        qualifier = b"test-qualifier"
        start_amount = 1
        increment_amount = 10
        temp_rows.add_row(
            row_key, value=start_amount, family=family, qualifier=qualifier
        )
        rule = [
            IncrementRule(family, qualifier, increment_amount),
            AppendValueRule(family, qualifier, "hello"),
            AppendValueRule(family, qualifier, "world"),
            AppendValueRule(family, qualifier, "!"),
        ]
        result = target.read_modify_write_row(row_key, rule)
        assert result.row_key == row_key
        assert result[0].family == family
        assert result[0].qualifier == qualifier
        assert (
            result[0].value
            == (start_amount + increment_amount).to_bytes(8, "big", signed=True)
            + b"helloworld!"
        )
        assert temp_rows.retrieve_cell_value(target, row_key) == result[0].value

    @pytest.mark.usefixtures("client")
    @pytest.mark.usefixtures("target")
    @pytest.mark.parametrize(
        "start_val,predicate_range,expected_result",
        [(1, (0, 2), True), (-1, (0, 2), False)],
    )
    def test_check_and_mutate(
        self, client, target, temp_rows, start_val, predicate_range, expected_result
    ):
        """test that check_and_mutate_row works applies the right mutations, and returns the right result"""
        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"
        temp_rows.add_row(row_key, value=start_val, 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(predicate_range[0], predicate_range[1])
        result = target.check_and_mutate_row(
            row_key,
            predicate,
            true_case_mutations=true_mutation,
            false_case_mutations=false_mutation,
        )
        assert result == expected_result
        expected_value = (
            true_mutation_value if expected_result else false_mutation_value
        )
        assert temp_rows.retrieve_cell_value(target, row_key) == expected_value

    @pytest.mark.skipif(
        bool(os.environ.get(BIGTABLE_EMULATOR)),
        reason="emulator doesn't raise InvalidArgument",
    )
    @pytest.mark.usefixtures("client")
    @pytest.mark.usefixtures("target")
    def test_check_and_mutate_empty_request(self, client, target):
        """check_and_mutate with no true or fale mutations should raise an error"""
        from google.api_core import exceptions

        with pytest.raises(exceptions.InvalidArgument) as e:
            target.check_and_mutate_row(
                b"row_key", None, true_case_mutations=None, false_case_mutations=None
            )
        assert "No mutations provided" in str(e.value)

    @pytest.mark.usefixtures("target")
    @CrossSync._Sync_Impl.Retry(
        predicate=retry.if_exception_type(ClientError), initial=1, maximum=5
    )
    def test_read_rows_stream(self, target, temp_rows):
        """Ensure that the read_rows_stream method works"""
        temp_rows.add_row(b"row_key_1")
        temp_rows.add_row(b"row_key_2")
        generator = target.read_rows_stream({})
        first_row = generator.__next__()
        second_row = generator.__next__()
        assert first_row.row_key == b"row_key_1"
        assert second_row.row_key == b"row_key_2"
        with pytest.raises(CrossSync._Sync_Impl.StopIteration):
            generator.__next__()

    @pytest.mark.usefixtures("target")
    @CrossSync._Sync_Impl.Retry(
        predicate=retry.if_exception_type(ClientError), initial=1, maximum=5
    )
    def test_read_rows(self, target, temp_rows):
        """Ensure that the read_rows method works"""
        temp_rows.add_row(b"row_key_1")
        temp_rows.add_row(b"row_key_2")
        row_list = target.read_rows({})
        assert len(row_list) == 2
        assert row_list[0].row_key == b"row_key_1"
        assert row_list[1].row_key == b"row_key_2"

    @pytest.mark.usefixtures("target")
    @CrossSync._Sync_Impl.Retry(
        predicate=retry.if_exception_type(ClientError), initial=1, maximum=5
    )
    def test_read_rows_sharded_simple(self, target, temp_rows):
        """Test read rows sharded with two queries"""
        from google.cloud.bigtable.data.read_rows_query import ReadRowsQuery

        temp_rows.add_row(b"a")
        temp_rows.add_row(b"b")
        temp_rows.add_row(b"c")
        temp_rows.add_row(b"d")
        query1 = ReadRowsQuery(row_keys=[b"a", b"c"])
        query2 = ReadRowsQuery(row_keys=[b"b", b"d"])
        row_list = target.read_rows_sharded([query1, query2])
        assert len(row_list) == 4
        assert row_list[0].row_key == b"a"
        assert row_list[1].row_key == b"c"
        assert row_list[2].row_key == b"b"
        assert row_list[3].row_key == b"d"

    @pytest.mark.usefixtures("target")
    @CrossSync._Sync_Impl.Retry(
        predicate=retry.if_exception_type(ClientError), initial=1, maximum=5
    )
    def test_read_rows_sharded_from_sample(self, target, temp_rows):
        """Test end-to-end sharding"""
        from google.cloud.bigtable.data.read_rows_query import ReadRowsQuery, RowRange

        temp_rows.add_row(b"a")
        temp_rows.add_row(b"b")
        temp_rows.add_row(b"c")
        temp_rows.add_row(b"d")
        table_shard_keys = target.sample_row_keys()
        query = ReadRowsQuery(row_ranges=[RowRange(start_key=b"b", end_key=b"z")])
        shard_queries = query.shard(table_shard_keys)
        row_list = target.read_rows_sharded(shard_queries)
        assert len(row_list) == 3
        assert row_list[0].row_key == b"b"
        assert row_list[1].row_key == b"c"
        assert row_list[2].row_key == b"d"

    @pytest.mark.usefixtures("target")
    @CrossSync._Sync_Impl.Retry(
        predicate=retry.if_exception_type(ClientError), initial=1, maximum=5
    )
    def test_read_rows_sharded_filters_limits(self, target, temp_rows):
        """Test read rows sharded with filters and limits"""
        from google.cloud.bigtable.data.read_rows_query import ReadRowsQuery
        from google.cloud.bigtable.data.row_filters import ApplyLabelFilter

        temp_rows.add_row(b"a")
        temp_rows.add_row(b"b")
        temp_rows.add_row(b"c")
        temp_rows.add_row(b"d")
        label_filter1 = ApplyLabelFilter("first")
        label_filter2 = ApplyLabelFilter("second")
        query1 = ReadRowsQuery(row_keys=[b"a", b"c"], limit=1, row_filter=label_filter1)
        query2 = ReadRowsQuery(row_keys=[b"b", b"d"], row_filter=label_filter2)
        row_list = target.read_rows_sharded([query1, query2])
        assert len(row_list) == 3
        assert row_list[0].row_key == b"a"
        assert row_list[1].row_key == b"b"
        assert row_list[2].row_key == b"d"
        assert row_list[0][0].labels == ["first"]
        assert row_list[1][0].labels == ["second"]
        assert row_list[2][0].labels == ["second"]

    @pytest.mark.usefixtures("target")
    @CrossSync._Sync_Impl.Retry(
        predicate=retry.if_exception_type(ClientError), initial=1, maximum=5
    )
    def test_read_rows_range_query(self, target, temp_rows):
        """Ensure that the read_rows method works"""
        from google.cloud.bigtable.data import ReadRowsQuery, RowRange

        temp_rows.add_row(b"a")
        temp_rows.add_row(b"b")
        temp_rows.add_row(b"c")
        temp_rows.add_row(b"d")
        query = ReadRowsQuery(row_ranges=RowRange(start_key=b"b", end_key=b"d"))
        row_list = target.read_rows(query)
        assert len(row_list) == 2
        assert row_list[0].row_key == b"b"
        assert row_list[1].row_key == b"c"

    @pytest.mark.usefixtures("target")
    @CrossSync._Sync_Impl.Retry(
        predicate=retry.if_exception_type(ClientError), initial=1, maximum=5
    )
    def test_read_rows_single_key_query(self, target, temp_rows):
        """Ensure that the read_rows method works with specified query"""
        from google.cloud.bigtable.data import ReadRowsQuery

        temp_rows.add_row(b"a")
        temp_rows.add_row(b"b")
        temp_rows.add_row(b"c")
        temp_rows.add_row(b"d")
        query = ReadRowsQuery(row_keys=[b"a", b"c"])
        row_list = target.read_rows(query)
        assert len(row_list) == 2
        assert row_list[0].row_key == b"a"
        assert row_list[1].row_key == b"c"

    @pytest.mark.usefixtures("target")
    @CrossSync._Sync_Impl.Retry(
        predicate=retry.if_exception_type(ClientError), initial=1, maximum=5
    )
    def test_read_rows_with_filter(self, target, temp_rows):
        """ensure filters are applied"""
        from google.cloud.bigtable.data import ReadRowsQuery
        from google.cloud.bigtable.data.row_filters import ApplyLabelFilter

        temp_rows.add_row(b"a")
        temp_rows.add_row(b"b")
        temp_rows.add_row(b"c")
        temp_rows.add_row(b"d")
        expected_label = "test-label"
        row_filter = ApplyLabelFilter(expected_label)
        query = ReadRowsQuery(row_filter=row_filter)
        row_list = target.read_rows(query)
        assert len(row_list) == 4
        for row in row_list:
            assert row[0].labels == [expected_label]

    @pytest.mark.usefixtures("target")
    def test_read_rows_stream_close(self, target, temp_rows):
        """Ensure that the read_rows_stream can be closed"""
        from google.cloud.bigtable.data import ReadRowsQuery

        temp_rows.add_row(b"row_key_1")
        temp_rows.add_row(b"row_key_2")
        query = ReadRowsQuery()
        generator = target.read_rows_stream(query)
        first_row = generator.__next__()
        assert first_row.row_key == b"row_key_1"
        generator.close()
        with pytest.raises(CrossSync._Sync_Impl.StopIteration):
            generator.__next__()

    @pytest.mark.usefixtures("target")
    def test_read_row(self, target, temp_rows):
        """Test read_row (single row helper)"""
        from google.cloud.bigtable.data import Row

        temp_rows.add_row(b"row_key_1", value=b"value")
        row = target.read_row(b"row_key_1")
        assert isinstance(row, Row)
        assert row.row_key == b"row_key_1"
        assert row.cells[0].value == b"value"

    @pytest.mark.skipif(
        bool(os.environ.get(BIGTABLE_EMULATOR)),
        reason="emulator doesn't raise InvalidArgument",
    )
    @pytest.mark.usefixtures("target")
    def test_read_row_missing(self, target):
        """Test read_row when row does not exist"""
        from google.api_core import exceptions

        row_key = "row_key_not_exist"
        result = target.read_row(row_key)
        assert result is None
        with pytest.raises(exceptions.InvalidArgument) as e:
            target.read_row("")
        assert "Row keys must be non-empty" in str(e)

    @pytest.mark.usefixtures("target")
    def test_read_row_w_filter(self, target, temp_rows):
        """Test read_row (single row helper)"""
        from google.cloud.bigtable.data import Row
        from google.cloud.bigtable.data.row_filters import ApplyLabelFilter

        temp_rows.add_row(b"row_key_1", value=b"value")
        expected_label = "test-label"
        label_filter = ApplyLabelFilter(expected_label)
        row = target.read_row(b"row_key_1", row_filter=label_filter)
        assert isinstance(row, Row)
        assert row.row_key == b"row_key_1"
        assert row.cells[0].value == b"value"
        assert row.cells[0].labels == [expected_label]

    @pytest.mark.skipif(
        bool(os.environ.get(BIGTABLE_EMULATOR)),
        reason="emulator doesn't raise InvalidArgument",
    )
    @pytest.mark.usefixtures("target")
    def test_row_exists(self, target, temp_rows):
        from google.api_core import exceptions

        "Test row_exists with rows that exist and don't exist"
        assert target.row_exists(b"row_key_1") is False
        temp_rows.add_row(b"row_key_1")
        assert target.row_exists(b"row_key_1") is True
        assert target.row_exists("row_key_1") is True
        assert target.row_exists(b"row_key_2") is False
        assert target.row_exists("row_key_2") is False
        assert target.row_exists("3") is False
        temp_rows.add_row(b"3")
        assert target.row_exists(b"3") is True
        with pytest.raises(exceptions.InvalidArgument) as e:
            target.row_exists("")
        assert "Row keys must be non-empty" in str(e)

    @pytest.mark.usefixtures("target")
    @CrossSync._Sync_Impl.Retry(
        predicate=retry.if_exception_type(ClientError), initial=1, maximum=5
    )
    @pytest.mark.parametrize(
        "cell_value,filter_input,expect_match",
        [
            (b"abc", b"abc", True),
            (b"abc", "abc", True),
            (b".", ".", True),
            (".*", ".*", True),
            (".*", b".*", True),
            ("a", ".*", False),
            (b".*", b".*", True),
            ("\\a", "\\a", True),
            (b"\xe2\x98\x83", "☃", True),
            ("☃", "☃", True),
            ("\\C☃", "\\C☃", True),
            (1, 1, True),
            (2, 1, False),
            (68, 68, True),
            ("D", 68, False),
            (68, "D", False),
            (-1, -1, True),
            (2852126720, 2852126720, True),
            (-1431655766, -1431655766, True),
            (-1431655766, -1, False),
        ],
    )
    def test_literal_value_filter(
        self, target, temp_rows, cell_value, filter_input, expect_match
    ):
        """Literal value filter does complex escaping on re2 strings.
        Make sure inputs are properly interpreted by the server"""
        from google.cloud.bigtable.data import ReadRowsQuery
        from google.cloud.bigtable.data.row_filters import LiteralValueFilter

        f = LiteralValueFilter(filter_input)
        temp_rows.add_row(b"row_key_1", value=cell_value)
        query = ReadRowsQuery(row_filter=f)
        row_list = target.read_rows(query)
        assert len(row_list) == bool(expect_match), (
            f"row {type(cell_value)}({cell_value}) not found with {type(filter_input)}({filter_input}) filter"
        )

    @pytest.mark.usefixtures("target")
    @CrossSync._Sync_Impl.Retry(
        predicate=retry.if_exception_type(ClientError), initial=1, maximum=5
    )
    @pytest.mark.parametrize(
        "cell_value,mask,expect_match",
        [
            (b"\x01\x02\x03", b"\x01\x02\x03", True),
            (b"\x01\x02\x03", b"\x01\x00\x00", True),
            (b"\x00\x02\x03", b"\x01\x00\x00", False),
        ],
    )
    @pytest.mark.skipif(
        bool(os.environ.get(BIGTABLE_EMULATOR)),
        reason="value_bitmask_filter not supported by emulator",
    )
    def test_value_bitmask_filter(
        self, target, temp_rows, cell_value, mask, expect_match
    ):
        """ValueBitmaskFilter matches cells where (value & mask) == mask.
        Make sure inputs are properly interpreted by the server."""
        from google.cloud.bigtable.data import ReadRowsQuery
        from google.cloud.bigtable.data.row_filters import ValueBitmaskFilter

        f = ValueBitmaskFilter(mask)
        temp_rows.add_row(b"row_key_1", value=cell_value)
        query = ReadRowsQuery(row_keys=[b"row_key_1"], row_filter=f)
        row_list = target.read_rows(query)
        assert len(row_list) == bool(expect_match), (
            f"row {cell_value!r} not matched as {expect_match} with {mask!r} bitmask filter"
        )

    @pytest.mark.skipif(
        bool(os.environ.get(BIGTABLE_EMULATOR)), reason="emulator doesn't support SQL"
    )
    def test_authorized_view_unauthenticated(
        self, client, authorized_view_id, instance_id, table_id
    ):
        """Requesting family outside authorized family_subset should raise exception"""
        from google.cloud.bigtable.data.mutations import SetCell

        with client.get_authorized_view(
            instance_id, table_id, authorized_view_id
        ) as view:
            mutation = SetCell(family="unauthorized", qualifier="q", new_value="v")
            with pytest.raises(PermissionDenied) as e:
                view.mutate_row(b"row-key", mutation)
            assert "outside the Authorized View" in e.value.message

    @pytest.mark.skipif(
        bool(os.environ.get(BIGTABLE_EMULATOR)), reason="emulator doesn't support SQL"
    )
    @pytest.mark.usefixtures("client")
    @CrossSync._Sync_Impl.Retry(
        predicate=retry.if_exception_type(ClientError), initial=1, maximum=5
    )
    def test_execute_query_simple(self, client, table_id, instance_id):
        result = client.execute_query("SELECT 1 AS a, 'foo' AS b", instance_id)
        rows = [r for r in result]
        assert len(rows) == 1
        row = rows[0]
        assert row["a"] == 1
        assert row["b"] == "foo"

    @pytest.mark.skipif(
        bool(os.environ.get(BIGTABLE_EMULATOR)), reason="emulator doesn't support SQL"
    )
    @pytest.mark.usefixtures("target")
    @CrossSync._Sync_Impl.Retry(
        predicate=retry.if_exception_type(ClientError), initial=1, maximum=5
    )
    def test_execute_against_target(
        self, client, instance_id, table_id, temp_rows, column_family_config
    ):
        temp_rows.add_row(b"row_key_1")
        result = client.execute_query("SELECT * FROM `" + table_id + "`", instance_id)
        rows = [r for r in result]
        assert len(rows) == 1
        assert rows[0]["_key"] == b"row_key_1"
        family_map = rows[0][TEST_FAMILY]
        assert len(family_map) == 1
        assert family_map[b"q"] == b"test-value"
        assert len(rows[0][TEST_FAMILY_2]) == 0
        md = result.metadata
        assert len(md) == len(column_family_config) + 1
        assert md["_key"].column_type == SqlType.Bytes()
        assert md[TEST_FAMILY].column_type == SqlType.Map(
            SqlType.Bytes(), SqlType.Bytes()
        )
        assert md[TEST_FAMILY_2].column_type == SqlType.Map(
            SqlType.Bytes(), SqlType.Bytes()
        )
        assert md[TEST_AGGREGATE_FAMILY].column_type == SqlType.Map(
            SqlType.Bytes(), SqlType.Int64()
        )

    @pytest.mark.skipif(
        bool(os.environ.get(BIGTABLE_EMULATOR)), reason="emulator doesn't support SQL"
    )
    @pytest.mark.usefixtures("client")
    @CrossSync._Sync_Impl.Retry(
        predicate=retry.if_exception_type(ClientError), initial=1, maximum=5
    )
    def test_execute_query_params(self, client, table_id, instance_id):
        query = "SELECT @stringParam AS strCol, @bytesParam as bytesCol, @int64Param AS intCol, @float32Param AS float32Col, @float64Param AS float64Col, @boolParam AS boolCol, @tsParam AS tsCol, @dateParam AS dateCol, @byteArrayParam AS byteArrayCol, @stringArrayParam AS stringArrayCol, @intArrayParam AS intArrayCol, @float32ArrayParam AS float32ArrayCol, @float64ArrayParam AS float64ArrayCol, @boolArrayParam AS boolArrayCol, @tsArrayParam AS tsArrayCol, @dateArrayParam AS dateArrayCol"
        parameters = {
            "stringParam": "foo",
            "bytesParam": b"bar",
            "int64Param": 12,
            "float32Param": 1.1,
            "float64Param": 1.2,
            "boolParam": True,
            "tsParam": datetime.datetime.fromtimestamp(1000, tz=datetime.timezone.utc),
            "dateParam": datetime.date(2025, 1, 16),
            "byteArrayParam": [b"foo", b"bar", None],
            "stringArrayParam": ["foo", "bar", None],
            "intArrayParam": [1, None, 2],
            "float32ArrayParam": [1.2, None, 1.3],
            "float64ArrayParam": [1.4, None, 1.5],
            "boolArrayParam": [None, False, True],
            "tsArrayParam": [
                datetime.datetime.fromtimestamp(1000, tz=datetime.timezone.utc),
                datetime.datetime.fromtimestamp(2000, tz=datetime.timezone.utc),
                None,
            ],
            "dateArrayParam": [
                datetime.date(2025, 1, 16),
                datetime.date(2025, 1, 17),
                None,
            ],
        }
        param_types = {
            "stringParam": SqlType.String(),
            "bytesParam": SqlType.Bytes(),
            "int64Param": SqlType.Int64(),
            "float32Param": SqlType.Float32(),
            "float64Param": SqlType.Float64(),
            "boolParam": SqlType.Bool(),
            "tsParam": SqlType.Timestamp(),
            "dateParam": SqlType.Date(),
            "byteArrayParam": SqlType.Array(SqlType.Bytes()),
            "stringArrayParam": SqlType.Array(SqlType.String()),
            "intArrayParam": SqlType.Array(SqlType.Int64()),
            "float32ArrayParam": SqlType.Array(SqlType.Float32()),
            "float64ArrayParam": SqlType.Array(SqlType.Float64()),
            "boolArrayParam": SqlType.Array(SqlType.Bool()),
            "tsArrayParam": SqlType.Array(SqlType.Timestamp()),
            "dateArrayParam": SqlType.Array(SqlType.Date()),
        }
        result = client.execute_query(
            query, instance_id, parameters=parameters, parameter_types=param_types
        )
        rows = [r for r in result]
        assert len(rows) == 1
        row = rows[0]
        assert row["strCol"] == parameters["stringParam"]
        assert row["bytesCol"] == parameters["bytesParam"]
        assert row["intCol"] == parameters["int64Param"]
        assert row["float32Col"] == pytest.approx(parameters["float32Param"])
        assert row["float64Col"] == pytest.approx(parameters["float64Param"])
        assert row["boolCol"] == parameters["boolParam"]
        assert row["tsCol"] == parameters["tsParam"]
        assert row["dateCol"] == date_pb2.Date(year=2025, month=1, day=16)
        assert row["stringArrayCol"] == parameters["stringArrayParam"]
        assert row["byteArrayCol"] == parameters["byteArrayParam"]
        assert row["intArrayCol"] == parameters["intArrayParam"]
        assert row["float32ArrayCol"] == pytest.approx(parameters["float32ArrayParam"])
        assert row["float64ArrayCol"] == pytest.approx(parameters["float64ArrayParam"])
        assert row["boolArrayCol"] == parameters["boolArrayParam"]
        assert row["tsArrayCol"] == parameters["tsArrayParam"]
        assert row["dateArrayCol"] == [
            date_pb2.Date(year=2025, month=1, day=16),
            date_pb2.Date(year=2025, month=1, day=17),
            None,
        ]

    @pytest.mark.skipif(
        bool(os.environ.get(BIGTABLE_EMULATOR)), reason="emulator doesn't support SQL"
    )
    @pytest.mark.usefixtures("target")
    @CrossSync._Sync_Impl.Retry(
        predicate=retry.if_exception_type(ClientError), initial=1, maximum=5
    )
    def test_execute_metadata_on_empty_response(
        self, client, instance_id, table_id, temp_rows, column_family_config
    ):
        temp_rows.add_row(b"row_key_1")
        result = client.execute_query(
            "SELECT * FROM `" + table_id + "` WHERE _key='non-existent'", instance_id
        )
        rows = [r for r in result]
        assert len(rows) == 0
        md = result.metadata
        assert len(md) == len(column_family_config) + 1
        assert md["_key"].column_type == SqlType.Bytes()
        assert md[TEST_FAMILY].column_type == SqlType.Map(
            SqlType.Bytes(), SqlType.Bytes()
        )
        assert md[TEST_FAMILY_2].column_type == SqlType.Map(
            SqlType.Bytes(), SqlType.Bytes()
        )
        assert md[TEST_AGGREGATE_FAMILY].column_type == SqlType.Map(
            SqlType.Bytes(), SqlType.Int64()
        )