test(bigtable): set up accelerator system test environment and harness

packages/google-cloud-bigtable/conftest.py

@@ -0,0 +1,36 @@
+# Copyright 2026 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.
+"""Root pytest configuration for the package.
+
+Keeps ``pytest.ini`` portable across ``pytest-asyncio`` versions by registering
+``asyncio_default_fixture_loop_scope`` and ``asyncio_default_test_loop_scope``
+when running under older ``pytest-asyncio`` (<0.23).
+"""
+
+
+def pytest_addoption(parser):
+    for name, help_text in (
+        (
+            "asyncio_default_fixture_loop_scope",
+            "default event-loop scope for asyncio fixtures (pytest-asyncio)",
+        ),
+        (
+            "asyncio_default_test_loop_scope",
+            "default event-loop scope for asyncio tests (pytest-asyncio)",
+        ),
+    ):
+        try:
+            parser.addini(name, help_text, default=None)
+        except ValueError:
+            pass

packages/google-cloud-bigtable/pytest.ini

@@ -0,0 +1,3 @@
+[pytest]
+asyncio_default_fixture_loop_scope = session
+asyncio_default_test_loop_scope = session

packages/google-cloud-bigtable/tests/system/data/accelerator/README.md

@@ -0,0 +1,31 @@
+# Cloud Bigtable Accelerator System Tests
+
+Shared environment setup, fixtures, and state/process helpers for live Cloud Bigtable Python accelerator system tests. Tests using this harness run against a real Cloud Bigtable instance and a real Go accelerator binary; emulator and binary-less environments skip cleanly.
+
+## Harness Components
+
+- [`conftest.py`](conftest.py): Registers the `accelerator` and `slow` pytest markers, provides `tmp_bin_dir`, and wraps `get_table()` on both sync and async data clients to inject `BIGTABLE_TEST_APP_PROFILE` (or `CBT_APP_PROFILE_ID`) when no `app_profile_id` is passed.
+- [`_base_async.py`](_base_async.py) / [`_base_autogen.py`](_base_autogen.py): Define `AcceleratorTestBaseAsync` and its CrossSync sibling `AcceleratorTestBase`. They enforce live-instance and binary prerequisites per session, expose `client`, `accel_client`, `native_client`, `accel_table`, `native_table`, `default_table`, and `janitor` cleanup fixtures, and assert `table._accelerator_client` state via `assert_accelerator_active()` and `assert_native()`.
+- [`_harness.py`](_harness.py): Provides the in-memory reference model (`ExpectedState`, `normalize_row`, `assert_rows_equivalent`), seeded workload generator (`RandomOps` with atomic multi-family/multi-column batches and `DELETE_AFTER_WRITE_GUARD_OPS = 512`), process leak/kill helpers (`ProcessIntrospector`, `force_kill_daemon`, `daemon_pid`, `pid_alive`), and binary/environment gates (`resolve_binary`, `require_binary_or_skip`, `require_real_bigtable_or_skip`).
+- Package root [`pytest.ini`](../../../../pytest.ini) & [`conftest.py`](../../../../conftest.py): Set session-scoped `pytest-asyncio` loops and register the loop-scope ini options so older `pytest-asyncio` (<0.23) versions do not warn or fail.
+
+## Environment Variables
+
+| Variable | Required | Purpose |
+| --- | --- | --- |
+| `GOOGLE_CLOUD_PROJECT` (or `PROJECT_ID`) | Yes | Target GCP project; tests skip if unset. |
+| `BIGTABLE_EMULATOR_HOST` | Must be unset | Emulator runs skip because the Go accelerator does not support the emulator. |
+| `BIGTABLE_ACCELERATOR_BIN` | No | Path to a locally built Go accelerator binary; overrides the wheel-bundled binary lookup (`_daemon._resolve_binary_path()`). |
+| `BIGTABLE_TEST_APP_PROFILE` (or `CBT_APP_PROFILE_ID`) | No | Default app profile ID injected into `get_table()` calls. |
+
+## Running Locally & Syncing
+
+Run pytest from `packages/google-cloud-bigtable/` once a compiled Go binary (or bundled wheel) and GCP project are available:
+
+```bash
+BIGTABLE_ACCELERATOR_BIN=/path/to/accelerator \
+GOOGLE_CLOUD_PROJECT=my-project \
+pytest tests/system/data/accelerator -v
+```
+
+If you edit `_base_async.py`, regenerate `_base_autogen.py` from `packages/google-cloud-bigtable/` with `nox -s generate_sync`.

