# -*- coding: utf-8 -*-

#

# Copyright 2021 Google LLC

#

# 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

#

# https://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 datetime

import decimal

import operator

import os

import pytest

import random

import time

from unittest import mock

from google.cloud.spanner_v1 import RequestOptions, Client

import sqlalchemy

from sqlalchemy import create_engine

from sqlalchemy import inspect

from sqlalchemy import testing

from sqlalchemy import ForeignKey

from sqlalchemy import MetaData

from sqlalchemy.schema import DDL

from sqlalchemy.schema import Computed

from sqlalchemy.testing import config

from sqlalchemy.testing import engines

from sqlalchemy.testing import eq_

from sqlalchemy.testing import is_instance_of

from sqlalchemy.testing import provide_metadata, emits_warning

from sqlalchemy.testing import fixtures

from sqlalchemy.testing import is_true

from sqlalchemy.testing.schema import Column

from sqlalchemy.testing.schema import Table

from sqlalchemy import literal_column

from sqlalchemy import select

from sqlalchemy import util

from sqlalchemy import event

from sqlalchemy import exists

from sqlalchemy import Boolean

from sqlalchemy import Float

from sqlalchemy import LargeBinary

from sqlalchemy import String

from sqlalchemy.ext.declarative import declarative_base

from sqlalchemy.orm import relation

from sqlalchemy.orm import Session

from sqlalchemy.types import ARRAY

from sqlalchemy.types import Integer

from sqlalchemy.types import Numeric

from sqlalchemy.types import Text

from sqlalchemy.testing import requires

from sqlalchemy.testing.fixtures import (

ComputedReflectionFixtureTest as _ComputedReflectionFixtureTest,

)

from google.api_core.datetime_helpers import DatetimeWithNanoseconds

from google.cloud import spanner_dbapi

from sqlalchemy.testing.suite.test_cte import * # noqa: F401, F403

from sqlalchemy.testing.suite.test_ddl import * # noqa: F401, F403

from sqlalchemy.testing.suite.test_dialect import * # noqa: F401, F403

from sqlalchemy.testing.suite.test_insert import * # noqa: F401, F403

from sqlalchemy.testing.suite.test_reflection import * # noqa: F401, F403

from sqlalchemy.testing.suite.test_results import * # noqa: F401, F403

from sqlalchemy.testing.suite.test_select import * # noqa: F401, F403

from sqlalchemy.testing.suite.test_sequence import (

SequenceTest as _SequenceTest,

HasSequenceTest as _HasSequenceTest,

) # noqa: F401, F403

from sqlalchemy.testing.suite.test_update_delete import * # noqa: F401, F403

from sqlalchemy.testing.suite.test_cte import CTETest as _CTETest

from sqlalchemy.testing.suite.test_ddl import TableDDLTest as _TableDDLTest

from sqlalchemy.testing.suite.test_ddl import (

LongNameBlowoutTest as _LongNameBlowoutTest,

)

from sqlalchemy.testing.suite.test_dialect import EscapingTest as _EscapingTest

from sqlalchemy.testing.suite.test_insert import (

InsertBehaviorTest as _InsertBehaviorTest,

)

from sqlalchemy.testing.suite.test_select import ( # noqa: F401, F403

CompoundSelectTest as _CompoundSelectTest,

ExistsTest as _ExistsTest,

IsOrIsNotDistinctFromTest as _IsOrIsNotDistinctFromTest,

LikeFunctionsTest as _LikeFunctionsTest,

OrderByLabelTest as _OrderByLabelTest,

)

from sqlalchemy.testing.suite.test_reflection import (

QuotedNameArgumentTest as _QuotedNameArgumentTest,

ComponentReflectionTest as _ComponentReflectionTest,

CompositeKeyReflectionTest as _CompositeKeyReflectionTest,

ComputedReflectionTest as _ComputedReflectionTest,

)

from sqlalchemy.testing.suite.test_results import RowFetchTest as _RowFetchTest

