test: fix integer compliance tests

google/cloud/sqlalchemy_spanner/sqlalchemy_spanner.py

@@ -146,6 +146,9 @@ def visit_BOOLEAN(self, type_, **kw):
     def visit_DATETIME(self, type_, **kw):
         return "TIMESTAMP"
 
+    def visit_BIGINT(self, type_, **kw):
+        return "INT64"
+
 
 class SpannerDialect(DefaultDialect):
     """Cloud Spanner dialect.

test/test_suite.py

@@ -16,18 +16,20 @@
 
 import pytest
 
-from sqlalchemy.testing import config
+from sqlalchemy.testing import config, db
 from sqlalchemy.testing import eq_
 from sqlalchemy.testing import provide_metadata
 from sqlalchemy.testing.schema import Column
 from sqlalchemy.testing.schema import Table
 from sqlalchemy import literal_column
-from sqlalchemy import select, case, bindparam
+
+from sqlalchemy import bindparam, case, literal, select, util
 from sqlalchemy import exists
 from sqlalchemy import Boolean
 from sqlalchemy import String
-from sqlalchemy.testing import requires
 from sqlalchemy.types import Integer
+from sqlalchemy.testing import requires
+
 from google.api_core.datetime_helpers import DatetimeWithNanoseconds
 
 from sqlalchemy.testing.suite.test_ddl import *  # noqa: F401, F403
@@ -45,6 +47,8 @@
 from sqlalchemy.testing.suite.test_select import ExistsTest as _ExistsTest
 from sqlalchemy.testing.suite.test_types import BooleanTest as _BooleanTest
 
+from sqlalchemy.testing.suite.test_types import IntegerTest as _IntegerTest
+
 
 from sqlalchemy.testing.suite.test_types import (  # noqa: F401, F403
     DateTest as _DateTest,
@@ -420,3 +424,87 @@ class TimeTests(_TimeMicrosecondsTest, _TimeTest):
 @pytest.mark.skip("Spanner doesn't coerce dates from datetime.")
 class DateTimeCoercedToDateTimeTest(_DateTimeCoercedToDateTimeTest):
     pass
+
+
+class IntegerTest(_IntegerTest):
+    @provide_metadata
+    def _round_trip(self, datatype, data):
+        """
+        SPANNER OVERRIDE:
+
+        This is the helper method for integer class tests which creates table and
+        perform insert operation.
+        Cloud Spanner supports tables with empty primary key, but only single one
+        row can be inserted into such a table - following insertions will fail with
+        `400 id must not be NULL in table date_table`.
+        Overriding the tests and add a manual primary key value to avoid the same
+        failures and delete the from the table at the end.
+        """
+        metadata = self.metadata
+        int_table = Table(
+            "integer_table",
+            metadata,
+            Column("id", Integer, primary_key=True, test_needs_autoincrement=True),
+            Column("integer_data", datatype),
+        )
+
+        metadata.create_all(config.db)
+
+        config.db.execute(int_table.insert(), {"id": 1, "integer_data": data})
+
+        row = config.db.execute(select([int_table.c.integer_data])).first()
+
+        eq_(row, (data,))
+
+        if util.py3k:
+            assert isinstance(row[0], int)
+        else:
+            assert isinstance(row[0], (long, int))  # noqa
+
+        config.db.execute(int_table.delete())
+
+    @provide_metadata
+    def _literal_round_trip(self, type_, input_, output, filter_=None):
+        """
+        SPANNER OVERRIDE:
+
+        Spanner is not able cleanup data and drop the table correctly,
+        table was already exists after related tests finished, so it doesn't
+        create a new table and when started tests for other data type  following
+        insertions will fail with `400 Duplicate name in schema: t.
+        Overriding the tests to create a new table for test and drop table manually
+        before it creates a new table to avoid the same failures.
+        """
+
+        # for literal, we test the literal render in an INSERT
+        # into a typed column.  we can then SELECT it back as its
+        # official type; ideally we'd be able to use CAST here
+        # but MySQL in particular can't CAST fully
+        t = Table("int_t", self.metadata, Column("x", type_))
+        t.drop(checkfirst=True)
+        t.create()
+
+        with db.connect() as conn:
+            for value in input_:
+                ins = (
+                    t.insert()
+                    .values(x=literal(value))
+                    .compile(
+                        dialect=db.dialect, compile_kwargs=dict(literal_binds=True),
+                    )
+                )
+                conn.execute(ins)
+
+            if self.supports_whereclause:
+                stmt = t.select().where(t.c.x == literal(value))
+            else:
+                stmt = t.select()
+
+            stmt = stmt.compile(
+                dialect=db.dialect, compile_kwargs=dict(literal_binds=True),
+            )
+            for row in conn.execute(stmt):
+                value = row[0]
+                if filter_ is not None:
+                    value = filter_(value)
+                assert value in output