packages/google-cloud-bigtable/tests/system/data/accelerator/__init__.py

@@ -0,0 +1,13 @@
+# Copyright 2026 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.

packages/google-cloud-bigtable/tests/system/data/accelerator/_base_async.py

@@ -0,0 +1,117 @@
+# Copyright 2026 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.
+
+import os
+
+import pytest
+
+from google.cloud.bigtable.data._accelerator import _daemon
+from google.cloud.bigtable.data._cross_sync import CrossSync
+from google.cloud.bigtable.data.mutations import DeleteAllFromRow, RowMutationEntry
+
+from .. import SystemTestRunner
+from . import _harness
+
+__CROSS_SYNC_OUTPUT__ = "tests.system.data.accelerator._base_autogen"
+
+
+@CrossSync.convert_class(sync_name="AcceleratorTestBase")
+class AcceleratorTestBaseAsync(SystemTestRunner):
+    @pytest.fixture(scope="session", autouse=True)
+    def _require_accel_env(self):
+        _harness.require_real_bigtable_or_skip()
+        path = _harness.require_binary_or_skip()
+        orig_resolve = _daemon._resolve_binary_path
+        if os.environ.get(_harness.BIN_ENV_VAR):
+            _daemon._resolve_binary_path = lambda: path
+        try:
+            yield
+        finally:
+            _daemon._resolve_binary_path = orig_resolve
+
+    def _project_id(self):
+        return os.getenv("GOOGLE_CLOUD_PROJECT") or os.getenv("PROJECT_ID") or None
+
+    def _make_client(self, use_accelerator=None, **kwargs):
+        return CrossSync.DataClient(
+            project=self._project_id(), use_accelerator=use_accelerator, **kwargs
+        )
+
+    def assert_accelerator_active(self, table):
+        assert table._accelerator_client is not None, (
+            "expected the accelerator to be active (use_accelerator=True) but the "
+            "target fell back to native. Check ADC principal / identity "
+            "verification and that the bundled daemon binary can start."
+        )
+
+    def assert_native(self, table):
+        assert table._accelerator_client is None, (
+            "expected a native target (use_accelerator=False) but an accelerator "
+            "client was attached."
+        )
+
+    @CrossSync.convert
+    @CrossSync.pytest_fixture(scope="session")
+    async def client(self):
+        async with self._make_client() as client:
+            yield client
+
+    @CrossSync.convert
+    @CrossSync.pytest_fixture(scope="session")
+    async def accel_client(self):
+        async with self._make_client(use_accelerator=True) as client:
+            yield client
+
+    @CrossSync.convert
+    @CrossSync.pytest_fixture(scope="session")
+    async def native_client(self):
+        async with self._make_client(use_accelerator=False) as client:
+            yield client
+
+    @CrossSync.convert
+    @CrossSync.pytest_fixture(scope="session")
+    async def accel_table(self, accel_client, instance_id, table_id):
+        async with accel_client.get_table(instance_id, table_id) as table:
+            self.assert_accelerator_active(table)
+            yield table
+
+    @CrossSync.convert
+    @CrossSync.pytest_fixture(scope="session")
+    async def native_table(self, native_client, instance_id, table_id):
+        async with native_client.get_table(instance_id, table_id) as table:
+            self.assert_native(table)
+            yield table
+
+    @CrossSync.convert
+    @CrossSync.pytest_fixture(scope="session")
+    async def default_table(self, client, instance_id, table_id):
+        async with client.get_table(instance_id, table_id) as table:
+            yield table
+
+    @CrossSync.convert
+    @CrossSync.pytest_fixture(scope="function")
+    async def janitor(self, native_table):
+        class _Janitor:
+            def __init__(self):
+                self.keys = set()
+
+            def track(self, key):
+                self.keys.add(key)
+                return key
+
+        j = _Janitor()
+        yield j
+        if j.keys:
+            entries = [RowMutationEntry(key, [DeleteAllFromRow()]) for key in j.keys]
+            await native_table.bulk_mutate_rows(entries)

