# Path: spanner_orm/api.py
# class SpannerRetryableApi(abc.ABC):
# class SpannerReadApi(SpannerRetryableApi):
# class SpannerWriteApi(SpannerRetryableApi):
# class SpannerConnection:
# class SpannerApi(SpannerReadApi, SpannerWriteApi):
#   def _ensure_session(self, api_method, *args, **kwargs):
#   def _connection(self) -> spanner_database.Database:
#   def run_read_only(self, method: Callable[..., CallableReturn], *args: Any,
#                     **kwargs: Any) -> CallableReturn:
#   def _run_read_only(self, method, *args, **kwargs):
#   def _connection(self) -> spanner_database.Database:
#   def run_write(self, method: Callable[..., CallableReturn], *args: Any,
#                 **kwargs: Any) -> CallableReturn:
#   def __init__(
#       self,
#       instance: str,
#       database: str,
#       project: Optional[str] = None,
#       credentials: Optional[auth_credentials.Credentials] = None,
#       pool: Optional[spanner_pool.AbstractSessionPool] = None,
#       create_ddl: Optional[Iterable[str]] = None,
#       *,
#       client_options: Union[api_client_options.ClientOptions, Dict[Any, Any],
#                             None] = None,
#   ):
#   def connect(self):
#   def __init__(self, connection: SpannerConnection):
#   def spanner_connection(self) -> SpannerConnection:
#   def _connection(self):
# def connect(
#     instance: str,
#     database: str,
#     project: Optional[str] = None,
#     credentials: Optional[auth_credentials.Credentials] = None,
#     pool: Optional[spanner_pool.AbstractSessionPool] = None) -> SpannerApi:
# def from_connection(connection: SpannerConnection) -> SpannerApi:
# def hangup() -> None:
# def spanner_api() -> SpannerApi:
# 
# Path: spanner_orm/error.py
# class SpannerError(Exception):
# class ValidationError(Exception):
# 
# Path: spanner_orm/admin/api.py
# class SpannerAdminApi(api.SpannerReadApi, api.SpannerWriteApi):
#   def __init__(self, connection: api.SpannerConnection):
#   def _connection(self) -> spanner_database.Database:
#   def create_database(self) -> None:
#   def drop_database(self) -> None:
#   def update_schema(self, change: str) -> None:
#   def execute_partitioned_dml(self, dml: str) -> None:
# def connect(instance: str,
#             database: str,
#             project: Optional[str] = None,
#             credentials: Optional[auth_credentials.Credentials] = None,
#             pool: Optional[spanner_pool.AbstractSessionPool] = None,
#             create_ddl: Optional[Iterable[str]] = None) -> SpannerAdminApi:
# def from_connection(connection: api.SpannerConnection) -> SpannerAdminApi:
# def hangup() -> None:
# def spanner_admin_api() -> SpannerAdminApi:


import logging
import unittest
import warnings
from unittest import mock
from absl.testing import parameterized
from google.api_core import exceptions
from google.cloud import spanner  # pytype: disable=import-error
from spanner_orm import api
from spanner_orm import error
from spanner_orm.admin import api as admin_api
            ),
        ),
        client.mock_calls,
    )

  @mock.patch('google.cloud.spanner.Client')
  def test_api_connection(self, client):
    connection = self.mock_connection(client)
    with warnings.catch_warnings(record=True) as connect_warnings:
      api.connect('', '', '')
    self.assertEqual(api.spanner_api()._connection, connection)
    self.assertLen(connect_warnings, 1)
    connect_warning, = connect_warnings
    self.assertIn('spanner_orm.from_connection', str(connect_warning.message))
    self.assertIs(DeprecationWarning, connect_warning.category)
    self.assertEqual(api.__file__, connect_warning.filename)

    api.hangup()
    with self.assertRaises(error.SpannerError):
      api.spanner_api()

  def test_api_error_when_not_connected(self):
    with self.assertRaises(error.SpannerError):
      api.spanner_api()

  @mock.patch('google.cloud.spanner.Client')
  def test_admin_api_connection(self, client):
    connection = self.mock_connection(client)
    with warnings.catch_warnings(record=True) as connect_warnings:
      admin_api.connect('', '', '')

