Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

tests: avoid using real credentials unit tests #432

Merged
merged 2 commits into from Jul 29, 2021
Merged

tests: avoid using real credentials unit tests #432

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
603a377
tests: avoid using real credentials unit tests
tseaver Jul 28, 2021
31c987b
Merge branch 'master' into 431-unittests-avoid-env-credentials
tseaver Jul 28, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
181 changes: 77 additions & 104 deletions tests/unit/spanner_dbapi/test_connect.py
View file
Open in desktop
Expand Up @@ -20,6 +20,12 @@
import google.auth.credentials


INSTANCE = "test-instance"
DATABASE = "test-database"
PROJECT = "test-project"
USER_AGENT = "user-agent"


def _make_credentials():
class _CredentialsWithScopes(
google.auth.credentials.Credentials, google.auth.credentials.Scoped
Expand All @@ -29,138 +35,105 @@ class _CredentialsWithScopes(
return mock.Mock(spec=_CredentialsWithScopes)


@mock.patch("google.cloud.spanner_v1.Client")
class Test_connect(unittest.TestCase):
def test_connect(self):
def test_w_implicit(self, mock_client):
from google.cloud.spanner_dbapi import connect
from google.cloud.spanner_dbapi import Connection

PROJECT = "test-project"
USER_AGENT = "user-agent"
CREDENTIALS = _make_credentials()

with mock.patch("google.cloud.spanner_v1.Client") as client_mock:
connection = connect(
"test-instance",
"test-database",
PROJECT,
CREDENTIALS,
user_agent=USER_AGENT,
)
client = mock_client.return_value
instance = client.instance.return_value
database = instance.database.return_value

self.assertIsInstance(connection, Connection)

client_mock.assert_called_once_with(
project=PROJECT, credentials=CREDENTIALS, client_info=mock.ANY
)

def test_instance_not_found(self):
from google.cloud.spanner_dbapi import connect
connection = connect(INSTANCE, DATABASE)

with mock.patch(
"google.cloud.spanner_v1.instance.Instance.exists", return_value=False,
) as exists_mock:
self.assertIsInstance(connection, Connection)

with self.assertRaises(ValueError):
connect("test-instance", "test-database")
self.assertIs(connection.instance, instance)
client.instance.assert_called_once_with(INSTANCE)

exists_mock.assert_called_once_with()
self.assertIs(connection.database, database)
instance.database.assert_called_once_with(DATABASE, pool=None)
# Datbase constructs its own pool
self.assertIsNotNone(connection.database._pool)

def test_database_not_found(self):
def test_w_explicit(self, mock_client):
from google.cloud.spanner_v1.pool import AbstractSessionPool
from google.cloud.spanner_dbapi import connect
from google.cloud.spanner_dbapi import Connection
from google.cloud.spanner_dbapi.version import PY_VERSION

with mock.patch(
"google.cloud.spanner_v1.instance.Instance.exists", return_value=True,
):
with mock.patch(
"google.cloud.spanner_v1.database.Database.exists", return_value=False,
) as exists_mock:

with self.assertRaises(ValueError):
connect("test-instance", "test-database")

exists_mock.assert_called_once_with()
credentials = _make_credentials()
pool = mock.create_autospec(AbstractSessionPool)
client = mock_client.return_value
instance = client.instance.return_value
database = instance.database.return_value

def test_connect_instance_id(self):
from google.cloud.spanner_dbapi import connect
from google.cloud.spanner_dbapi import Connection
connection = connect(
INSTANCE, DATABASE, PROJECT, credentials, pool=pool, user_agent=USER_AGENT,
)

INSTANCE = "test-instance"
self.assertIsInstance(connection, Connection)

with mock.patch(
"google.cloud.spanner_v1.client.Client.instance"
) as instance_mock:
connection = connect(INSTANCE, "test-database")
mock_client.assert_called_once_with(
project=PROJECT, credentials=credentials, client_info=mock.ANY
)
client_info = mock_client.call_args_list[0][1]["client_info"]
self.assertEqual(client_info.user_agent, USER_AGENT)
self.assertEqual(client_info.python_version, PY_VERSION)

instance_mock.assert_called_once_with(INSTANCE)
self.assertIs(connection.instance, instance)
client.instance.assert_called_once_with(INSTANCE)

self.assertIsInstance(connection, Connection)
self.assertIs(connection.database, database)
instance.database.assert_called_once_with(DATABASE, pool=pool)

def test_connect_database_id(self):
def test_w_instance_not_found(self, mock_client):
from google.cloud.spanner_dbapi import connect
from google.cloud.spanner_dbapi import Connection

DATABASE = "test-database"

with mock.patch(
"google.cloud.spanner_v1.instance.Instance.database"
) as database_mock:
with mock.patch(
"google.cloud.spanner_v1.instance.Instance.exists", return_value=True,
):
connection = connect("test-instance", DATABASE)
client = mock_client.return_value
instance = client.instance.return_value
instance.exists.return_value = False

database_mock.assert_called_once_with(DATABASE, pool=mock.ANY)
with self.assertRaises(ValueError):
connect(INSTANCE, DATABASE)

self.assertIsInstance(connection, Connection)
instance.exists.assert_called_once_with()

def test_default_sessions_pool(self):
def test_w_database_not_found(self, mock_client):
from google.cloud.spanner_dbapi import connect

with mock.patch("google.cloud.spanner_v1.instance.Instance.database"):
with mock.patch(
"google.cloud.spanner_v1.instance.Instance.exists", return_value=True,
):
connection = connect("test-instance", "test-database")
client = mock_client.return_value
instance = client.instance.return_value
database = instance.database.return_value
database.exists.return_value = False

self.assertIsNotNone(connection.database._pool)
with self.assertRaises(ValueError):
connect(INSTANCE, DATABASE)

def test_sessions_pool(self):
database.exists.assert_called_once_with()

def test_w_credential_file_path(self, mock_client):
from google.cloud.spanner_dbapi import connect
from google.cloud.spanner_v1.pool import FixedSizePool
from google.cloud.spanner_dbapi import Connection
from google.cloud.spanner_dbapi.version import PY_VERSION

database_id = "test-database"
pool = FixedSizePool()
credentials_path = "dummy/file/path.json"

with mock.patch(
"google.cloud.spanner_v1.instance.Instance.database"
) as database_mock:
with mock.patch(
"google.cloud.spanner_v1.instance.Instance.exists", return_value=True,
):
connect("test-instance", database_id, pool=pool)
database_mock.assert_called_once_with(database_id, pool=pool)
connection = connect(
INSTANCE,
DATABASE,
PROJECT,
credentials=credentials_path,
user_agent=USER_AGENT,
)

def test_connect_w_credential_file_path(self):
from google.cloud.spanner_dbapi import connect
from google.cloud.spanner_dbapi import Connection
self.assertIsInstance(connection, Connection)

PROJECT = "test-project"
USER_AGENT = "user-agent"
credentials = "dummy/file/path.json"

with mock.patch(
"google.cloud.spanner_v1.Client.from_service_account_json"
) as client_mock:
connection = connect(
"test-instance",
"test-database",
PROJECT,
credentials=credentials,
user_agent=USER_AGENT,
)

self.assertIsInstance(connection, Connection)

client_mock.assert_called_once_with(
credentials, project=PROJECT, client_info=mock.ANY
)
factory = mock_client.from_service_account_json
factory.assert_called_once_with(
credentials_path, project=PROJECT, client_info=mock.ANY,
)
client_info = factory.call_args_list[0][1]["client_info"]
self.assertEqual(client_info.user_agent, USER_AGENT)
self.assertEqual(client_info.python_version, PY_VERSION)