packages/google-cloud-bigtable/tests/system/data/accelerator/_base_autogen.py

@@ -0,0 +1,107 @@
+# Copyright 2026 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 os
+
+import pytest
+
+from google.cloud.bigtable.data._accelerator import _daemon
+from google.cloud.bigtable.data._cross_sync import CrossSync
+from google.cloud.bigtable.data.mutations import DeleteAllFromRow, RowMutationEntry
+
+from .. import SystemTestRunner
+from . import _harness
+
+
+class AcceleratorTestBase(SystemTestRunner):
+    @pytest.fixture(scope="session", autouse=True)
+    def _require_accel_env(self):
+        _harness.require_real_bigtable_or_skip()
+        path = _harness.require_binary_or_skip()
+        orig_resolve = _daemon._resolve_binary_path
+        if os.environ.get(_harness.BIN_ENV_VAR):
+            _daemon._resolve_binary_path = lambda: path
+        try:
+            yield
+        finally:
+            _daemon._resolve_binary_path = orig_resolve
+
+    def _project_id(self):
+        return os.getenv("GOOGLE_CLOUD_PROJECT") or os.getenv("PROJECT_ID") or None
+
+    def _make_client(self, use_accelerator=None, **kwargs):
+        return CrossSync._Sync_Impl.DataClient(
+            project=self._project_id(), use_accelerator=use_accelerator, **kwargs
+        )
+
+    def assert_accelerator_active(self, table):
+        assert table._accelerator_client is not None, (
+            "expected the accelerator to be active (use_accelerator=True) but the target fell back to native. Check ADC principal / identity verification and that the bundled daemon binary can start."
+        )
+
+    def assert_native(self, table):
+        assert table._accelerator_client is None, (
+            "expected a native target (use_accelerator=False) but an accelerator client was attached."
+        )
+
+    @pytest.fixture(scope="session")
+    def client(self):
+        with self._make_client() as client:
+            yield client
+
+    @pytest.fixture(scope="session")
+    def accel_client(self):
+        with self._make_client(use_accelerator=True) as client:
+            yield client
+
+    @pytest.fixture(scope="session")
+    def native_client(self):
+        with self._make_client(use_accelerator=False) as client:
+            yield client
+
+    @pytest.fixture(scope="session")
+    def accel_table(self, accel_client, instance_id, table_id):
+        with accel_client.get_table(instance_id, table_id) as table:
+            self.assert_accelerator_active(table)
+            yield table
+
+    @pytest.fixture(scope="session")
+    def native_table(self, native_client, instance_id, table_id):
+        with native_client.get_table(instance_id, table_id) as table:
+            self.assert_native(table)
+            yield table
+
+    @pytest.fixture(scope="session")
+    def default_table(self, client, instance_id, table_id):
+        with client.get_table(instance_id, table_id) as table:
+            yield table
+
+    @pytest.fixture(scope="function")
+    def janitor(self, native_table):
+        class _Janitor:
+            def __init__(self):
+                self.keys = set()
+
+            def track(self, key):
+                self.keys.add(key)
+                return key
+
+        j = _Janitor()
+        yield j
+        if j.keys:
+            entries = [RowMutationEntry(key, [DeleteAllFromRow()]) for key in j.keys]
+            native_table.bulk_mutate_rows(entries)