from sqlalchemy.testing.suite.test_types import ( # noqa: F401, F403

_DateFixture as _DateFixtureTest,

_LiteralRoundTripFixture,

_UnicodeFixture as _UnicodeFixtureTest,

BooleanTest as _BooleanTest,

DateTest as _DateTest,

DateTimeHistoricTest,

DateTimeCoercedToDateTimeTest as _DateTimeCoercedToDateTimeTest,

DateTimeMicrosecondsTest as _DateTimeMicrosecondsTest,

DateTimeTest as _DateTimeTest,

IntegerTest as _IntegerTest,

JSONTest as _JSONTest,

NumericTest as _NumericTest,

StringTest as _StringTest,

TextTest as _TextTest,

TimeTest as _TimeTest,

TimeMicrosecondsTest as _TimeMicrosecondsTest,

TimestampMicrosecondsTest,

UnicodeVarcharTest as _UnicodeVarcharTest,

UnicodeTextTest as _UnicodeTextTest,

)

from test._helpers import get_db_url, get_project

from google.cloud.sqlalchemy_spanner import version as sqlalchemy_spanner_version

config.test_schema = ""

class EscapingTest(_EscapingTest):

@provide_metadata

def test_percent_sign_round_trip(self):

"""Test that the DBAPI accommodates for escaped / nonescaped

percent signs in a way that matches the compiler

SPANNER OVERRIDE

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 `Row [] already exists".

Overriding the test to avoid the same failure.

"""

m = self.metadata

t = Table("t", m, Column("data", String(50)))

t.create(config.db)

with config.db.begin() as conn:

conn.execute(t.insert(), dict(data="some % value"))

eq_(

conn.scalar(

select([t.c.data]).where(

t.c.data == literal_column("'some % value'")

)

),

"some % value",

)

conn.execute(t.delete())

conn.execute(t.insert(), dict(data="some %% other value"))

eq_(

conn.scalar(

select([t.c.data]).where(

t.c.data == literal_column("'some %% other value'")

)

),

"some %% other value",

)

class CTETest(_CTETest):

@classmethod

def define_tables(cls, metadata):

"""

The original method creates a foreign key without a name,

which causes troubles on test cleanup. Overriding the

method to explicitly set a foreign key name.

"""

Table(

"some_table",

metadata,

Column("id", Integer, primary_key=True),

Column("data", String(50)),

Column("parent_id", ForeignKey("some_table.id", name="fk_some_table")),

)

Table(

"some_other_table",

metadata,

Column("id", Integer, primary_key=True),

Column("data", String(50)),

Column("parent_id", Integer),

)

@pytest.mark.skip("INSERT from WITH subquery is not supported")

def test_insert_from_select_round_trip(self):

"""

The test checks if an INSERT can be done from a cte, like:

WITH some_cte AS (...)

INSERT INTO some_other_table (... SELECT * FROM some_cte)

Such queries are not supported by Spanner.

"""

pass

@pytest.mark.skip("DELETE from WITH subquery is not supported")

def test_delete_scalar_subq_round_trip(self):

"""

The test checks if a DELETE can be done from a cte, like:

WITH some_cte AS (...)

DELETE FROM some_other_table (... SELECT * FROM some_cte)

Such queries are not supported by Spanner.

"""

pass

@pytest.mark.skip("DELETE from WITH subquery is not supported")

def test_delete_from_round_trip(self):

"""

The test checks if a DELETE can be done from a cte, like:

WITH some_cte AS (...)

DELETE FROM some_other_table (... SELECT * FROM some_cte)

Such queries are not supported by Spanner.

"""

pass

@pytest.mark.skip("UPDATE from WITH subquery is not supported")

def test_update_from_round_trip(self):

"""

The test checks if an UPDATE can be done from a cte, like:

WITH some_cte AS (...)

UPDATE some_other_table

SET (... SELECT * FROM some_cte)

Such queries are not supported by Spanner.

"""

pass

@pytest.mark.skip("WITH RECURSIVE subqueries are not supported")

def test_select_recursive_round_trip(self):

pass

class BooleanTest(_BooleanTest):

def test_render_literal_bool(self):

"""

SPANNER OVERRIDE:

Cloud Spanner supports tables with an empty primary key, but

only a single row can be inserted into such a table -

following insertions will fail with `Row [] already exists".

Overriding the test to avoid the same failure.

"""

self._literal_round_trip(Boolean(), [True], [True])