# Path: spanner_orm/api.py
# class SpannerRetryableApi(abc.ABC):
# class SpannerReadApi(SpannerRetryableApi):
# class SpannerWriteApi(SpannerRetryableApi):
# class SpannerConnection:
# class SpannerApi(SpannerReadApi, SpannerWriteApi):
#   def _ensure_session(self, api_method, *args, **kwargs):
#   def _connection(self) -> spanner_database.Database:
#   def run_read_only(self, method: Callable[..., CallableReturn], *args: Any,
#                     **kwargs: Any) -> CallableReturn:
#   def _run_read_only(self, method, *args, **kwargs):
#   def _connection(self) -> spanner_database.Database:
#   def run_write(self, method: Callable[..., CallableReturn], *args: Any,
#                 **kwargs: Any) -> CallableReturn:
#   def __init__(
#       self,
#       instance: str,
#       database: str,
#       project: Optional[str] = None,
#       credentials: Optional[auth_credentials.Credentials] = None,
#       pool: Optional[spanner_pool.AbstractSessionPool] = None,
#       create_ddl: Optional[Iterable[str]] = None,
#       *,
#       client_options: Union[api_client_options.ClientOptions, Dict[Any, Any],
#                             None] = None,
#   ):
#   def connect(self):
#   def __init__(self, connection: SpannerConnection):
#   def spanner_connection(self) -> SpannerConnection:
#   def _connection(self):
# def connect(
#     instance: str,
#     database: str,
#     project: Optional[str] = None,
#     credentials: Optional[auth_credentials.Credentials] = None,
#     pool: Optional[spanner_pool.AbstractSessionPool] = None) -> SpannerApi:
# def from_connection(connection: SpannerConnection) -> SpannerApi:
# def hangup() -> None:
# def spanner_api() -> SpannerApi:
# 
# Path: spanner_orm/error.py
# class SpannerError(Exception):
# class ValidationError(Exception):
# 
# Path: spanner_orm/admin/api.py
# class SpannerAdminApi(api.SpannerReadApi, api.SpannerWriteApi):
#   def __init__(self, connection: api.SpannerConnection):
#   def _connection(self) -> spanner_database.Database:
#   def create_database(self) -> None:
#   def drop_database(self) -> None:
#   def update_schema(self, change: str) -> None:
#   def execute_partitioned_dml(self, dml: str) -> None:
# def connect(instance: str,
#             database: str,
#             project: Optional[str] = None,
#             credentials: Optional[auth_credentials.Credentials] = None,
#             pool: Optional[spanner_pool.AbstractSessionPool] = None,
#             create_ddl: Optional[Iterable[str]] = None) -> SpannerAdminApi:
# def from_connection(connection: api.SpannerConnection) -> SpannerAdminApi:
# def hangup() -> None:
# def spanner_admin_api() -> SpannerAdminApi:


# Path: spanner_orm/tests/api_test.py
import logging
import unittest
import warnings
from unittest import mock
from absl.testing import parameterized
from google.api_core import exceptions
from google.cloud import spanner  # pytype: disable=import-error
from spanner_orm import api
from spanner_orm import error
from spanner_orm.admin import api as admin_api

            ),
        ),
        client.mock_calls,
    )

  @mock.patch('google.cloud.spanner.Client')
  def test_api_connection(self, client):
    connection = self.mock_connection(client)
    with warnings.catch_warnings(record=True) as connect_warnings:
      api.connect('', '', '')
    self.assertEqual(api.spanner_api()._connection, connection)
    self.assertLen(connect_warnings, 1)
    connect_warning, = connect_warnings
    self.assertIn('spanner_orm.from_connection', str(connect_warning.message))
    self.assertIs(DeprecationWarning, connect_warning.category)
    self.assertEqual(api.__file__, connect_warning.filename)

    api.hangup()
    with self.assertRaises(error.SpannerError):
      api.spanner_api()

  def test_api_error_when_not_connected(self):
    with self.assertRaises(error.SpannerError):
      api.spanner_api()

  @mock.patch('google.cloud.spanner.Client')
  def test_admin_api_connection(self, client):
    connection = self.mock_connection(client)
    with warnings.catch_warnings(record=True) as connect_warnings:
      admin_api.connect('', '', '')

    self.assertEqual(admin_api.spanner_admin_api()._connection, connection)