test: increase the package testing coverage (#161)

google/cloud/sqlalchemy_spanner/_opentelemetry_tracing.py

@@ -52,9 +52,9 @@ def trace_call(name, extra_attributes=None):
         name, kind=trace.SpanKind.CLIENT, attributes=attributes
     ) as span:
         try:
-            span.set_status(Status(StatusCode.OK))
             yield span
         except GoogleAPICallError as error:
             span.set_status(Status(StatusCode.ERROR))
             span.record_exception(error)
             raise
+        span.set_status(Status(StatusCode.OK))

google/cloud/sqlalchemy_spanner/provision.py

@@ -15,6 +15,6 @@
 from sqlalchemy.testing.provision import temp_table_keyword_args
 
 
-@temp_table_keyword_args.for_db("spanner")
+@temp_table_keyword_args.for_db("spanner")  # pragma: no cover
 def _spanner_temp_table_keyword_args(cfg, eng):
     return {"prefixes": ["TEMPORARY"]}

google/cloud/sqlalchemy_spanner/requirements.py

@@ -16,7 +16,7 @@
 from sqlalchemy.testing.requirements import SuiteRequirements
 
 
-class Requirements(SuiteRequirements):
+class Requirements(SuiteRequirements):  # pragma: no cover
     @property
     def computed_columns(self):
         return exclusions.open()

google/cloud/sqlalchemy_spanner/sqlalchemy_spanner.py

@@ -373,10 +373,14 @@ def visit_TEXT(self, type_, **kw):
     def visit_ARRAY(self, type_, **kw):
         return "ARRAY<{}>".format(self.process(type_.item_type, **kw))
 
-    def visit_BINARY(self, type_, **kw):
+    def visit_BINARY(self, type_, **kw):  # pragma: no cover
+        """
+        The BINARY type is superseded by large_binary in
+        newer versions of SQLAlchemy (>1.4).
+        """
         return "BYTES({})".format(type_.length or "MAX")
 
-    def visit_large_binary(self, type_, **kw):
+    def visit_large_binary(self, type_, **kw):  # pragma: no cover
         return "BYTES({})".format(type_.length or "MAX")
 
     def visit_DECIMAL(self, type_, **kw):

noxfile.py

@@ -128,11 +128,25 @@ def compliance_test(session):
             "Credentials or emulator host must be set via environment variable"
         )
 
+    session.install(
+        "pytest", "pytest-cov", "pytest-asyncio",
+    )
+
     session.install("pytest")
     session.install("mock")
-    session.install("-e", ".")
+    session.install("-e", ".[tracing]")
     session.run("python", "create_test_database.py")
-    session.run("pytest", "-v")
+
+    session.run(
+        "py.test",
+        "--cov=google.cloud.sqlalchemy_spanner",
+        "--cov=tests",
+        "--cov-append",
+        "--cov-config=.coveragerc",
+        "--cov-report=",
+        "--cov-fail-under=0",
+        "test",
+    )
 
 
 @nox.session(python=DEFAULT_PYTHON_VERSION)

test/test_suite.py

@@ -20,6 +20,7 @@
 import os
 import pkg_resources
 import pytest
+import unittest
 from unittest import mock
 
 import sqlalchemy
@@ -726,6 +727,7 @@ def test_reflect_bytes_column_max_len(self):
         self.metadata.create_all()
 
         Table("bytes_table", MetaData(bind=self.bind), autoload=True)
+        inspect(config.db).get_columns("bytes_table")
 
     @testing.provide_metadata
     def _test_get_unique_constraints(self, schema=None):
@@ -1576,24 +1578,25 @@ def test_user_agent(self):
             )
 
 
-class ExecutionOptionsTest(fixtures.TestBase):
+class ExecutionOptionsTest(fixtures.TestBase, unittest.TestCase):
     """
     Check that `execution_options()` method correctly
     sets parameters on the underlying DB API connection.
     """
 
-    def setUp(self):
-        self._engine = create_engine(get_db_url(), pool_size=1)
-        self._metadata = MetaData(bind=self._engine)
+    @classmethod
+    def setUpClass(cls):
+        cls._engine = create_engine(get_db_url(), pool_size=1)
+        cls._metadata = MetaData(bind=cls._engine)
 
-        self._table = Table(
+        cls._table = Table(
             "execution_options",
-            self._metadata,
+            cls._metadata,
             Column("opt_id", Integer, primary_key=True),
             Column("opt_name", String(16), nullable=False),
         )
 
-        self._metadata.create_all(self._engine)
+        cls._metadata.create_all(cls._engine)
 
     def test_read_only(self):
         with self._engine.connect().execution_options(read_only=True) as connection:
@@ -1602,11 +1605,11 @@ def test_read_only(self):
 
     def test_staleness(self):
         with self._engine.connect().execution_options(
-            read_only=True, staleness={"max_staleness": datetime.timedelta(seconds=5)}
+            read_only=True, staleness={"exact_staleness": datetime.timedelta(seconds=5)}
         ) as connection:
             connection.execute(select(["*"], from_obj=self._table)).fetchall()
             assert connection.connection.staleness == {
-                "max_staleness": datetime.timedelta(seconds=5)
+                "exact_staleness": datetime.timedelta(seconds=5)
             }
 
         with self._engine.connect() as connection:
@@ -1638,6 +1641,26 @@ def test_offset_only(self):
                 list(connection.execute(self._table.select().offset(offset)).fetchall())
 
 
+class TemporaryTableTest(fixtures.TestBase):
+    """
+    Check that temporary tables raise an error on creation.
+    """
+
+    def setUp(self):
+        self._engine = create_engine(get_db_url(), pool_size=1)
+        self._metadata = MetaData(bind=self._engine)
+
+    def test_temporary_prefix(self):
+        with pytest.raises(NotImplementedError):
+            Table(
+                "users",
+                self._metadata,
+                Column("user_id", Integer, primary_key=True),
+                Column("user_name", String(16), nullable=False),
+                prefixes=["TEMPORARY"],
+            ).create()
+
+
 class ComputedReflectionFixtureTest(_ComputedReflectionFixtureTest):
     @classmethod
     def define_tables(cls, metadata):