iframe-proxy | Sunbelt Computer Software

This repository was archived by the owner on Jun 8, 2026. It is now read-only.
277 lines (194 loc) · 9.01 KB
# Copyright 2025 Google LLC All rights reserved.
# 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.
from datetime import datetime
import inspect
from logging import Logger
from typing import Mapping
from google.auth.credentials import Credentials, Scoped
from mock import create_autospec
from google.cloud._helpers import _datetime_to_pb_timestamp
from google.cloud.aio._cross_sync import CrossSync
from google.cloud.spanner_dbapi import Connection
from google.cloud.spanner_v1 import SpannerAsyncClient, SpannerClient
from google.cloud.spanner_v1._async.client import Client as AsyncClient
from google.cloud.spanner_v1._async.database import Database as AsyncDatabase
from google.cloud.spanner_v1._async.instance import Instance as AsyncInstance
from google.cloud.spanner_v1._async.session import Session as AsyncSession
from google.cloud.spanner_v1.client import Client
from google.cloud.spanner_v1.database import Database
from google.cloud.spanner_v1.instance import Instance
from google.cloud.spanner_v1.session import Session
from google.cloud.spanner_v1.transaction import Transaction
from google.cloud.spanner_v1.types import CommitResponse as CommitResponsePB
from google.cloud.spanner_v1.types import Session as SessionPB
from google.cloud.spanner_v1.types import Transaction as TransactionPB
from google.cloud.spanner_v1.types import (
    MultiplexedSessionPrecommitToken as PrecommitTokenPB,
# Default values used to populate required or expected attributes.
# Tests should not depend on them: if a test requires a specific
# identifier or name, it should set it explicitly.
_PROJECT_ID = "default-project-id"
_INSTANCE_ID = "default-instance-id"
_DATABASE_ID = "default-database-id"
_SESSION_ID = "default-session-id"
_PROJECT_NAME = "projects/" + _PROJECT_ID
_INSTANCE_NAME = _PROJECT_NAME + "/instances/" + _INSTANCE_ID
_DATABASE_NAME = _INSTANCE_NAME + "/databases/" + _DATABASE_ID
_SESSION_NAME = _DATABASE_NAME + "/sessions/" + _SESSION_ID
_TRANSACTION_ID = b"default-transaction-id"
_PRECOMMIT_TOKEN = b"default-precommit-token"
_SEQUENCE_NUMBER = -1
_TIMESTAMP = _datetime_to_pb_timestamp(datetime.now())
# Protocol buffers
# ----------------
def build_commit_response_pb(**kwargs) -> CommitResponsePB:
    """Builds and returns a commit response protocol buffer for testing using the given arguments.
    If an expected argument is not provided, a default value will be used."""
    if "commit_timestamp" not in kwargs:
        kwargs["commit_timestamp"] = _TIMESTAMP
    return CommitResponsePB(**kwargs)
def build_precommit_token_pb(**kwargs) -> PrecommitTokenPB:
    """Builds and returns a multiplexed session precommit token protocol buffer for
    testing using the given arguments. If an expected argument is not provided, a
    default value will be used."""
    if "precommit_token" not in kwargs:
        kwargs["precommit_token"] = _PRECOMMIT_TOKEN
    if "seq_num" not in kwargs:
        kwargs["seq_num"] = _SEQUENCE_NUMBER
    return PrecommitTokenPB(**kwargs)
def build_session_pb(**kwargs) -> SessionPB:
    """Builds and returns a session protocol buffer for testing using the given arguments.
    If an expected argument is not provided, a default value will be used."""
    if "name" not in kwargs:
        kwargs["name"] = _SESSION_NAME
    return SessionPB(**kwargs)
def build_transaction_pb(**kwargs) -> TransactionPB:
    """Builds and returns a transaction protocol buffer for testing using the given arguments..
    If an expected argument is not provided, a default value will be used."""
    if "id" not in kwargs:
        kwargs["id"] = _TRANSACTION_ID
    return TransactionPB(**kwargs)
# Client classes
# --------------
def _is_async_test_caller():
    # Check if the caller is in an async test directory
    for frame in inspect.stack():
        module = inspect.getmodule(frame[0])
        if module:
            filename = getattr(module, "__file__", "")
            if filename and "tests/unit/_async" in filename:
                return True
            if "_async" in module.__name__:
                return True
    return False
def build_client(**kwargs: Mapping) -> Client:
    """Builds and returns a client for testing using the given arguments.
    If a required argument is not provided, a default value will be used."""
    if "project" not in kwargs:
        kwargs["project"] = _PROJECT_ID
    if "credentials" not in kwargs:
        kwargs["credentials"] = build_scoped_credentials()
    is_async = CrossSync.is_async
    if is_async and not _is_async_test_caller():
        is_async = False
    cls = AsyncClient if is_async else Client
    return cls(**kwargs)
def build_connection(**kwargs: Mapping) -> Connection:
    """Builds and returns a connection for testing using the given arguments.
    If a required argument is not provided, a default value will be used."""
    if "instance" not in kwargs:
        kwargs["instance"] = build_instance()
    if "database" not in kwargs:
        kwargs["database"] = build_database(instance=kwargs["instance"])
    return Connection(**kwargs)
def build_database(**kwargs: Mapping) -> Database:
    """Builds and returns a database for testing using the given arguments.
    If a required argument is not provided, a default value will be used."""
    if "database_id" not in kwargs:
        kwargs["database_id"] = _DATABASE_ID
    if "logger" not in kwargs:
        kwargs["logger"] = build_logger()
    if "instance" not in kwargs:
        kwargs["instance"] = build_instance()
    is_async = CrossSync.is_async
    if is_async and not _is_async_test_caller():
        is_async = False
    cls = AsyncDatabase if is_async else Database
    database = cls(**kwargs)
    database._spanner_api = build_spanner_api()
    if not is_async:
        database._pool.bind(database)
    return database
def build_instance(**kwargs: Mapping) -> Instance:
    """Builds and returns an instance for testing using the given arguments.
    If a required argument is not provided, a default value will be used."""
    if "instance_id" not in kwargs:
        kwargs["instance_id"] = _INSTANCE_ID
    if "client" not in kwargs:
        kwargs["client"] = build_client()
    is_async = CrossSync.is_async
    if is_async and not _is_async_test_caller():
        is_async = False
    cls = AsyncInstance if is_async else Instance
    return cls(**kwargs)
def build_session(**kwargs: Mapping) -> Session:
    """Builds and returns a session for testing using the given arguments.
    If a required argument is not provided, a default value will be used."""
    if "database" not in kwargs:
        kwargs["database"] = build_database()
    is_async = CrossSync.is_async
    if is_async and not _is_async_test_caller():
        is_async = False
    cls = AsyncSession if is_async else Session
    return cls(**kwargs)
def build_snapshot(**kwargs):
    """Builds and returns a snapshot for testing using the given arguments.
    If a required argument is not provided, a default value will be used."""
    session = kwargs.pop("session", build_session())
    # Ensure session exists.
    if session.session_id is None:
        session._session_id = _SESSION_ID
    return session.snapshot(**kwargs)
def build_transaction(session=None) -> Transaction:
    """Builds and returns a transaction for testing using the given arguments.
    If a required argument is not provided, a default value will be used."""
    session = session or build_session()
    # Ensure session exists.
    if session.session_id is None:
        session._session_id = _SESSION_ID
    return session.transaction()
# Other classes
# -------------
def build_logger() -> Logger:
    """Builds and returns a logger for testing."""
    return create_autospec(Logger, instance=True)
def build_scoped_credentials() -> Credentials:
    """Builds and returns a mock scoped credentials for testing."""
    class _ScopedCredentials(Credentials, Scoped):
        pass
    return create_autospec(spec=_ScopedCredentials, instance=True)
def build_spanner_api() -> SpannerClient:
    """Builds and returns a mock Spanner Client API for testing using the given arguments.
    Commonly used methods are mocked to return default values."""
    is_async = CrossSync.is_async
    if is_async and not _is_async_test_caller():
        is_async = False
    cls = SpannerAsyncClient if is_async else SpannerClient
    api = create_autospec(cls, instance=True)
    # Mock API calls with default return values.
    api.begin_transaction.return_value = build_transaction_pb()
    api.commit.return_value = build_commit_response_pb()
    api.create_session.return_value = build_session_pb()
    return api
Sunbelt Computer Software

PL/B Language Development and Support

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

Sunbelt Computer Software

PL/B Language Development and Support

Uh oh!

FilesExpand file tree

_builders.py

Latest commit

History

_builders.py

File metadata and controls
