diff --git a/packages/google-cloud-bigtable/google/cloud/bigtable/data/_accelerator/_daemon.py b/packages/google-cloud-bigtable/google/cloud/bigtable/data/_accelerator/_daemon.py index 5e341d9c76fa..887d43446943 100644 --- a/packages/google-cloud-bigtable/google/cloud/bigtable/data/_accelerator/_daemon.py +++ b/packages/google-cloud-bigtable/google/cloud/bigtable/data/_accelerator/_daemon.py @@ -16,9 +16,11 @@ The daemon binary embeds an in-process Go Bigtable client and exposes the ``google.bigtable.v2.Bigtable`` service over a Unix domain socket. This module -owns spawning the binary, waiting for the UDS to become connectable, and -tearing the process down. It does NOT speak gRPC; that's the -``_AcceleratorClient`` companion's job. +owns spawning the binary, waiting for its gRPC server to come up, and tearing +the process down. It issues no RPCs — that's the ``_AcceleratorClient`` +companion's job — but it does open a throwaway channel to decide when the +daemon is ready, since on Windows that is the only probe available from +Python (see ``_wait_until_ready``). """ from __future__ import annotations @@ -30,10 +32,20 @@ import shutil import socket import subprocess +import sys import tempfile import time from typing import Any, Mapping, Sequence +import grpc + +# Windows differs from every other supported platform in two ways that matter +# here, so it gets a named constant rather than scattered ``sys.platform`` +# checks: the daemon binary carries a ``.exe`` suffix, and CPython's socket +# module exposes no ``AF_UNIX`` even though the OS and gRPC's C-core both speak +# it (see ``start`` and ``_wait_until_ready``). +_IS_WINDOWS = sys.platform == "win32" + # Environment variable that overrides the daemon startup readiness timeout, in # seconds (float). Lets a deployment tune the cold-start budget without a code # change or rebuild — e.g. raise it on a very constrained/gVisor host, or (when @@ -42,7 +54,10 @@ _STARTUP_TIMEOUT_ENV_VAR = "BIGTABLE_ACCELERATOR_STARTUP_TIMEOUT" # Wheels ship the binary at this path relative to the `_accelerator/` package. +# The Windows wheels (win_amd64, win32) stage it under the ``.exe`` name Go +# gives it, so the suffix is appended there. _DEFAULT_BIN_RELATIVE_PATH = "bin/accelerator" +_WINDOWS_BIN_SUFFIX = ".exe" # The daemon writes the principal it resolved to this file in its tempdir # (alongside the socket) before binding, so the client can verify it matches @@ -76,21 +91,20 @@ # ``_resolve_startup_timeout``). _DEFAULT_STARTUP_TIMEOUT = 20.0 +# How long each poll of the readiness future waits before the loop re-checks +# that the daemon process is still alive. Doubles as the loop's sleep. +_READY_POLL_SECONDS = 0.05 + # Sequence: close stdin, wait this long; SIGTERM, wait again; SIGKILL. _STDIN_GRACE_SECONDS = 2.0 _SIGTERM_GRACE_SECONDS = 2.0 _SIGKILL_GRACE_SECONDS = 2.0 -# How long a single connect() probe of a daemon's socket may take. A UDS -# connect against a listening socket completes in the kernel without the -# daemon doing anything, so this only ever elapses on a badly overloaded -# host — where the honest answer is "I cannot tell", not "it is dead". -_SOCKET_PROBE_TIMEOUT_SECONDS = 0.25 - def _resolve_binary_path() -> str: """Return the path of the bundled daemon binary, raising if it is absent.""" - bundled = os.path.join(os.path.dirname(__file__), _DEFAULT_BIN_RELATIVE_PATH) + relative = _DEFAULT_BIN_RELATIVE_PATH + (_WINDOWS_BIN_SUFFIX if _IS_WINDOWS else "") + bundled = os.path.join(os.path.dirname(__file__), relative) if not os.path.isfile(bundled): raise FileNotFoundError( "Accelerator binary not found. Install a wheel that bundles the binary." @@ -126,21 +140,16 @@ def _resolve_startup_timeout() -> float: return value -def _socket_is_live(uds_path: str) -> bool: - """Return True if something is currently accepting connections on the UDS. +def _uds_target(uds_path: str) -> str: + """Build the gRPC target for the daemon's socket at ``uds_path``. - Used by ``_wait_until_ready`` to poll for the daemon coming up, so every - failure simply answers "not yet" and the caller tries again. + Must be the single-colon form. ``unix://`` opens an *authority* component, + which gRPC only accepts when the rest is an absolute POSIX path; hand it a + Windows path and ``C:`` parses as the authority, so channel construction + fails outright with "authority-based URIs not supported by the unix + scheme". ``unix:`` takes the remainder as a plain path on every platform. """ - if not os.path.exists(uds_path): - return False - with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as probe: - probe.settimeout(_SOCKET_PROBE_TIMEOUT_SECONDS) - try: - probe.connect(uds_path) - return True - except OSError: - return False + return f"unix:{uds_path}" class AcceleratorDaemon: @@ -245,11 +254,21 @@ def start(self) -> None: """ if self._proc is not None: raise RuntimeError("AcceleratorDaemon.start() called twice") - if not hasattr(socket, "AF_UNIX"): + # The daemon only listens on AF_UNIX, so a platform without it cannot + # run the accelerator at all. Windows is the one case where a missing + # ``socket.AF_UNIX`` says nothing about the platform — the OS and + # gRPC's C-core both support the family, only CPython's socket module + # does not — so it is excluded from the check. + if not hasattr(socket, "AF_UNIX") and not _IS_WINDOWS: raise OSError( "Unix domain sockets (AF_UNIX) are not supported on this platform." ) + # mkdtemp is what confines the socket to this user: 0700 on POSIX, and + # on Windows an inherited ACL from the per-user %TEMP% tree. The latter + # is the only access control the socket gets there — Windows has no + # umask and chmod only toggles the read-only bit — so the daemon + # deliberately relies on the directory we pick here. self._tempdir = tempfile.mkdtemp(prefix=_TEMPDIR_PREFIX) self._uds_path = os.path.join(self._tempdir, _SOCKET_FILENAME) @@ -346,7 +365,12 @@ def close(self) -> None: except OSError: pass if not self._wait_for_exit(_STDIN_GRACE_SECONDS): - # Step 2: SIGTERM. + # Step 2: SIGTERM. On Windows this is TerminateProcess, so + # steps 2 and 3 are the same hard kill and the daemon's one + # chance to shut down cleanly is the stdin EOF above — + # which is also the signal it uses there to notice the + # parent died, since a ppid watchdog cannot work on + # Windows. try: proc.terminate() except OSError: @@ -396,25 +420,51 @@ def read_identity(self) -> dict[str, Any]: pass def _wait_until_ready(self, timeout: float) -> None: - """Poll until the UDS accepts a connection, the child dies, or timeout. + """Poll until the daemon's gRPC server is reachable, the child dies, + or timeout. + + Readiness is a gRPC channel reaching ``READY``, not a bare socket + connect. Two reasons. It is the stronger signal: ``net.Listen`` puts + the socket in the listen backlog, so a raw connect succeeds before the + Go gRPC server has accepted anything, whereas ``READY`` means the + HTTP/2 handshake completed and an RPC would actually be served. And it + is the only signal available on Windows, where CPython's socket module + has no ``AF_UNIX`` (``PC/pyconfig.h`` leaves ``HAVE_SYS_UN_H`` + commented out) even though the OS and gRPC's C-core both support the + family — which is why the accelerator runs there at all. Raises RuntimeError if the daemon exits during startup or does not - become connectable within ``timeout`` seconds. + become ready within ``timeout`` seconds. """ if self._proc is None or self._uds_path is None: raise RuntimeError("AcceleratorDaemon._wait_until_ready() before spawn") deadline = time.monotonic() + timeout - while time.monotonic() < deadline: - exit_code = self._proc.poll() - if exit_code is not None: - log_tail = self._read_log_tail() - raise RuntimeError( - "Accelerator daemon exited during startup " - f"(exit code {exit_code}). log: {log_tail!r}" - ) - if _socket_is_live(self._uds_path): - return - time.sleep(0.05) + # One channel for the whole wait. gRPC connects lazily and retries + # with backoff on its own, so the future below is re-polled rather + # than rebuilt; each short wait on it doubles as this loop's sleep. + channel = grpc.insecure_channel(_uds_target(self._uds_path)) + ready = grpc.channel_ready_future(channel) + try: + while time.monotonic() < deadline: + exit_code = self._proc.poll() + if exit_code is not None: + log_tail = self._read_log_tail() + raise RuntimeError( + "Accelerator daemon exited during startup " + f"(exit code {exit_code}). log: {log_tail!r}" + ) + try: + ready.result(timeout=_READY_POLL_SECONDS) + return + except grpc.FutureTimeoutError: + # Not up yet. Nothing distinguishes "still starting" from + # "never will" here — a failed connect leaves the channel + # in TRANSIENT_FAILURE and it keeps retrying — so the + # deadline above is what ends the wait. + continue + finally: + ready.cancel() + channel.close() log_tail = self._read_log_tail() raise RuntimeError( "Accelerator daemon did not become ready within " diff --git a/packages/google-cloud-bigtable/google/cloud/bigtable/data/_async/_accelerator_client.py b/packages/google-cloud-bigtable/google/cloud/bigtable/data/_async/_accelerator_client.py index 7b9b94993535..e6c2dd9db41c 100644 --- a/packages/google-cloud-bigtable/google/cloud/bigtable/data/_async/_accelerator_client.py +++ b/packages/google-cloud-bigtable/google/cloud/bigtable/data/_async/_accelerator_client.py @@ -21,6 +21,7 @@ from __future__ import annotations +from google.cloud.bigtable.data._accelerator._daemon import _uds_target from google.cloud.bigtable.data._accelerator._health import ( HEALTH_CHECK_METHOD, ServingStatus, @@ -58,7 +59,7 @@ class _AsyncAcceleratorClient: def __init__(self, uds_path: str, auth_secret: str): self._uds_path = uds_path self._metadata = (("x-accelerator-token", auth_secret),) - self._channel = insecure_channel(f"unix://{uds_path}") + self._channel = insecure_channel(_uds_target(uds_path)) self._mutate_row_stub = self._channel.unary_unary( _MUTATE_ROW_METHOD, request_serializer=MutateRowRequest.serialize, diff --git a/packages/google-cloud-bigtable/google/cloud/bigtable/data/_async/client.py b/packages/google-cloud-bigtable/google/cloud/bigtable/data/_async/client.py index ea9168abae6b..73c866f24553 100644 --- a/packages/google-cloud-bigtable/google/cloud/bigtable/data/_async/client.py +++ b/packages/google-cloud-bigtable/google/cloud/bigtable/data/_async/client.py @@ -20,7 +20,6 @@ import logging import os import random -import sys import time import warnings import weakref @@ -1384,13 +1383,6 @@ def _maybe_start_accelerator(self, *, explicit: bool) -> None: hard error rather than a silent fallback, both for the emulator (a genuine misconfiguration) and for a daemon that fails to start. """ - if sys.platform == "win32": - if explicit: - raise RuntimeError( - "use_accelerator=True is not supported on Windows; pass " - "use_accelerator=False to use the native client." - ) - return if self.client._emulator_host is not None: if explicit: raise RuntimeError( diff --git a/packages/google-cloud-bigtable/google/cloud/bigtable/data/_sync_autogen/_accelerator_client.py b/packages/google-cloud-bigtable/google/cloud/bigtable/data/_sync_autogen/_accelerator_client.py index bd0965843620..452fcc9c7b72 100644 --- a/packages/google-cloud-bigtable/google/cloud/bigtable/data/_sync_autogen/_accelerator_client.py +++ b/packages/google-cloud-bigtable/google/cloud/bigtable/data/_sync_autogen/_accelerator_client.py @@ -26,6 +26,7 @@ from grpc import insecure_channel +from google.cloud.bigtable.data._accelerator._daemon import _uds_target from google.cloud.bigtable.data._accelerator._health import ( HEALTH_CHECK_METHOD, ServingStatus, @@ -54,7 +55,7 @@ class _AcceleratorClient: def __init__(self, uds_path: str, auth_secret: str): self._uds_path = uds_path self._metadata = (("x-accelerator-token", auth_secret),) - self._channel = insecure_channel(f"unix://{uds_path}") + self._channel = insecure_channel(_uds_target(uds_path)) self._mutate_row_stub = self._channel.unary_unary( _MUTATE_ROW_METHOD, request_serializer=MutateRowRequest.serialize, diff --git a/packages/google-cloud-bigtable/google/cloud/bigtable/data/_sync_autogen/client.py b/packages/google-cloud-bigtable/google/cloud/bigtable/data/_sync_autogen/client.py index 67a8b3f12175..9f767ec87ca6 100644 --- a/packages/google-cloud-bigtable/google/cloud/bigtable/data/_sync_autogen/client.py +++ b/packages/google-cloud-bigtable/google/cloud/bigtable/data/_sync_autogen/client.py @@ -23,7 +23,6 @@ import logging import os import random -import sys import time import warnings import weakref @@ -1079,12 +1078,6 @@ def _maybe_start_accelerator(self, *, explicit: bool) -> None: ``use_accelerator=True``: a caller who required the accelerator gets a hard error rather than a silent fallback, both for the emulator (a genuine misconfiguration) and for a daemon that fails to start.""" - if sys.platform == "win32": - if explicit: - raise RuntimeError( - "use_accelerator=True is not supported on Windows; pass use_accelerator=False to use the native client." - ) - return if self.client._emulator_host is not None: if explicit: raise RuntimeError( diff --git a/packages/google-cloud-bigtable/tests/unit/data/_async/test__accelerator_health.py b/packages/google-cloud-bigtable/tests/unit/data/_async/test__accelerator_health.py index 458f80857ffa..c7e8a9a1a3c0 100644 --- a/packages/google-cloud-bigtable/tests/unit/data/_async/test__accelerator_health.py +++ b/packages/google-cloud-bigtable/tests/unit/data/_async/test__accelerator_health.py @@ -26,6 +26,8 @@ ``tests/unit/data/test_accelerator_health.py``. """ +import pathlib +import tempfile import threading import time from concurrent import futures @@ -68,6 +70,19 @@ } +# sockaddr_un.sun_path is 108 bytes on Linux and Windows (104 on macOS). +# Pytest's built-in tmp_path nests pytest-of-/pytest-N// under +# %TEMP%, which on Windows and macOS can exceed 107 chars and make +# add_insecure_port fail to bind. Matching the daemon's own +# tempfile.mkdtemp(prefix="bt-accel-") placement keeps the socket path well +# under the limit on every platform. +@pytest.fixture +def tmp_path(): + """Short UDS directory directly under the system temp root.""" + with tempfile.TemporaryDirectory(prefix="bt-accel-") as d: + yield pathlib.Path(d) + + class _FakeDaemonHealthServer: """A real gRPC server standing in for the daemon's health service. @@ -99,7 +114,9 @@ def __init__(self, uds_path, *, status=ServingStatus.SERVING, serve_health=True) self._server.add_generic_rpc_handlers( (grpc.method_handlers_generic_handler(service, {method: handler}),) ) - self._server.add_insecure_port(f"unix://{uds_path}") + # Single-colon form, same as the client — ``unix://`` cannot express a + # Windows path. See ``_accelerator_client._uds_target``. + self._server.add_insecure_port(f"unix:{uds_path}") self._server.start() def _check(self, request, context): diff --git a/packages/google-cloud-bigtable/tests/unit/data/_sync_autogen/test__accelerator_health.py b/packages/google-cloud-bigtable/tests/unit/data/_sync_autogen/test__accelerator_health.py index e2619cb9d523..e9508d0761be 100644 --- a/packages/google-cloud-bigtable/tests/unit/data/_sync_autogen/test__accelerator_health.py +++ b/packages/google-cloud-bigtable/tests/unit/data/_sync_autogen/test__accelerator_health.py @@ -29,6 +29,8 @@ ``tests/unit/data/test_accelerator_health.py``. """ +import pathlib +import tempfile import threading import time from concurrent import futures @@ -58,6 +60,13 @@ } +@pytest.fixture +def tmp_path(): + """Short UDS directory directly under the system temp root.""" + with tempfile.TemporaryDirectory(prefix="bt-accel-") as d: + yield pathlib.Path(d) + + class _FakeDaemonHealthServer: """A real gRPC server standing in for the daemon's health service. @@ -89,7 +98,7 @@ def __init__(self, uds_path, *, status=ServingStatus.SERVING, serve_health=True) self._server.add_generic_rpc_handlers( (grpc.method_handlers_generic_handler(service, {method: handler}),) ) - self._server.add_insecure_port(f"unix://{uds_path}") + self._server.add_insecure_port(f"unix:{uds_path}") self._server.start() def _check(self, request, context): diff --git a/packages/google-cloud-bigtable/tests/unit/data/test_accelerator_daemon.py b/packages/google-cloud-bigtable/tests/unit/data/test_accelerator_daemon.py index 2c3fb6a0108e..c8f4c8df638a 100644 --- a/packages/google-cloud-bigtable/tests/unit/data/test_accelerator_daemon.py +++ b/packages/google-cloud-bigtable/tests/unit/data/test_accelerator_daemon.py @@ -15,7 +15,6 @@ import json import os import re -import socket import unittest.mock import pytest @@ -205,6 +204,17 @@ def fake_mkdtemp(prefix=""): yield +@pytest.fixture +def no_af_unix(monkeypatch): + """Make the socket module look the way CPython's does on Windows. + + The family is simply absent there — the OS and gRPC's C-core support + AF_UNIX, the socket module just never exposed it — so every branch keyed on + ``hasattr(socket, "AF_UNIX")`` is reachable from a POSIX test run. + """ + monkeypatch.delattr(_daemon.socket, "AF_UNIX", raising=False) + + # --------------------------------------------------------------------------- # Properties before/after start # --------------------------------------------------------------------------- @@ -258,6 +268,17 @@ def test_missing_binary_raises_file_not_found(self, tmp_path): with pytest.raises(FileNotFoundError, match="Accelerator binary not found"): _resolve_binary_path() + @pytest.mark.parametrize( + "is_windows,expected", + [(False, "bin/accelerator"), (True, "bin/accelerator.exe")], + ) + def test_binary_name_follows_platform(self, monkeypatch, is_windows, expected): + """Windows wheels stage the binary under the .exe name Go gives it.""" + monkeypatch.setattr(_daemon, "_IS_WINDOWS", is_windows) + with unittest.mock.patch("os.path.isfile", return_value=True): + resolved = _daemon._resolve_binary_path() + assert resolved.endswith(expected) + # --------------------------------------------------------------------------- # start() failure modes @@ -308,30 +329,17 @@ def test_uds_never_appears_raises_timeout(self, tmp_path, monkeypatch): proc_mock.stdin = unittest.mock.Mock() monkeypatch.setattr(_daemon.subprocess, "Popen", lambda *a, **kw: proc_mock) - # Make os.path.exists always False so UDS never appears. - monkeypatch.setattr(_daemon.os.path, "exists", lambda p: False) + # Nothing ever binds the socket, so the readiness channel stays in + # TRANSIENT_FAILURE and the deadline is what ends the wait. monkeypatch.setenv(_STARTUP_TIMEOUT_ENV_VAR, "0.05") d = _make_daemon(tmp_path) with pytest.raises(RuntimeError, match="did not become ready"): d.start() - def test_no_af_unix_raises_os_error(self, tmp_path, monkeypatch): - import socket as _socket - - monkeypatch.delattr(_socket, "AF_UNIX", raising=False) - # Reload the attribute check in _daemon - monkeypatch.setattr(_daemon.socket, "AF_UNIX", None, raising=False) - # Simpler: just check that hasattr(socket, "AF_UNIX") guards the call. - with ( - unittest.mock.patch.object(_daemon.socket, "AF_UNIX", None, create=True), - unittest.mock.patch( - "google.cloud.bigtable.data._accelerator._daemon.hasattr", - return_value=False, - ), - ): - d = _make_daemon(tmp_path) - with pytest.raises(OSError, match="AF_UNIX"): - d.start() + def test_no_af_unix_raises_os_error(self, tmp_path, monkeypatch, no_af_unix): + d = _make_daemon(tmp_path) + with pytest.raises(OSError, match="AF_UNIX"): + d.start() # --------------------------------------------------------------------------- @@ -700,31 +708,105 @@ def test_daemon_constructor_invalid_env_var_raises(self, tmp_path, monkeypatch): _make_daemon(tmp_path) -@pytest.mark.skipif( - not hasattr(_daemon.socket, "AF_UNIX"), - reason="Unix domain sockets (AF_UNIX) are not supported on this platform", -) -@pytest.mark.skipif(not hasattr(_daemon.socket, "AF_UNIX"), reason="requires AF_UNIX") -class TestSocketIsLive: - """_socket_is_live backs the _wait_until_ready poll: is the daemon up?""" - - def test_true_while_listening(self, tmp_path): - path = str(tmp_path / "sock") - with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as sock: - sock.bind(path) - sock.listen(1) - assert _daemon._socket_is_live(path) is True - - def test_false_when_path_is_absent(self, tmp_path): - assert _daemon._socket_is_live(str(tmp_path / "nope")) is False - - def test_false_for_socket_file_whose_owner_exited(self, tmp_path): - # The socket file outlives the process that bound it; nothing is - # accepting, so the daemon is not up yet (or any more). - path = str(tmp_path / "sock") - sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) - sock.bind(path) - sock.listen(1) - sock.close() - assert os.path.exists(path) - assert _daemon._socket_is_live(path) is False +# --------------------------------------------------------------------------- +# Windows (win_amd64 / win32) +# --------------------------------------------------------------------------- + + +class TestWindows: + """Windows reaches the daemon over the same AF_UNIX socket as every other + platform. What differs is that CPython's socket module does not expose the + family, so nothing here may depend on ``socket.AF_UNIX`` — readiness goes + through gRPC's C-core, which does speak it.""" + + def test_start_is_not_blocked_by_missing_af_unix( + self, tmp_path, monkeypatch, no_af_unix + ): + """The AF_UNIX guard must not fire on Windows: a missing family there + says something about CPython, not about the platform.""" + monkeypatch.setattr(_daemon, "_IS_WINDOWS", True) + + def fake_popen(argv, **kwargs): + raise _StopStart() + + monkeypatch.setattr(_daemon.subprocess, "Popen", fake_popen) + d = _make_daemon(tmp_path) + # Reaching Popen means the guard let us through. + with pytest.raises(_StopStart): + d.start() + + def test_readiness_does_not_touch_socket_af_unix( + self, tmp_path, monkeypatch, no_af_unix + ): + """Readiness must work with no ``socket.AF_UNIX`` at all. + + This is the whole reason the probe is a gRPC channel: with the family + absent, the wait still has to reach its timeout honestly rather than + raise AttributeError or shortcut to "ready" off the bind artifact. + """ + import subprocess + + proc_mock = unittest.mock.Mock(spec=subprocess.Popen) + proc_mock.poll.return_value = None # alive, but never binds + proc_mock.stdin = unittest.mock.Mock() + monkeypatch.setattr(_daemon, "_IS_WINDOWS", True) + monkeypatch.setattr(_daemon.subprocess, "Popen", lambda *a, **kw: proc_mock) + monkeypatch.setenv(_STARTUP_TIMEOUT_ENV_VAR, "0.05") + d = _make_daemon(tmp_path) + with pytest.raises(RuntimeError, match="did not become ready"): + d.start() + + def test_readiness_reaches_a_real_grpc_server(self, tmp_path): + """And the positive case: a server that completes the HTTP/2 handshake + is what READY means, so the probe returns rather than timing out.""" + from concurrent import futures + + import grpc + + sock = tmp_path / _daemon._SOCKET_FILENAME + server = grpc.server(futures.ThreadPoolExecutor(max_workers=1)) + server.add_insecure_port(_daemon._uds_target(str(sock))) + server.start() + try: + d = _make_daemon(tmp_path) + d._proc = unittest.mock.Mock(**{"poll.return_value": None}) + d._uds_path = str(sock) + d._wait_until_ready(10.0) # returns, or the test times out + finally: + server.stop(None) + + def test_uds_target_accepts_a_windows_path(self): + """The channel target must survive a drive-lettered path. + + ``unix://`` opens a URI authority component, so ``unix://C:\\...`` + parses ``C:`` as the authority and gRPC refuses to build the channel at + all — the accelerator would start, verify, and then fail every routed + RPC. Constructing the channel is the assertion; it raises on a bad + target before any connection is attempted. + """ + import grpc + + from google.cloud.bigtable.data._async._accelerator_client import ( + _uds_target as _async_uds_target, + ) + from google.cloud.bigtable.data._sync_autogen._accelerator_client import ( + _uds_target, + ) + + win_path = r"C:\Users\me\AppData\Local\Temp\bt-accel-abc123\sock" + assert _uds_target(win_path) == "unix:" + win_path + assert _async_uds_target(win_path) == _uds_target(win_path) + channel = grpc.insecure_channel(_uds_target(win_path)) + try: + stub = channel.unary_unary( + "/google.bigtable.v2.Bigtable/MutateRow", + request_serializer=lambda _: b"", + response_deserializer=lambda payload: payload, + ) + with pytest.raises(grpc.RpcError) as exc_info: + stub(None, timeout=0.5) + finally: + channel.close() + # UNAVAILABLE means the target resolved and nothing was listening, which + # is the expected outcome. A malformed target fails earlier, as UNKNOWN. + assert exc_info.value.code() == grpc.StatusCode.UNAVAILABLE