spannermockserver 0.1.2

A lightweight in-memory mock server for Google Cloud Spanner. This is an internal library that can make breaking changes without prior notice.

Spanner Python Mockserver

NOTICE: This is an internal library that can make breaking changes without prior notice.

Introduction

The spannermockserver is a lightweight in-memory mock server for Google Cloud Spanner.

Installation

To install the library, use pip:

pip install spannermockserver

Usage

Here is a simple example of how to use the mock server to test your Spanner application.

1. Start the Mock Server

from spannermockserver import start_mock_server

# Start the server
server, spanner_servicer, database_admin_servicer, port = start_mock_server()

2. Configure Your Client

Configure your Spanner client to connect to the mock server.

import os
from google.cloud import spanner
from google.auth.credentials import AnonymousCredentials

# Set environment variable to use the emulator/mock server
os.environ["SPANNER_EMULATOR_HOST"] = f"localhost:{port}"

# Create a client with anonymous credentials
client = spanner.Client(
    project="test-project",
    credentials=AnonymousCredentials()
)
instance = client.instance("test-instance")
database = instance.database("test-database")

3. Mock Results

Use spanner_servicer.mock_spanner to define the results for your queries.

from google.cloud.spanner_v1.types import ResultSet, ResultSetMetadata, StructType, Type, TypeCode

# Define the SQL you expect
sql = "SELECT 1"

# Create a mock result set with schema definition
metadata = ResultSetMetadata(
    row_type=StructType(
        fields=[
            StructType.Field(
                name="Result",
                type=Type(code=TypeCode.INT64)
            )
        ]
    )
)
result = ResultSet(metadata=metadata)
result.rows.append(["1"])

# Add the result to the mock spanner
spanner_servicer.mock_spanner.add_result(sql, result)

4. Run Your Code

Execute your application code that queries Spanner.

with database.snapshot() as snapshot:
    results = snapshot.execute_sql(sql)
    for row in results:
        print(f"Row: {row}")

5. Verify Requests (Optional)

You can inspect spanner_servicer.requests to verify that your code made the expected calls.

print(f"Total requests received: {len(spanner_servicer.requests)}")

6. Stop the Server

server.stop(None)

API Reference

start_mock_server()

Starts the gRPC server and returns a tuple containing:

• server: The gRPC server instance.
• spanner_servicer: The SpannerServicer instance.
• database_admin_servicer: The DatabaseAdminServicer instance.
• port: The port number the server is listening on.

SpannerServicer

The implementation of the Spanner gRPC service.

• mock_spanner: Access the MockSpanner instance to configure behavior.
• requests: A list of all request objects received by the servicer.
• clear_requests(): Clears the list of received requests.

MockSpanner

Handles the in-memory state and responses.

• add_result(sql, result): Maps a SQL string to a ResultSet response.
• add_error(method, error): Maps a gRPC method name (e.g., "ExecuteSql") to an error (an instance of grpc_status.rpc_status._Status) to simulate failures.