self._literal_round_trip(Boolean(), [False], [False])

class ExistsTest(_ExistsTest):

def test_select_exists(self, connection):

"""

SPANNER OVERRIDE:

The original test is trying to execute a query like:

SELECT ...

WHERE EXISTS (SELECT ...)

SELECT WHERE without FROM clause is not supported by Spanner.

Rewriting the test to force it to generate a query like:

SELECT EXISTS (SELECT ...)

"""

stuff = self.tables.stuff

eq_(

connection.execute(

select((exists().where(stuff.c.data == "some data"),))

).fetchall(),

[(True,)],

)

def test_select_exists_false(self, connection):

"""

SPANNER OVERRIDE:

The original test is trying to execute a query like:

SELECT ...

WHERE EXISTS (SELECT ...)

SELECT WHERE without FROM clause is not supported by Spanner.

Rewriting the test to force it to generate a query like:

SELECT EXISTS (SELECT ...)

"""

stuff = self.tables.stuff

eq_(

connection.execute(

select((exists().where(stuff.c.data == "no data"),))

).fetchall(),

[(False,)],

)

class TableDDLTest(_TableDDLTest):

@pytest.mark.skip(

"Spanner table name must start with an uppercase or lowercase letter"

)

def test_underscore_names(self):

pass

@pytest.mark.skip("Max identifier length in Spanner is 128")

class LongNameBlowoutTest(_LongNameBlowoutTest):

pass

class DateFixtureTest(_DateFixtureTest):

@classmethod

def define_tables(cls, metadata):

"""

SPANNER OVERRIDE:

Cloud Spanner doesn't support auto incrementing ids feature,

which is used by the original test. Overriding the test data

creation method to disable autoincrement and make id column

nullable.

"""

Table(

"date_table",

metadata,

Column("id", Integer, primary_key=True, nullable=True),

Column("date_data", cls.datatype),

)

class DateTest(DateFixtureTest, _DateTest):

"""

SPANNER OVERRIDE:

DateTest tests used same class method to create table, so to avoid those failures

and maintain DRY concept just inherit the class to run tests successfully.

"""

@pytest.mark.skipif(

bool(os.environ.get("SPANNER_EMULATOR_HOST")), reason="Skipped on emulator"

)

def test_null_bound_comparison(self):

super().test_null_bound_comparison()

@pytest.mark.skipif(

bool(os.environ.get("SPANNER_EMULATOR_HOST")), reason="Skipped on emulator"

)

def test_null(self):

super().test_null()

class DateTimeMicrosecondsTest(_DateTimeMicrosecondsTest, DateTest):

def test_round_trip(self):

"""

SPANNER OVERRIDE:

Spanner converts timestamp into `%Y-%m-%dT%H:%M:%S.%fZ` format, so to avoid

assert failures convert datetime input to the desire timestamp format.

"""

date_table = self.tables.date_table

config.db.execute(date_table.insert(), {"date_data": self.data})

row = config.db.execute(select([date_table.c.date_data])).first()

compare = self.compare or self.data

compare = compare.strftime("%Y-%m-%dT%H:%M:%S.%fZ")

eq_(row[0].rfc3339(), compare)

assert isinstance(row[0], DatetimeWithNanoseconds)

@pytest.mark.skipif(

bool(os.environ.get("SPANNER_EMULATOR_HOST")), reason="Skipped on emulator"

)

def test_null_bound_comparison(self):

super().test_null_bound_comparison()

@pytest.mark.skipif(

bool(os.environ.get("SPANNER_EMULATOR_HOST")), reason="Skipped on emulator"

)

def test_null(self):

super().test_null()

class DateTimeTest(_DateTimeTest, DateTimeMicrosecondsTest):

"""

SPANNER OVERRIDE:

DateTimeTest tests have the same failures same as DateTimeMicrosecondsTest tests,

so to avoid those failures and maintain DRY concept just inherit the class to run

tests successfully.

"""

@pytest.mark.skipif(

bool(os.environ.get("SPANNER_EMULATOR_HOST")), reason="Skipped on emulator"

)

def test_null_bound_comparison(self):

super().test_null_bound_comparison()

