summaryrefslogtreecommitdiff
path: root/test/dialect/mssql/test_reflection.py
diff options
context:
space:
mode:
authorMike Bayer <mike_mp@zzzcomputing.com>2019-01-06 01:14:26 -0500
committermike bayer <mike_mp@zzzcomputing.com>2019-01-06 17:34:50 +0000
commit1e1a38e7801f410f244e4bbb44ec795ae152e04e (patch)
tree28e725c5c8188bd0cfd133d1e268dbca9b524978 /test/dialect/mssql/test_reflection.py
parent404e69426b05a82d905cbb3ad33adafccddb00dd (diff)
downloadsqlalchemy-1e1a38e7801f410f244e4bbb44ec795ae152e04e.tar.gz
Run black -l 79 against all source files
This is a straight reformat run using black as is, with no edits applied at all. The black run will format code consistently, however in some cases that are prevalent in SQLAlchemy code it produces too-long lines. The too-long lines will be resolved in the following commit that will resolve all remaining flake8 issues including shadowed builtins, long lines, import order, unused imports, duplicate imports, and docstring issues. Change-Id: I7eda77fed3d8e73df84b3651fd6cfcfe858d4dc9
Diffstat (limited to 'test/dialect/mssql/test_reflection.py')
-rw-r--r--test/dialect/mssql/test_reflection.py303
1 files changed, 164 insertions, 139 deletions
diff --git a/test/dialect/mssql/test_reflection.py b/test/dialect/mssql/test_reflection.py
index e526168f1..47322ced6 100644
--- a/test/dialect/mssql/test_reflection.py
+++ b/test/dialect/mssql/test_reflection.py
@@ -3,9 +3,7 @@ from sqlalchemy.testing import eq_, is_, in_
from sqlalchemy import *
from sqlalchemy import types, schema, event
from sqlalchemy.databases import mssql
-from sqlalchemy.testing import (fixtures,
- AssertsCompiledSQL,
- ComparesTables)
+from sqlalchemy.testing import fixtures, AssertsCompiledSQL, ComparesTables
from sqlalchemy import testing
from sqlalchemy.engine.reflection import Inspector
from sqlalchemy import util
@@ -13,8 +11,9 @@ from sqlalchemy.dialects.mssql.information_schema import CoerceUnicode, tables
from sqlalchemy.dialects.mssql import base
from sqlalchemy.testing import mock
+
class ReflectionTest(fixtures.TestBase, ComparesTables, AssertsCompiledSQL):
- __only_on__ = 'mssql'
+ __only_on__ = "mssql"
__backend__ = True
@testing.provide_metadata
@@ -22,44 +21,49 @@ class ReflectionTest(fixtures.TestBase, ComparesTables, AssertsCompiledSQL):
meta = self.metadata
users = Table(
- 'engine_users',
+ "engine_users",
meta,
- Column('user_id', types.INT, primary_key=True),
- Column('user_name', types.VARCHAR(20), nullable=False),
- Column('test1', types.CHAR(5), nullable=False),
- Column('test2', types.Float(5), nullable=False),
- Column('test3', types.Text()),
- Column('test4', types.Numeric, nullable=False),
- Column('test5', types.DateTime),
- Column('parent_user_id', types.Integer,
- ForeignKey('engine_users.user_id')),
- Column('test6', types.DateTime, nullable=False),
- Column('test7', types.Text()),
- Column('test8', types.LargeBinary()),
- Column('test_passivedefault2', types.Integer,
- server_default='5'),
- Column('test9', types.BINARY(100)),
- Column('test_numeric', types.Numeric()),
+ Column("user_id", types.INT, primary_key=True),
+ Column("user_name", types.VARCHAR(20), nullable=False),
+ Column("test1", types.CHAR(5), nullable=False),
+ Column("test2", types.Float(5), nullable=False),
+ Column("test3", types.Text()),
+ Column("test4", types.Numeric, nullable=False),
+ Column("test5", types.DateTime),
+ Column(
+ "parent_user_id",
+ types.Integer,
+ ForeignKey("engine_users.user_id"),
+ ),
+ Column("test6", types.DateTime, nullable=False),
+ Column("test7", types.Text()),
+ Column("test8", types.LargeBinary()),
+ Column("test_passivedefault2", types.Integer, server_default="5"),
+ Column("test9", types.BINARY(100)),
+ Column("test_numeric", types.Numeric()),
)
addresses = Table(
- 'engine_email_addresses',
+ "engine_email_addresses",
meta,
- Column('address_id', types.Integer, primary_key=True),
- Column('remote_user_id', types.Integer,
- ForeignKey(users.c.user_id)),
- Column('email_address', types.String(20)),
- )
+ Column("address_id", types.Integer, primary_key=True),
+ Column(
+ "remote_user_id", types.Integer, ForeignKey(users.c.user_id)
+ ),
+ Column("email_address", types.String(20)),
+ )
meta.create_all()
meta2 = MetaData()
- reflected_users = Table('engine_users', meta2,
- autoload=True,
- autoload_with=testing.db)
- reflected_addresses = Table('engine_email_addresses',
- meta2,
- autoload=True,
- autoload_with=testing.db)
+ reflected_users = Table(
+ "engine_users", meta2, autoload=True, autoload_with=testing.db
+ )
+ reflected_addresses = Table(
+ "engine_email_addresses",
+ meta2,
+ autoload=True,
+ autoload_with=testing.db,
+ )
self.assert_tables_equal(users, reflected_users)
self.assert_tables_equal(addresses, reflected_addresses)
@@ -67,17 +71,14 @@ class ReflectionTest(fixtures.TestBase, ComparesTables, AssertsCompiledSQL):
def _test_specific_type(self, type_obj, ddl):
metadata = self.metadata
- table = Table(
- 'type_test', metadata,
- Column('col1', type_obj)
- )
+ table = Table("type_test", metadata, Column("col1", type_obj))
table.create()
m2 = MetaData()
- table2 = Table('type_test', m2, autoload_with=testing.db)
+ table2 = Table("type_test", m2, autoload_with=testing.db)
self.assert_compile(
schema.CreateTable(table2),
- "CREATE TABLE type_test (col1 %s NULL)" % ddl
+ "CREATE TABLE type_test (col1 %s NULL)" % ddl,
)
def test_xml_type(self):
@@ -93,31 +94,36 @@ class ReflectionTest(fixtures.TestBase, ComparesTables, AssertsCompiledSQL):
def test_identity(self):
metadata = self.metadata
table = Table(
- 'identity_test', metadata,
- Column('col1', Integer, mssql_identity_start=2,
- mssql_identity_increment=3, primary_key=True)
+ "identity_test",
+ metadata,
+ Column(
+ "col1",
+ Integer,
+ mssql_identity_start=2,
+ mssql_identity_increment=3,
+ primary_key=True,
+ ),
)
table.create()
meta2 = MetaData(testing.db)
- table2 = Table('identity_test', meta2, autoload=True)
- eq_(
- table2.c['col1'].dialect_options['mssql'][
- 'identity_start'], 2)
- eq_(
- table2.c['col1'].dialect_options['mssql'][
- 'identity_increment'], 3)
+ table2 = Table("identity_test", meta2, autoload=True)
+ eq_(table2.c["col1"].dialect_options["mssql"]["identity_start"], 2)
+ eq_(table2.c["col1"].dialect_options["mssql"]["identity_increment"], 3)
@testing.emits_warning("Did not recognize")
@testing.provide_metadata
def test_skip_types(self):
metadata = self.metadata
- testing.db.execute("""
+ testing.db.execute(
+ """
create table foo (id integer primary key, data xml)
- """)
+ """
+ )
with mock.patch.object(
- testing.db.dialect, "ischema_names", {"int": mssql.INTEGER}):
- t1 = Table('foo', metadata, autoload=True)
+ testing.db.dialect, "ischema_names", {"int": mssql.INTEGER}
+ ):
+ t1 = Table("foo", metadata, autoload=True)
assert isinstance(t1.c.id.type, Integer)
assert isinstance(t1.c.data.type, types.NullType)
@@ -127,29 +133,33 @@ class ReflectionTest(fixtures.TestBase, ComparesTables, AssertsCompiledSQL):
metadata = self.metadata
Table(
- "subject", metadata,
+ "subject",
+ metadata,
Column("id", Integer),
PrimaryKeyConstraint("id", name="subj_pk"),
schema=testing.config.test_schema,
)
Table(
- "referrer", metadata,
+ "referrer",
+ metadata,
Column("id", Integer, primary_key=True),
Column(
- 'sid',
+ "sid",
ForeignKey(
"%s.subject.id" % testing.config.test_schema,
- name='fk_subject')
+ name="fk_subject",
+ ),
),
- schema=testing.config.test_schema
+ schema=testing.config.test_schema,
)
Table(
- "subject", metadata,
+ "subject",
+ metadata,
Column("id", Integer),
PrimaryKeyConstraint("id", name="subj_pk"),
- schema=testing.config.test_schema_2
+ schema=testing.config.test_schema_2,
)
metadata.create_all()
@@ -157,102 +167,111 @@ class ReflectionTest(fixtures.TestBase, ComparesTables, AssertsCompiledSQL):
insp = inspect(testing.db)
eq_(
insp.get_foreign_keys("referrer", testing.config.test_schema),
- [{
- 'name': 'fk_subject',
- 'constrained_columns': ['sid'],
- 'referred_schema': 'test_schema',
- 'referred_table': 'subject',
- 'referred_columns': ['id']}]
+ [
+ {
+ "name": "fk_subject",
+ "constrained_columns": ["sid"],
+ "referred_schema": "test_schema",
+ "referred_table": "subject",
+ "referred_columns": ["id"],
+ }
+ ],
)
@testing.provide_metadata
def test_db_qualified_items(self):
metadata = self.metadata
- Table('foo', metadata, Column('id', Integer, primary_key=True))
- Table('bar',
- metadata,
- Column('id', Integer, primary_key=True),
- Column('foo_id', Integer, ForeignKey('foo.id', name="fkfoo")))
+ Table("foo", metadata, Column("id", Integer, primary_key=True))
+ Table(
+ "bar",
+ metadata,
+ Column("id", Integer, primary_key=True),
+ Column("foo_id", Integer, ForeignKey("foo.id", name="fkfoo")),
+ )
metadata.create_all()
dbname = testing.db.scalar("select db_name()")
owner = testing.db.scalar("SELECT user_name()")
- referred_schema = '%(dbname)s.%(owner)s' % {
- "dbname": dbname, "owner": owner}
+ referred_schema = "%(dbname)s.%(owner)s" % {
+ "dbname": dbname,
+ "owner": owner,
+ }
inspector = inspect(testing.db)
- bar_via_db = inspector.get_foreign_keys(
- "bar", schema=referred_schema)
+ bar_via_db = inspector.get_foreign_keys("bar", schema=referred_schema)
eq_(
bar_via_db,
- [{
- 'referred_table': 'foo',
- 'referred_columns': ['id'],
- 'referred_schema': referred_schema,
- 'name': 'fkfoo',
- 'constrained_columns': ['foo_id']}]
+ [
+ {
+ "referred_table": "foo",
+ "referred_columns": ["id"],
+ "referred_schema": referred_schema,
+ "name": "fkfoo",
+ "constrained_columns": ["foo_id"],
+ }
+ ],
)
assert testing.db.has_table("bar", schema=referred_schema)
m2 = MetaData()
- Table('bar', m2, schema=referred_schema, autoload=True,
- autoload_with=testing.db)
+ Table(
+ "bar",
+ m2,
+ schema=referred_schema,
+ autoload=True,
+ autoload_with=testing.db,
+ )
eq_(m2.tables["%s.foo" % referred_schema].schema, referred_schema)
@testing.provide_metadata
def test_indexes_cols(self):
metadata = self.metadata
- t1 = Table('t', metadata, Column('x', Integer), Column('y', Integer))
- Index('foo', t1.c.x, t1.c.y)
+ t1 = Table("t", metadata, Column("x", Integer), Column("y", Integer))
+ Index("foo", t1.c.x, t1.c.y)
metadata.create_all()
m2 = MetaData()
- t2 = Table('t', m2, autoload=True, autoload_with=testing.db)
+ t2 = Table("t", m2, autoload=True, autoload_with=testing.db)
- eq_(
- set(list(t2.indexes)[0].columns),
- set([t2.c['x'], t2.c.y])
- )
+ eq_(set(list(t2.indexes)[0].columns), set([t2.c["x"], t2.c.y]))
@testing.provide_metadata
def test_indexes_cols_with_commas(self):
metadata = self.metadata
- t1 = Table('t',
- metadata,
- Column('x, col', Integer, key='x'),
- Column('y', Integer))
- Index('foo', t1.c.x, t1.c.y)
+ t1 = Table(
+ "t",
+ metadata,
+ Column("x, col", Integer, key="x"),
+ Column("y", Integer),
+ )
+ Index("foo", t1.c.x, t1.c.y)
metadata.create_all()
m2 = MetaData()
- t2 = Table('t', m2, autoload=True, autoload_with=testing.db)
+ t2 = Table("t", m2, autoload=True, autoload_with=testing.db)
- eq_(
- set(list(t2.indexes)[0].columns),
- set([t2.c['x, col'], t2.c.y])
- )
+ eq_(set(list(t2.indexes)[0].columns), set([t2.c["x, col"], t2.c.y]))
@testing.provide_metadata
def test_indexes_cols_with_spaces(self):
metadata = self.metadata
- t1 = Table('t',
- metadata,
- Column('x col', Integer, key='x'),
- Column('y', Integer))
- Index('foo', t1.c.x, t1.c.y)
+ t1 = Table(
+ "t",
+ metadata,
+ Column("x col", Integer, key="x"),
+ Column("y", Integer),
+ )
+ Index("foo", t1.c.x, t1.c.y)
metadata.create_all()
m2 = MetaData()
- t2 = Table('t', m2, autoload=True, autoload_with=testing.db)
+ t2 = Table("t", m2, autoload=True, autoload_with=testing.db)
- eq_(
- set(list(t2.indexes)[0].columns),
- set([t2.c['x col'], t2.c.y])
- )
+ eq_(set(list(t2.indexes)[0].columns), set([t2.c["x col"], t2.c.y]))
@testing.provide_metadata
def test_max_ident_in_varchar_not_present(self):
@@ -267,75 +286,81 @@ class ReflectionTest(fixtures.TestBase, ComparesTables, AssertsCompiledSQL):
metadata = self.metadata
Table(
- 't', metadata,
- Column('t1', types.String),
- Column('t2', types.Text('max')),
- Column('t3', types.Text('max')),
- Column('t4', types.LargeBinary('max')),
- Column('t5', types.VARBINARY('max')),
+ "t",
+ metadata,
+ Column("t1", types.String),
+ Column("t2", types.Text("max")),
+ Column("t3", types.Text("max")),
+ Column("t4", types.LargeBinary("max")),
+ Column("t5", types.VARBINARY("max")),
)
metadata.create_all()
- for col in inspect(testing.db).get_columns('t'):
- is_(col['type'].length, None)
- in_('max', str(col['type'].compile(dialect=testing.db.dialect)))
+ for col in inspect(testing.db).get_columns("t"):
+ is_(col["type"].length, None)
+ in_("max", str(col["type"].compile(dialect=testing.db.dialect)))
class InfoCoerceUnicodeTest(fixtures.TestBase, AssertsCompiledSQL):
def test_info_unicode_coercion(self):
dialect = mssql.dialect()
- value = CoerceUnicode().bind_processor(dialect)('a string')
+ value = CoerceUnicode().bind_processor(dialect)("a string")
assert isinstance(value, util.text_type)
def test_info_unicode_cast_no_2000(self):
dialect = mssql.dialect()
dialect.server_version_info = base.MS_2000_VERSION
- stmt = tables.c.table_name == 'somename'
+ stmt = tables.c.table_name == "somename"
self.assert_compile(
stmt,
"[INFORMATION_SCHEMA].[TABLES].[TABLE_NAME] = :table_name_1",
- dialect=dialect
+ dialect=dialect,
)
def test_info_unicode_cast(self):
dialect = mssql.dialect()
dialect.server_version_info = base.MS_2005_VERSION
- stmt = tables.c.table_name == 'somename'
+ stmt = tables.c.table_name == "somename"
self.assert_compile(
stmt,
"[INFORMATION_SCHEMA].[TABLES].[TABLE_NAME] = "
"CAST(:table_name_1 AS NVARCHAR(max))",
- dialect=dialect
+ dialect=dialect,
)
class ReflectHugeViewTest(fixtures.TestBase):
- __only_on__ = 'mssql'
+ __only_on__ = "mssql"
__backend__ = True
# crashes on freetds 0.91, not worth it
- __skip_if__ = (
- lambda: testing.requires.mssql_freetds.enabled,
- )
+ __skip_if__ = (lambda: testing.requires.mssql_freetds.enabled,)
def setup(self):
self.col_num = 150
self.metadata = MetaData(testing.db)
- t = Table('base_table', self.metadata,
- *[Column("long_named_column_number_%d" % i, Integer)
+ t = Table(
+ "base_table",
+ self.metadata,
+ *[
+ Column("long_named_column_number_%d" % i, Integer)
+ for i in range(self.col_num)
+ ]
+ )
+ self.view_str = view_str = (
+ "CREATE VIEW huge_named_view AS SELECT %s FROM base_table"
+ % (
+ ",".join(
+ "long_named_column_number_%d" % i
for i in range(self.col_num)
- ]
- )
- self.view_str = view_str = \
- "CREATE VIEW huge_named_view AS SELECT %s FROM base_table" % (
- ",".join("long_named_column_number_%d" % i
- for i in range(self.col_num))
+ )
)
+ )
assert len(view_str) > 4000
- event.listen(t, 'after_create', DDL(view_str))
- event.listen(t, 'before_drop', DDL("DROP VIEW huge_named_view"))
+ event.listen(t, "after_create", DDL(view_str))
+ event.listen(t, "before_drop", DDL("DROP VIEW huge_named_view"))
self.metadata.create_all()