chore(tests): add conformance tests to CI for v3 (#870)

.github/sync-repo-settings.yaml

@@ -31,6 +31,24 @@ branchProtectionRules:
     - 'Kokoro'
     - 'Kokoro system-3.8'
     - 'cla/google'
+- pattern: experimental_v3
+  # Can admins overwrite branch protection.
+  # Defaults to `true`
+  isAdminEnforced: false
+  # Number of approving reviews required to update matching branches.
+  # Defaults to `1`
+  requiredApprovingReviewCount: 1
+  # Are reviews from code owners required to update matching branches.
+  # Defaults to `false`
+  requiresCodeOwnerReviews: false
+  # Require up to date branches
+  requiresStrictStatusChecks: false
+  # List of required status check contexts that must pass for commits to be accepted to matching branches.
+  requiredStatusCheckContexts:
+    - 'Kokoro'
+    - 'Kokoro system-3.8'
+    - 'cla/google'
+    - 'Conformance / Async v3 Client / Python 3.8'
 # List of explicit permissions to add (additive only)
 permissionRules:
     # Team slug to add to repository permissions

.github/workflows/conformance.yaml

@@ -0,0 +1,54 @@
+# Copyright 2023 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.
+# Github action job to test core java library features on
+# downstream client libraries before they are released.
+on:
+  push:
+    branches:
+    - main
+  pull_request:
+name: Conformance
+jobs:
+  conformance:
+    runs-on: ubuntu-latest
+    strategy:
+      matrix:
+        py-version: [ 3.8 ]
+        client-type: [ "Async v3", "Legacy" ]
+    name: "${{ matrix.client-type }} Client / Python ${{ matrix.py-version }}"
+    steps:
+    - uses: actions/checkout@v3
+      name: "Checkout python-bigtable"
+    - uses: actions/checkout@v3
+      name: "Checkout conformance tests"
+      with:
+        repository: googleapis/cloud-bigtable-clients-test
+        ref: main
+        path: cloud-bigtable-clients-test
+    - uses: actions/setup-python@v4
+      with:
+        python-version: ${{ matrix.py-version }}
+    - uses: actions/setup-go@v4
+      with:
+        go-version: '>=1.20.2'
+    - run: chmod +x .kokoro/conformance.sh
+    - run: pip install -e .
+      name: "Install python-bigtable from HEAD"
+    - run: go version
+    - run: .kokoro/conformance.sh
+      name: "Run tests"
+      env:
+        CLIENT_TYPE: ${{ matrix.client-type }}
+        PYTHONUNBUFFERED: 1
+

.kokoro/conformance.sh

@@ -0,0 +1,52 @@
+#!/bin/bash
+
+# Copyright 2023 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.
+
+set -eo pipefail
+
+## cd to the parent directory, i.e. the root of the git repo
+cd $(dirname $0)/..
+
+PROXY_ARGS=""
+TEST_ARGS=""
+if [[ "${CLIENT_TYPE^^}" == "LEGACY" ]]; then
+  echo "Using legacy client"
+  PROXY_ARGS="--legacy-client"
+  # legacy client does not expose mutate_row. Disable those tests
+  TEST_ARGS="-skip TestMutateRow_"
+fi
+
+# Build and start the proxy in a separate process
+PROXY_PORT=9999
+pushd test_proxy
+nohup python test_proxy.py --port $PROXY_PORT $PROXY_ARGS &
+proxyPID=$!
+popd
+
+# Kill proxy on exit
+function cleanup() {
+    echo "Cleanup testbench";
+    kill $proxyPID
+}
+trap cleanup EXIT
+
+# Run the conformance test
+pushd cloud-bigtable-clients-test/tests
+eval "go test -v -proxy_addr=:$PROXY_PORT $TEST_ARGS"
+RETURN_CODE=$?
+popd
+
+echo "exiting with ${RETURN_CODE}"
+exit ${RETURN_CODE}

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

@@ -132,6 +132,10 @@ def __init__(
         client_options = cast(
             Optional[client_options_lib.ClientOptions], client_options
         )
+        self._emulator_host = os.getenv(BIGTABLE_EMULATOR)
+        if self._emulator_host is not None and credentials is None:
+            # use insecure channel if emulator is set
+            credentials = google.auth.credentials.AnonymousCredentials()
         # initialize client
         ClientWithProject.__init__(
             self,
@@ -155,7 +159,6 @@ def __init__(
         self._instance_owners: dict[_WarmedInstanceKey, Set[int]] = {}
         self._channel_init_time = time.monotonic()
         self._channel_refresh_tasks: list[asyncio.Task[None]] = []
-        self._emulator_host = os.getenv(BIGTABLE_EMULATOR)
         if self._emulator_host is not None:
             # connect to an emulator host
             warnings.warn(

test_proxy/handlers/client_handler_legacy.py

@@ -44,7 +44,7 @@ def __init__(
         self.app_profile_id = app_profile_id
         self.per_operation_timeout = per_operation_timeout
 
-    async def close(self):
+    def close(self):
         self.closed = True
 
     @client_handler.error_safe
@@ -79,54 +79,157 @@ async def ReadRows(self, request, **kwargs):
             row_list.append(dict_val)
         return row_list
 
+    @client_handler.error_safe
+    async def ReadRow(self, row_key, **kwargs):
+        table_id = kwargs["table_name"].split("/")[-1]
+        instance = self.client.instance(self.instance_id)
+        table = instance.table(table_id)
+
+        row = table.read_row(row_key)
+        # parse results into proto formatted dict
+        dict_val = {"row_key": row.row_key}
+        for family, family_cells in row.cells.items():
+            family_dict = {"name": family}
+            for qualifier, qualifier_cells in family_cells.items():
+                column_dict = {"qualifier": qualifier}
+                for cell in qualifier_cells:
+                    cell_dict = {
+                        "value": cell.value,
+                        "timestamp_micros": cell.timestamp.timestamp() * 1000000,
+                        "labels": cell.labels,
+                    }
+                    column_dict.setdefault("cells", []).append(cell_dict)
+                family_dict.setdefault("columns", []).append(column_dict)
+            dict_val.setdefault("families", []).append(family_dict)
+        return dict_val
+
     @client_handler.error_safe
     async def MutateRow(self, request, **kwargs):
+        from datetime import datetime
         from google.cloud.bigtable.row import DirectRow
         table_id = request["table_name"].split("/")[-1]
         instance = self.client.instance(self.instance_id)
         table = instance.table(table_id)
         row_key = request["row_key"]
         new_row = DirectRow(row_key, table)
         for m_dict in request.get("mutations", []):
+            details = m_dict.get("set_cell") or m_dict.get("delete_from_column") or m_dict.get("delete_from_family") or m_dict.get("delete_from_row")
+            timestamp = datetime.fromtimestamp(details.get("timestamp_micros")) if details.get("timestamp_micros") else None
             if m_dict.get("set_cell"):
-                details = m_dict["set_cell"]
-                new_row.set_cell(details["family_name"], details["column_qualifier"], details["value"], timestamp=details["timestamp_micros"])
+                new_row.set_cell(details["family_name"], details["column_qualifier"], details["value"], timestamp=timestamp)
             elif m_dict.get("delete_from_column"):
-                details = m_dict["delete_from_column"]
-                new_row.delete_cell(details["family_name"], details["column_qualifier"], timestamp=details["timestamp_micros"])
+                new_row.delete_cell(details["family_name"], details["column_qualifier"], timestamp=timestamp)
             elif m_dict.get("delete_from_family"):
-                details = m_dict["delete_from_family"]
-                new_row.delete_cells(details["family_name"], timestamp=details["timestamp_micros"])
+                new_row.delete_cells(details["family_name"], timestamp=timestamp)
             elif m_dict.get("delete_from_row"):
                 new_row.delete()
-        async with self.measure_call():
-            table.mutate_rows([new_row])
+        table.mutate_rows([new_row])
         return "OK"
 
     @client_handler.error_safe
     async def BulkMutateRows(self, request, **kwargs):
         from google.cloud.bigtable.row import DirectRow
+        from datetime import datetime
         table_id = request["table_name"].split("/")[-1]
         instance = self.client.instance(self.instance_id)
         table = instance.table(table_id)
         rows = []
         for entry in request.get("entries", []):
             row_key = entry["row_key"]
             new_row = DirectRow(row_key, table)
-            for m_dict in entry.get("mutations", {}):
+            for m_dict in entry.get("mutations"):
+                details = m_dict.get("set_cell") or m_dict.get("delete_from_column") or m_dict.get("delete_from_family") or m_dict.get("delete_from_row")
+                timestamp = datetime.fromtimestamp(details.get("timestamp_micros")) if details.get("timestamp_micros") else None
                 if m_dict.get("set_cell"):
-                    details = m_dict["set_cell"]
-                    new_row.set_cell(details["family_name"], details["column_qualifier"], details["value"], timestamp=details.get("timestamp_micros",None))
+                    new_row.set_cell(details["family_name"], details["column_qualifier"], details["value"], timestamp=timestamp)
                 elif m_dict.get("delete_from_column"):
-                    details = m_dict["delete_from_column"]
-                    new_row.delete_cell(details["family_name"], details["column_qualifier"], timestamp=details["timestamp_micros"])
+                    new_row.delete_cell(details["family_name"], details["column_qualifier"], timestamp=timestamp)
                 elif m_dict.get("delete_from_family"):
-                    details = m_dict["delete_from_family"]
-                    new_row.delete_cells(details["family_name"], timestamp=details["timestamp_micros"])
+                    new_row.delete_cells(details["family_name"], timestamp=timestamp)
                 elif m_dict.get("delete_from_row"):
                     new_row.delete()
             rows.append(new_row)
-        async with self.measure_call():
-            table.mutate_rows(rows)
+        table.mutate_rows(rows)
         return "OK"
 
+    @client_handler.error_safe
+    async def CheckAndMutateRow(self, request, **kwargs):
+        from google.cloud.bigtable.row import ConditionalRow
+        from google.cloud.bigtable.row_filters import PassAllFilter
+        table_id = request["table_name"].split("/")[-1]
+        instance = self.client.instance(self.instance_id)
+        table = instance.table(table_id)
+
+        predicate_filter = request.get("predicate_filter", PassAllFilter(True))
+        new_row = ConditionalRow(request["row_key"], table, predicate_filter)
+
+        combined_mutations = [{"state": True, **m} for m in request.get("true_mutations", [])]
+        combined_mutations.extend([{"state": False, **m} for m in request.get("false_mutations", [])])
+        for mut_dict in combined_mutations:
+            if "set_cell" in mut_dict:
+                details = mut_dict["set_cell"]
+                new_row.set_cell(
+                    details.get("family_name", ""),
+                    details.get("column_qualifier", ""),
+                    details.get("value", ""),
+                    timestamp=details.get("timestamp_micros", None),
+                    state=mut_dict["state"],
+                )
+            elif "delete_from_column" in mut_dict:
+                details = mut_dict["delete_from_column"]
+                new_row.delete_cell(
+                    details.get("family_name", ""),
+                    details.get("column_qualifier", ""),
+                    timestamp=details.get("timestamp_micros", None),
+                    state=mut_dict["state"],
+                )
+            elif "delete_from_family" in mut_dict:
+                details = mut_dict["delete_from_family"]
+                new_row.delete_cells(
+                    details.get("family_name", ""),
+                    timestamp=details.get("timestamp_micros", None),
+                    state=mut_dict["state"],
+                )
+            elif "delete_from_row" in mut_dict:
+                new_row.delete(state=mut_dict["state"])
+            else:
+                raise RuntimeError(f"Unknown mutation type: {mut_dict}")
+            return new_row.commit()
+
+    @client_handler.error_safe
+    async def ReadModifyWriteRow(self, request, **kwargs):
+        from google.cloud.bigtable.row import AppendRow
+        from google.cloud._helpers import _microseconds_from_datetime
+        table_id = request["table_name"].split("/")[-1]
+        instance = self.client.instance(self.instance_id)
+        table = instance.table(table_id)
+        row_key = request["row_key"]
+        new_row = AppendRow(row_key, table)
+        for rule_dict in request.get("rules", []):
+            qualifier = rule_dict["column_qualifier"]
+            family = rule_dict["family_name"]
+            if "append_value" in rule_dict:
+                new_row.append_cell_value(family, qualifier, rule_dict["append_value"])
+            else:
+                new_row.increment_cell_value(family, qualifier, rule_dict["increment_amount"])
+        raw_result = new_row.commit()
+        result_families = []
+        for family, column_dict in raw_result.items():
+            result_columns = []
+            for column, cell_list in column_dict.items():
+                result_cells = []
+                for cell_tuple in cell_list:
+                    cell_dict = {"value": cell_tuple[0], "timestamp_micros": _microseconds_from_datetime(cell_tuple[1])}
+                    result_cells.append(cell_dict)
+                result_columns.append({"qualifier": column, "cells": result_cells})
+            result_families.append({"name": family, "columns": result_columns})
+        return {"key": row_key, "families": result_families}
+
+    @client_handler.error_safe
+    async def SampleRowKeys(self, request, **kwargs):
+        table_id = request["table_name"].split("/")[-1]
+        instance = self.client.instance(self.instance_id)
+        table = instance.table(table_id)
+        response = list(table.sample_row_keys())
+        tuple_response = [(s.row_key, s.offset_bytes) for s in response]
+        return tuple_response

testing/constraints-3.8.txt

@@ -0,0 +1,13 @@
+# This constraints file is used to check that lower bounds
+# are correct in setup.py
+# List *all* library dependencies and extras in this file.
+# Pin the version to the lower bound.
+#
+# e.g., if setup.py has "foo >= 1.14.0, < 2.0.0dev",
+# Then this file should have foo==1.14.0
+google-api-core==2.12.0.dev1
+google-cloud-core==2.3.2
+grpc-google-iam-v1==0.12.4
+proto-plus==1.22.0
+libcst==0.2.5
+protobuf==3.19.5