@pytest.mark.skipif(

bool(os.environ.get("SPANNER_EMULATOR_HOST")), reason="Skipped on emulator"

)

def test_null(self):

super().test_null()

@pytest.mark.skip("Spanner doesn't support Time data type.")

class TimeTests(_TimeMicrosecondsTest, _TimeTest):

pass

@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 a table and

performs an insert operation.

Cloud Spanner supports tables with an empty primary key, but only 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 adding a manual primary key value to avoid the same

failures.

"""

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

class UnicodeFixtureTest(_UnicodeFixtureTest):

@classmethod

def define_tables(cls, metadata):

"""

SPANNER OVERRIDE:

Cloud Spanner doesn't support auto incrementing ids feature,

which is used by the original test. Overriding the test data

creation method to disable autoincrement and make id column

nullable.

"""

Table(

"unicode_table",

metadata,

Column("id", Integer, primary_key=True, nullable=True),

Column("unicode_data", cls.datatype),

)

def test_round_trip_executemany(self):

"""

SPANNER OVERRIDE

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 `Row [] already exists".

Overriding the test to avoid the same failure.

"""

unicode_table = self.tables.unicode_table

config.db.execute(

unicode_table.insert(),

[{"id": i, "unicode_data": self.data} for i in range(3)],

)

rows = config.db.execute(select([unicode_table.c.unicode_data])).fetchall()

eq_(rows, [(self.data,) for i in range(3)])

for row in rows:

assert isinstance(row[0], util.text_type)

@pytest.mark.skip("Spanner doesn't support non-ascii characters")

def test_literal(self):

pass

@pytest.mark.skip("Spanner doesn't support non-ascii characters")

def test_literal_non_ascii(self):

pass

class UnicodeVarcharTest(UnicodeFixtureTest, _UnicodeVarcharTest):

"""

SPANNER OVERRIDE:

UnicodeVarcharTest class inherits the _UnicodeFixtureTest class's tests,

so to avoid those failures and maintain DRY concept just inherit the class to run

tests successfully.

"""

pass

class UnicodeTextTest(UnicodeFixtureTest, _UnicodeTextTest):

"""

SPANNER OVERRIDE:

UnicodeTextTest class inherits the _UnicodeFixtureTest class's tests,

so to avoid those failures and maintain DRY concept just inherit the class to run

tests successfully.

"""

pass

class ComponentReflectionTest(_ComponentReflectionTest):

@classmethod

def define_views(cls, metadata, schema):

table_info = {

"users": ["user_id", "test1", "test2"],

"email_addresses": ["address_id", "remote_user_id", "email_address"],

}

if testing.requires.self_referential_foreign_keys.enabled:

table_info["users"] = table_info["users"] + ["parent_user_id"]

for table_name in ("users", "email_addresses"):

fullname = table_name

if schema:

fullname = "%s.%s" % (schema, table_name)

view_name = fullname + "_v"

columns = ""

for column in table_info[table_name]:

stmt = table_name + "." + column + " AS " + column

if columns:

columns = columns + ", " + stmt

else:

columns = stmt

query = f"""CREATE VIEW {view_name}

SQL SECURITY INVOKER

AS SELECT {columns}

FROM {fullname}"""

event.listen(metadata, "after_create", DDL(query))

event.listen(metadata, "before_drop", DDL("DROP VIEW %s" % view_name))

@classmethod

def define_reflected_tables(cls, metadata, schema):

if schema:

schema_prefix = schema + "."

else:

schema_prefix = ""

if testing.requires.self_referential_foreign_keys.enabled:

users = Table(

"users",

metadata,

Column("user_id", sqlalchemy.INT, primary_key=True),

Column("test1", sqlalchemy.CHAR(5), nullable=False),

Column("test2", sqlalchemy.Float(5), nullable=False),

Column(

"parent_user_id",

sqlalchemy.Integer,

sqlalchemy.ForeignKey(

"%susers.user_id" % schema_prefix, name="user_id_fk"

),

),

schema=schema,

test_needs_fk=True,

)

else:

users = Table(

"users",

metadata,

Column("user_id", sqlalchemy.INT, primary_key=True),

Column("test1", sqlalchemy.CHAR(5), nullable=False),

Column("test2", sqlalchemy.Float(5), nullable=False),

schema=schema,

test_needs_fk=True,

)

Table(

"dingalings",

metadata,

Column("dingaling_id", sqlalchemy.Integer, primary_key=True),

Column(

"address_id",

sqlalchemy.Integer,

sqlalchemy.ForeignKey("%semail_addresses.address_id" % schema_prefix),

),

Column("data", sqlalchemy.String(30)),

schema=schema,

test_needs_fk=True,

)

Table(

"email_addresses",

metadata,

Column("address_id", sqlalchemy.Integer, primary_key=True),

Column(

"remote_user_id",

sqlalchemy.Integer,

sqlalchemy.ForeignKey(users.c.user_id),

),

Column("email_address", sqlalchemy.String(20)),

sqlalchemy.PrimaryKeyConstraint("address_id", name="email_ad_pk"),

schema=schema,

test_needs_fk=True,

)

Table(

"comment_test",

metadata,

Column("id", sqlalchemy.Integer, primary_key=True, comment="id comment"),

Column("data", sqlalchemy.String(20), comment="data % comment"),

Column(

"d2",

sqlalchemy.String(20),

comment=r"""Comment types type speedily ' " \ '' Fun!""",

),

schema=schema,

comment=r"""the test % ' " \ table comment""",

)

if testing.requires.cross_schema_fk_reflection.enabled:

if schema is None:

Table(

"local_table",

metadata,

Column("id", sqlalchemy.Integer, primary_key=True),

Column("data", sqlalchemy.String(20)),

Column(

"remote_id",

ForeignKey("%s.remote_table_2.id" % testing.config.test_schema),

),

test_needs_fk=True,

schema=config.db.dialect.default_schema_name,

)

else:

Table(

"remote_table",

metadata,

Column("id", sqlalchemy.Integer, primary_key=True),

Column(

"local_id",

ForeignKey(

"%s.local_table.id" % config.db.dialect.default_schema_name

),

),

Column("data", sqlalchemy.String(20)),

schema=schema,

test_needs_fk=True,

)

Table(

"remote_table_2",

metadata,

Column("id", sqlalchemy.Integer, primary_key=True),

Column("data", sqlalchemy.String(20)),

schema=schema,

test_needs_fk=True,

)

if testing.requires.index_reflection.enabled:

cls.define_index(metadata, users)

if not schema:

# test_needs_fk is at the moment to force MySQL InnoDB

noncol_idx_test_nopk = Table(

"noncol_idx_test_nopk",

metadata,

Column("id", sqlalchemy.Integer, primary_key=True),

Column("q", sqlalchemy.String(5)),

test_needs_fk=True,

)

noncol_idx_test_pk = Table(

"noncol_idx_test_pk",

metadata,

Column("id", sqlalchemy.Integer, primary_key=True),

Column("q", sqlalchemy.String(5)),

test_needs_fk=True,

)

if testing.requires.indexes_with_ascdesc.enabled:

sqlalchemy.Index("noncol_idx_nopk", noncol_idx_test_nopk.c.q.desc())

sqlalchemy.Index("noncol_idx_pk", noncol_idx_test_pk.c.q.desc())

if testing.requires.view_column_reflection.enabled and not bool(

os.environ.get("SPANNER_EMULATOR_HOST")

):

cls.define_views(metadata, schema)

if not schema and testing.requires.temp_table_reflection.enabled:

cls.define_temp_tables(metadata)

def _test_get_columns(self, schema=None, table_type="table"):

if table_type == "view" and bool(os.environ.get("SPANNER_EMULATOR_HOST")):

pytest.skip("View tables not supported on emulator")

super()._test_get_columns(schema, table_type)

@testing.provide_metadata

def _test_get_view_definition(self, schema=None):

if bool(os.environ.get("SPANNER_EMULATOR_HOST")):

pytest.skip("View tables not supported on emulator")

super()._test_get_view_definition(schema)

@classmethod

def define_temp_tables(cls, metadata):

"""

SPANNER OVERRIDE:

In Cloud Spanner unique indexes are used instead of directly

creating unique constraints. Overriding the test to replace

constraints with indexes in testing data.

"""

user_tmp = Table(

"user_tmp",

metadata,

Column("id", sqlalchemy.INT, primary_key=True),

Column("name", sqlalchemy.VARCHAR(50)),

Column("foo", sqlalchemy.INT),

sqlalchemy.Index("user_tmp_uq", "name", unique=True),

sqlalchemy.Index("user_tmp_ix", "foo"),

)

if (

testing.requires.view_reflection.enabled

and testing.requires.temporary_views.enabled

):

event.listen(

user_tmp,

"after_create",

DDL("create temporary view user_tmp_v as " "select * from user_tmp"),

)

event.listen(user_tmp, "before_drop", DDL("drop view user_tmp_v"))

@testing.provide_metadata

def test_reflect_string_column_max_len(self):

"""

SPANNER SPECIFIC TEST:

In Spanner column of the STRING type can be

created with size defined as MAX. The test

checks that such a column is correctly reflected.

"""

Table("text_table", self.metadata, Column("TestColumn", Text, nullable=False))

self.metadata.create_all()

Table("text_table", MetaData(bind=self.bind), autoload=True)

@testing.provide_metadata

def test_reflect_bytes_column_max_len(self):

"""

SPANNER SPECIFIC TEST:

In Spanner column of the BYTES type can be

created with size defined as MAX. The test

checks that such a column is correctly reflected.

"""

Table(

"bytes_table",

self.metadata,

Column("TestColumn", LargeBinary, nullable=False),

)

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):

"""

SPANNER OVERRIDE:

In Cloud Spanner unique indexes are used instead of directly

creating unique constraints. Overriding the test to replace

constraints with indexes in testing data.

"""

# SQLite dialect needs to parse the names of the constraints

# separately from what it gets from PRAGMA index_list(), and

# then matches them up. so same set of column_names in two

# constraints will confuse it. Perhaps we should no longer

# bother with index_list() here since we have the whole

# CREATE TABLE?

uniques = sorted(

[

{"name": "unique_a", "column_names": ["a"]},

{"name": "unique_a_b_c", "column_names": ["a", "b", "c"]},

{"name": "unique_c_a_b", "column_names": ["c", "a", "b"]},

{"name": "unique_asc_key", "column_names": ["asc", "key"]},

{"name": "i.have.dots", "column_names": ["b"]},

{"name": "i have spaces", "column_names": ["c"]},

],

key=operator.itemgetter("name"),

)

orig_meta = self.metadata

Table(

"testtbl",

orig_meta,

Column("id", sqlalchemy.Integer, primary_key=True),

Column("a", sqlalchemy.String(20)),

Column("b", sqlalchemy.String(30)),

Column("c", sqlalchemy.Integer),

# reserved identifiers

Column("asc", sqlalchemy.String(30)),

Column("key", sqlalchemy.String(30)),

sqlalchemy.Index("unique_a", "a", unique=True),

sqlalchemy.Index("unique_a_b_c", "a", "b", "c", unique=True),

sqlalchemy.Index("unique_c_a_b", "c", "a", "b", unique=True),

sqlalchemy.Index("unique_asc_key", "asc", "key", unique=True),

schema=schema,

)

orig_meta.create_all()

inspector = inspect(orig_meta.bind)

reflected = sorted(

inspector.get_unique_constraints("testtbl", schema=schema),

key=operator.itemgetter("name"),

)

names_that_duplicate_index = set()

for orig, refl in zip(uniques, reflected):

# Different dialects handle duplicate index and constraints

# differently, so ignore this flag

dupe = refl.pop("duplicates_index", None)

if dupe:

names_that_duplicate_index.add(dupe)

eq_(orig, refl)

reflected_metadata = MetaData()

reflected = Table(

"testtbl",

reflected_metadata,

autoload_with=orig_meta.bind,

schema=schema,

)

# test "deduplicates for index" logic. MySQL and Oracle

# "unique constraints" are actually unique indexes (with possible

# exception of a unique that is a dupe of another one in the case

# of Oracle). make sure # they aren't duplicated.

idx_names = set([idx.name for idx in reflected.indexes])

uq_names = set(

[

uq.name

for uq in reflected.constraints

if isinstance(uq, sqlalchemy.UniqueConstraint)

]

).difference(["unique_c_a_b"])

assert not idx_names.intersection(uq_names)

if names_that_duplicate_index:

eq_(names_that_duplicate_index, idx_names)

eq_(uq_names, set())

@testing.provide_metadata

def test_unique_constraint_raises(self):

"""

Checking that unique constraint creation

fails due to a ProgrammingError.

"""

Table(

"user_tmp_failure",

self.metadata,

Column("id", sqlalchemy.INT, primary_key=True),

Column("name", sqlalchemy.VARCHAR(50)),

sqlalchemy.UniqueConstraint("name", name="user_tmp_uq"),

)

with pytest.raises(spanner_dbapi.exceptions.ProgrammingError):

self.metadata.create_all()

@testing.provide_metadata

def _test_get_table_names(self, schema=None, table_type="table", order_by=None):

"""

SPANNER OVERRIDE:

Spanner doesn't support temporary tables, so real tables are

used for testing. As the original test expects only real

tables to be read, and in Spanner all the tables are real,

expected results override is required.

"""

_ignore_tables = [

"comment_test",

"noncol_idx_test_pk",

"noncol_idx_test_nopk",

"local_table",

"remote_table",

"remote_table_2",

]

meta = self.metadata

insp = inspect(meta.bind)

if table_type == "view" and not bool(os.environ.get("SPANNER_EMULATOR_HOST")):

table_names = insp.get_view_names(schema)

table_names.sort()

answer = ["email_addresses_v", "users_v"]

eq_(sorted(table_names), answer)

else:

if order_by:

tables = [

rec[0]

for rec in insp.get_sorted_table_and_fkc_names(schema)

if rec[0]

]

else:

tables = insp.get_table_names(schema)

table_names = [t for t in tables if t not in _ignore_tables]

if order_by == "foreign_key":

answer = {"dingalings", "email_addresses", "user_tmp", "users"}

eq_(set(table_names), answer)

else:

answer = ["dingalings", "email_addresses", "user_tmp", "users"]

eq_(sorted(table_names), answer)

@pytest.mark.skip("Spanner doesn't support temporary tables")

def test_get_temp_table_indexes(self):

pass

@pytest.mark.skip("Spanner doesn't support temporary tables")

def test_get_temp_table_unique_constraints(self):

pass

@testing.requires.table_reflection

def test_numeric_reflection(self):

"""

SPANNER OVERRIDE:

Spanner defines NUMERIC type with the constant precision=38

and scale=9. Overriding the test to check if the NUMERIC

column is successfully created and has dimensions

correct for Cloud Spanner.

"""

for typ in self._type_round_trip(Numeric(18, 5)):

assert isinstance(typ, Numeric)

eq_(typ.precision, 38)

eq_(typ.scale, 9)

@testing.requires.table_reflection

def test_binary_reflection(self):

"""

Check that a BYTES column with an explicitly

set size is correctly reflected.

"""

for typ in self._type_round_trip(LargeBinary(20)):

assert isinstance(typ, LargeBinary)

eq_(typ.length, 20)

@testing.requires.table_reflection

def test_array_reflection(self):

"""Check array columns reflection."""

orig_meta = self.metadata

str_array = ARRAY(String(16))

int_array = ARRAY(Integer)

arrays_test = Table(

"arrays_test",

orig_meta,

Column("id", Integer, primary_key=True),

Column("str_array", str_array),

Column("int_array", int_array),

)

arrays_test.create(create_engine(get_db_url()))

# autoload the table and check its columns reflection

tab = Table("arrays_test", orig_meta, autoload=True)

col_types = [col.type for col in tab.columns]

for type_ in (

str_array,

int_array,

):

assert type_ in col_types

tab.drop()

def _assert_insp_indexes(self, indexes, expected_indexes):

expected_indexes.sort(key=lambda item: item["name"])

index_names = [d["name"] for d in indexes]

exp_index_names = [d["name"] for d in expected_indexes]

assert sorted(index_names) == sorted(exp_index_names)

class CompositeKeyReflectionTest(_CompositeKeyReflectionTest):

@testing.requires.foreign_key_constraint_reflection

@testing.provide_metadata