Skip to content

Commit

Permalink
tests: move table API systests to own module
Browse files Browse the repository at this point in the history
Refactor to use pytest fixtures / idioms, rather than old 'Config'
setup / teardown.

Toward #472.
  • Loading branch information
tseaver committed Jul 30, 2021
1 parent 26c9de0 commit e38267b
Show file tree
Hide file tree
Showing 2 changed files with 69 additions and 60 deletions.
60 changes: 0 additions & 60 deletions tests/system/test_system.py
Original file line number Diff line number Diff line change
Expand Up @@ -41,7 +41,6 @@
from google.cloud.spanner_v1 import KeySet
from google.cloud.spanner_v1.instance import Backup
from google.cloud.spanner_v1.instance import Instance
from google.cloud.spanner_v1.table import Table
from google.cloud.spanner_v1 import RequestOptions

from test_utils.retry import RetryErrors
Expand Down Expand Up @@ -225,65 +224,6 @@ def _check_cell_data(self, found_cell, expected_cell):
self.assertEqual(found_cell, expected_cell)


class TestTableAPI(unittest.TestCase, _TestData):
DATABASE_NAME = "test_database" + unique_resource_id("_")

@classmethod
def setUpClass(cls):
pool = BurstyPool(labels={"testcase": "database_api"})
ddl_statements = EMULATOR_DDL_STATEMENTS if USE_EMULATOR else DDL_STATEMENTS
cls._db = Config.INSTANCE.database(
cls.DATABASE_NAME, ddl_statements=ddl_statements, pool=pool
)
operation = cls._db.create()
operation.result(30) # raises on failure / timeout.

@classmethod
def tearDownClass(cls):
cls._db.drop()

def test_exists(self):
table = Table("all_types", self._db)
self.assertTrue(table.exists())

def test_exists_not_found(self):
table = Table("table_does_not_exist", self._db)
self.assertFalse(table.exists())

def test_list_tables(self):
tables = self._db.list_tables()
table_ids = set(table.table_id for table in tables)
self.assertIn("contacts", table_ids)
self.assertIn("contact_phones", table_ids)
self.assertIn("all_types", table_ids)

def test_list_tables_reload(self):
tables = self._db.list_tables()
for table in tables:
self.assertTrue(table.exists())
schema = table.schema
self.assertIsInstance(schema, list)

def test_reload_not_found(self):
table = Table("table_does_not_exist", self._db)
with self.assertRaises(exceptions.NotFound):
table.reload()

def test_schema(self):
table = Table("all_types", self._db)
schema = table.schema
names_and_types = set((field.name, field.type_.code) for field in schema)
self.assertIn(("pkey", TypeCode.INT64), names_and_types)
self.assertIn(("int_value", TypeCode.INT64), names_and_types)
self.assertIn(("int_array", TypeCode.ARRAY), names_and_types)
self.assertIn(("bool_value", TypeCode.BOOL), names_and_types)
self.assertIn(("bytes_value", TypeCode.BYTES), names_and_types)
self.assertIn(("date_value", TypeCode.DATE), names_and_types)
self.assertIn(("float_value", TypeCode.FLOAT64), names_and_types)
self.assertIn(("string_value", TypeCode.STRING), names_and_types)
self.assertIn(("timestamp_value", TypeCode.TIMESTAMP), names_and_types)


@unittest.skipIf(USE_EMULATOR, "Skipping backup tests")
@unittest.skipIf(SKIP_BACKUP_TESTS, "Skipping backup tests")
class TestBackupAPI(unittest.TestCase, _TestData):
Expand Down
69 changes: 69 additions & 0 deletions tests/system/test_table_api.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,69 @@
# Copyright 2021 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.

import pytest

from google.api_core import exceptions
from google.cloud import spanner_v1


def test_table_exists(shared_database):
table = shared_database.table("all_types")
assert table.exists()


def test_table_exists_not_found(shared_database):
table = shared_database.table("table_does_not_exist")
assert not table.exists()


def test_db_list_tables(shared_database):
tables = shared_database.list_tables()
table_ids = set(table.table_id for table in tables)
assert "contacts" in table_ids
assert "contact_phones" in table_ids
assert "all_types" in table_ids


def test_db_list_tables_reload(shared_database):
for table in shared_database.list_tables():
assert table.exists()
schema = table.schema
assert isinstance(schema, list)


def test_table_reload_miss(shared_database):
table = shared_database.table("table_does_not_exist")
with pytest.raises(exceptions.NotFound):
table.reload()


def test_table_schema(shared_database):
table = shared_database.table("all_types")
schema = table.schema
expected = [
("pkey", spanner_v1.TypeCode.INT64),
("int_value", spanner_v1.TypeCode.INT64),
("int_array", spanner_v1.TypeCode.ARRAY),
("bool_value", spanner_v1.TypeCode.BOOL),
("bytes_value", spanner_v1.TypeCode.BYTES),
("date_value", spanner_v1.TypeCode.DATE),
("float_value", spanner_v1.TypeCode.FLOAT64),
("string_value", spanner_v1.TypeCode.STRING),
("timestamp_value", spanner_v1.TypeCode.TIMESTAMP),
]
found = {field.name: field.type_.code for field in schema}

for field_name, type_code in expected:
assert found[field_name] == type_code

0 comments on commit e38267b

Please sign in to comment.