spanner

Author: provinzkraut

src/pytest_databases/docker/docker-compose.spanner.yml

@@ -1,142 +1,59 @@
 from __future__ import annotations
 
-import contextlib
-import os
-import sys
-from pathlib import Path
+from dataclasses import dataclass
 from typing import TYPE_CHECKING
 
 import pytest
+from google.api_core.client_options import ClientOptions
 from google.auth.credentials import AnonymousCredentials, Credentials
 from google.cloud import spanner
 
-from pytest_databases.docker import DockerServiceRegistry
-from pytest_databases.helpers import simple_string_hash
+from pytest_databases.types import ServiceContainer
 
 if TYPE_CHECKING:
     from collections.abc import Generator
 
+    from pytest_databases._service import DockerService
 
-COMPOSE_PROJECT_NAME: str = f"pytest-databases-spanner-{simple_string_hash(__file__)}"
 
+@dataclass
+class SpannerService(ServiceContainer):
+    credentials: Credentials
 
-def spanner_responsive(
-    host: str,
-    spanner_port: int,
-    spanner_instance: str,
-    spanner_database: str,
-    spanner_project: str,
-    spanner_credentials: Credentials,
-) -> bool:
-    try:
-        spanner_client = spanner.Client(project=spanner_project, credentials=spanner_credentials)
-        instance = spanner_client.instance(spanner_instance)
-        with contextlib.suppress(Exception):
-            instance.create()
-
-        database = instance.database(spanner_database)
-        with contextlib.suppress(Exception):
-            database.create()
-
-        with database.snapshot() as snapshot:
-            resp = next(iter(snapshot.execute_sql("SELECT 1")))
-        return resp[0] == 1
-    except Exception:  # noqa: BLE001
-        return False
-
+    @property
+    def endpoint(self) -> str:
+        return f"{self.host}:{self.port}"
 
-@pytest.fixture(scope="session")
-def spanner_compose_project_name() -> str:
-    return os.environ.get("COMPOSE_PROJECT_NAME", COMPOSE_PROJECT_NAME)
+    @property
+    def client_options(self) -> ClientOptions:
+        return ClientOptions(api_endpoint=self.endpoint)
 
 
 @pytest.fixture(autouse=False, scope="session")
-def spanner_docker_services(
-    spanner_compose_project_name: str, worker_id: str = "main"
-) -> Generator[DockerServiceRegistry, None, None]:
-    if os.getenv("GITHUB_ACTIONS") == "true" and sys.platform != "linux":
-        pytest.skip("Docker not available on this platform")
-
-    with DockerServiceRegistry(worker_id, compose_project_name=spanner_compose_project_name) as registry:
-        yield registry
-
-
-@pytest.fixture(scope="session")
-def spanner_port() -> int:
-    return 9010
-
-
-@pytest.fixture(scope="session")
-def spanner_instance() -> str:
-    return "test-instance"
-
-
-@pytest.fixture(scope="session")
-def spanner_database() -> str:
-    return "test-database"
-
-
-@pytest.fixture(scope="session")
-def spanner_project() -> str:
-    return "emulator-test-project"
-
-
-@pytest.fixture(scope="session")
-def spanner_credentials() -> Credentials:
-    return AnonymousCredentials()
-
-
-@pytest.fixture(scope="session")
-def spanner_docker_compose_files() -> list[Path]:
-    return [Path(Path(__file__).parent / "docker-compose.spanner.yml")]
-
-
-@pytest.fixture(scope="session")
-def default_spanner_service_name() -> str:
-    return "spanner"
-
-
-@pytest.fixture(scope="session")
-def spanner_docker_ip(spanner_docker_services: DockerServiceRegistry) -> str:
-    return spanner_docker_services.docker_ip
-
-
-@pytest.fixture(autouse=False, scope="session")
-def spanner_service(
-    spanner_docker_services: DockerServiceRegistry,
-    default_spanner_service_name: str,
-    spanner_docker_compose_files: list[Path],
-    spanner_docker_ip: str,
-    spanner_port: int,
-    spanner_instance: str,
-    spanner_database: str,
-    spanner_project: str,
-    spanner_credentials: Credentials,
-) -> Generator[None, None, None]:
-    os.environ["SPANNER_EMULATOR_HOST"] = f"{spanner_docker_ip}:{spanner_port}"
-    os.environ["SPANNER_DATABASE"] = spanner_database
-    os.environ["SPANNER_INSTANCE"] = spanner_instance
-    os.environ["SPANNER_PORT"] = str(spanner_port)
-    os.environ["GOOGLE_CLOUD_PROJECT"] = spanner_project
-    spanner_docker_services.start(
-        name=default_spanner_service_name,
-        docker_compose_files=spanner_docker_compose_files,
-        timeout=60,
-        check=spanner_responsive,
-        spanner_port=spanner_port,
-        spanner_instance=spanner_instance,
-        spanner_database=spanner_database,
-        spanner_project=spanner_project,
-        spanner_credentials=spanner_credentials,
-    )
-    yield
+def spanner_service(docker_service: DockerService) -> Generator[SpannerService, None, None]:
+    with docker_service.run(
+        image="gcr.io/cloud-spanner-emulator/emulator:latest",
+        name="spanner",
+        container_port=9010,
+        wait_for_log="gRPC server listening at",
+    ) as service:
+        yield SpannerService(
+            host=service.host,
+            port=service.port,
+            credentials=AnonymousCredentials(),
+        )
 
 
 @pytest.fixture(autouse=False, scope="session")
 def spanner_startup_connection(
-    spanner_service: DockerServiceRegistry,
-    spanner_project: str,
-    spanner_credentials: Credentials,
+    spanner_service: SpannerService,
 ) -> Generator[spanner.Client, None, None]:
-    c = spanner.Client(project=spanner_project, credentials=spanner_credentials)
-    yield c
+    client = spanner.Client(
+        project=spanner_service.project,
+        credentials=spanner_service.credentials,
+        client_options=spanner_service.client_options,
+    )
+    try:
+        yield client
+    finally:
+        client.close()

src/pytest_databases/docker/spanner.py

@@ -1,48 +1,37 @@
 from __future__ import annotations
 
+import contextlib
 from typing import TYPE_CHECKING
 
 from google.cloud import spanner
 
-from pytest_databases.docker.spanner import spanner_responsive
-
 if TYPE_CHECKING:
-    from google.auth.credentials import Credentials
-
-    from pytest_databases.docker import DockerServiceRegistry
+    from pytest_databases.docker.spanner import SpannerService
 
 pytest_plugins = [
     "pytest_databases.docker.spanner",
 ]
 
 
-def test_spanner_default_config(
-    spanner_port: int, spanner_instance: str, spanner_database: str, spanner_project: str
-) -> None:
-    assert spanner_port == 9010
-    assert spanner_instance == "test-instance"
-    assert spanner_database == "test-database"
-    assert spanner_project == "emulator-test-project"
-
-
 def test_spanner_services(
-    spanner_docker_ip: str,
-    spanner_service: DockerServiceRegistry,
-    spanner_port: int,
-    spanner_instance: str,
-    spanner_database: str,
-    spanner_project: str,
-    spanner_credentials: Credentials,
+    spanner_service: SpannerService,
 ) -> None:
-    ping = spanner_responsive(
-        spanner_docker_ip,
-        spanner_port=spanner_port,
-        spanner_instance=spanner_instance,
-        spanner_database=spanner_database,
-        spanner_project=spanner_project,
-        spanner_credentials=spanner_credentials,
+    spanner_client = spanner.Client(
+        project=spanner_service.project,
+        credentials=spanner_service.credentials,
+        client_options=spanner_service.client_options,
     )
-    assert ping
+    instance = spanner_client.instance(spanner_service.instance_name)
+    with contextlib.suppress(Exception):
+        instance.create()
+
+    database = instance.database(spanner_service.database_name)
+    with contextlib.suppress(Exception):
+        database.create()
+
+    with database.snapshot() as snapshot:
+        resp = next(iter(snapshot.execute_sql("SELECT 1")))
+    assert resp[0] == 1
 
 
 def test_spanner_service_after_start(