summaryrefslogtreecommitdiff
path: root/test/dialect
diff options
context:
space:
mode:
Diffstat (limited to 'test/dialect')
-rw-r--r--test/dialect/mssql/test_reflection.py14
-rw-r--r--test/dialect/mssql/test_types.py43
-rw-r--r--test/dialect/mysql/test_reflection.py290
-rw-r--r--test/dialect/oracle/test_reflection.py100
-rw-r--r--test/dialect/oracle/test_types.py100
-rw-r--r--test/dialect/postgresql/test_dialect.py4
-rw-r--r--test/dialect/postgresql/test_query.py12
-rw-r--r--test/dialect/postgresql/test_reflection.py169
-rw-r--r--test/dialect/postgresql/test_types.py34
-rw-r--r--test/dialect/test_sqlite.py158
10 files changed, 479 insertions, 445 deletions
diff --git a/test/dialect/mssql/test_reflection.py b/test/dialect/mssql/test_reflection.py
index d791e9ec1..6009bfb6c 100644
--- a/test/dialect/mssql/test_reflection.py
+++ b/test/dialect/mssql/test_reflection.py
@@ -81,12 +81,11 @@ class ReflectionTest(fixtures.TestBase, ComparesTables, AssertsCompiledSQL):
meta2 = MetaData()
reflected_users = Table(
- "engine_users", meta2, autoload=True, autoload_with=testing.db
+ "engine_users", meta2, autoload_with=testing.db
)
reflected_addresses = Table(
"engine_email_addresses",
meta2,
- autoload=True,
autoload_with=testing.db,
)
self.assert_tables_equal(users, reflected_users)
@@ -147,8 +146,8 @@ class ReflectionTest(fixtures.TestBase, ComparesTables, AssertsCompiledSQL):
):
table.create()
- meta2 = MetaData(testing.db)
- table2 = Table("identity_test", meta2, autoload=True)
+ meta2 = MetaData()
+ table2 = Table("identity_test", meta2, autoload_with=testing.db)
eq_(table2.c["col1"].dialect_options["mssql"]["identity_start"], None)
eq_(
table2.c["col1"].dialect_options["mssql"]["identity_increment"],
@@ -372,7 +371,6 @@ class ReflectionTest(fixtures.TestBase, ComparesTables, AssertsCompiledSQL):
"bar",
m2,
schema=referred_schema,
- autoload=True,
autoload_with=testing.db,
)
eq_(m2.tables["%s.foo" % referred_schema].schema, referred_schema)
@@ -386,7 +384,7 @@ class ReflectionTest(fixtures.TestBase, ComparesTables, AssertsCompiledSQL):
metadata.create_all()
m2 = MetaData()
- t2 = Table("t", m2, autoload=True, autoload_with=testing.db)
+ t2 = Table("t", m2, autoload_with=testing.db)
eq_(set(list(t2.indexes)[0].columns), set([t2.c["x"], t2.c.y]))
@@ -404,7 +402,7 @@ class ReflectionTest(fixtures.TestBase, ComparesTables, AssertsCompiledSQL):
metadata.create_all()
m2 = MetaData()
- t2 = Table("t", m2, autoload=True, autoload_with=testing.db)
+ t2 = Table("t", m2, autoload_with=testing.db)
eq_(set(list(t2.indexes)[0].columns), set([t2.c["x, col"], t2.c.y]))
@@ -422,7 +420,7 @@ class ReflectionTest(fixtures.TestBase, ComparesTables, AssertsCompiledSQL):
metadata.create_all()
m2 = MetaData()
- t2 = Table("t", m2, autoload=True, autoload_with=testing.db)
+ t2 = Table("t", m2, autoload_with=testing.db)
eq_(set(list(t2.indexes)[0].columns), set([t2.c["x col"], t2.c.y]))
diff --git a/test/dialect/mssql/test_types.py b/test/dialect/mssql/test_types.py
index cf5b66d57..11a2a25b3 100644
--- a/test/dialect/mssql/test_types.py
+++ b/test/dialect/mssql/test_types.py
@@ -537,9 +537,6 @@ class TypeDDLTest(fixtures.TestBase):
self.assert_(repr(col))
-metadata = None
-
-
class TypeRoundTripTest(
fixtures.TestBase, AssertsExecutionResults, ComparesTables
):
@@ -547,15 +544,9 @@ class TypeRoundTripTest(
__backend__ = True
- @classmethod
- def setup_class(cls):
- global metadata
- metadata = MetaData(testing.db)
-
- def teardown(self):
- metadata.drop_all()
-
+ @testing.provide_metadata
def test_decimal_notation(self):
+ metadata = self.metadata
numeric_table = Table(
"numeric_table",
metadata,
@@ -645,7 +636,10 @@ class TypeRoundTripTest(
)
eq_(value, returned)
+ @testing.provide_metadata
def test_float(self):
+ metadata = self.metadata
+
float_table = Table(
"float_table",
metadata,
@@ -696,6 +690,7 @@ class TypeRoundTripTest(
# todo this should suppress warnings, but it does not
@emits_warning_on("mssql+mxodbc", r".*does not have any indexes.*")
+ @testing.provide_metadata
def test_dates(self):
"Exercise type specification for date types."
@@ -727,6 +722,8 @@ class TypeRoundTripTest(
(mssql.MSDateTime2, [1], {}, "DATETIME2(1)", [">=", (10,)]),
]
+ metadata = self.metadata
+
table_args = ["test_mssql_dates", metadata]
for index, spec in enumerate(columns):
type_, args, kw, res, requires = spec[0:5]
@@ -751,7 +748,7 @@ class TypeRoundTripTest(
self.assert_(repr(col))
dates_table.create(checkfirst=True)
reflected_dates = Table(
- "test_mssql_dates", MetaData(testing.db), autoload=True
+ "test_mssql_dates", MetaData(), autoload_with=testing.db
)
for col in reflected_dates.c:
self.assert_types_base(col, dates_table.c[col.key])
@@ -915,7 +912,12 @@ class TypeRoundTripTest(
@emits_warning_on("mssql+mxodbc", r".*does not have any indexes.*")
@testing.provide_metadata
- def _test_binary_reflection(self, deprecate_large_types):
+ @testing.combinations(
+ ("legacy_large_types", False),
+ ("sql2012_large_types", True, lambda: testing.only_on("mssql >= 11")),
+ id_="ia",
+ )
+ def test_binary_reflection(self, deprecate_large_types):
"Exercise type specification for binary types."
columns = [
@@ -951,7 +953,7 @@ class TypeRoundTripTest(
binary_table = Table(*table_args)
metadata.create_all()
reflected_binary = Table(
- "test_mssql_binary", MetaData(testing.db), autoload=True
+ "test_mssql_binary", MetaData(), autoload_with=testing.db
)
for col, spec in zip(reflected_binary.c, columns):
eq_(
@@ -976,14 +978,9 @@ class TypeRoundTripTest(
col.type.length, binary_table.c[col.name].type.length
)
- def test_binary_reflection_legacy_large_types(self):
- self._test_binary_reflection(False)
-
- @testing.only_on("mssql >= 11")
- def test_binary_reflection_sql2012_large_types(self):
- self._test_binary_reflection(True)
-
+ @testing.provide_metadata
def test_autoincrement(self):
+ metadata = self.metadata
Table(
"ai_1",
metadata,
@@ -1046,10 +1043,10 @@ class TypeRoundTripTest(
"ai_7",
"ai_8",
]
- mr = MetaData(testing.db)
+ mr = MetaData()
for name in table_names:
- tbl = Table(name, mr, autoload=True)
+ tbl = Table(name, mr, autoload_with=testing.db)
tbl = metadata.tables[name]
# test that the flag itself reflects appropriately
diff --git a/test/dialect/mysql/test_reflection.py b/test/dialect/mysql/test_reflection.py
index 026025a88..d5684b22f 100644
--- a/test/dialect/mysql/test_reflection.py
+++ b/test/dialect/mysql/test_reflection.py
@@ -69,10 +69,10 @@ class TypeReflectionTest(fixtures.TestBase):
)
m.create_all()
- m2 = MetaData(testing.db)
- tables = [Table("mysql_types", m2, autoload=True)]
+ m2 = MetaData()
+ tables = [Table("mysql_types", m2, autoload_with=testing.db)]
if use_views:
- tables.append(Table("mysql_types_v", m2, autoload=True))
+ tables.append(Table("mysql_types_v", m2, autoload_with=testing.db))
for table in tables:
for i, (reflected_col, spec) in enumerate(zip(table.c, specs)):
@@ -238,7 +238,7 @@ class ReflectionTest(fixtures.TestBase, AssertsCompiledSQL):
def_table = Table(
"mysql_def",
- MetaData(testing.db),
+ MetaData(),
Column(
"c1",
VARCHAR(10, collation="utf8_unicode_ci"),
@@ -259,11 +259,14 @@ class ReflectionTest(fixtures.TestBase, AssertsCompiledSQL):
),
),
)
- def_table.create()
+
+ def_table.create(testing.db)
try:
- reflected = Table("mysql_def", MetaData(testing.db), autoload=True)
+ reflected = Table(
+ "mysql_def", MetaData(), autoload_with=testing.db
+ )
finally:
- def_table.drop()
+ def_table.drop(testing.db)
assert def_table.c.c1.server_default.arg == ""
assert def_table.c.c2.server_default.arg == "0"
assert def_table.c.c3.server_default.arg == "abc"
@@ -281,13 +284,13 @@ class ReflectionTest(fixtures.TestBase, AssertsCompiledSQL):
r"CURRENT_TIMESTAMP(\(\))? ON UPDATE CURRENT_TIMESTAMP(\(\))?",
str(reflected.c.c6.server_default.arg).upper(),
)
- reflected.create()
+ reflected.create(testing.db)
try:
reflected2 = Table(
- "mysql_def", MetaData(testing.db), autoload=True
+ "mysql_def", MetaData(), autoload_with=testing.db
)
finally:
- reflected.drop()
+ reflected.drop(testing.db)
assert str(reflected2.c.c1.server_default.arg) == "''"
assert str(reflected2.c.c2.server_default.arg) == "'0'"
assert str(reflected2.c.c3.server_default.arg) == "'abc'"
@@ -325,17 +328,18 @@ class ReflectionTest(fixtures.TestBase, AssertsCompiledSQL):
def_table = Table(
"mysql_def",
- MetaData(testing.db),
+ MetaData(),
Column("c1", Integer()),
comment=comment,
**kwargs
)
- def_table.create()
- try:
- reflected = Table("mysql_def", MetaData(testing.db), autoload=True)
- finally:
- def_table.drop()
+ with testing.db.connect() as conn:
+ def_table.create(conn)
+ try:
+ reflected = Table("mysql_def", MetaData(), autoload_with=conn)
+ finally:
+ def_table.drop(conn)
if testing.against("mariadb"):
assert def_table.kwargs["mariadb_engine"] == "MEMORY"
@@ -382,153 +386,153 @@ class ReflectionTest(fixtures.TestBase, AssertsCompiledSQL):
# This is explicitly ignored when reflecting schema.
# assert reflected.kwargs['mysql_auto_increment'] == '5'
+ @testing.provide_metadata
def test_reflection_on_include_columns(self):
"""Test reflection of include_columns to be sure they respect case."""
+ meta = self.metadata
case_table = Table(
"mysql_case",
- MetaData(testing.db),
+ meta,
Column("c1", String(10)),
Column("C2", String(10)),
Column("C3", String(10)),
)
- try:
- case_table.create()
- reflected = Table(
- "mysql_case",
- MetaData(testing.db),
- autoload=True,
- include_columns=["c1", "C2"],
- )
- for t in case_table, reflected:
- assert "c1" in t.c.keys()
- assert "C2" in t.c.keys()
- reflected2 = Table(
- "mysql_case",
- MetaData(testing.db),
- autoload=True,
- include_columns=["c1", "c2"],
- )
- assert "c1" in reflected2.c.keys()
- for c in ["c2", "C2", "C3"]:
- assert c not in reflected2.c.keys()
- finally:
- case_table.drop()
+ case_table.create(testing.db)
+ reflected = Table(
+ "mysql_case",
+ MetaData(),
+ autoload_with=testing.db,
+ include_columns=["c1", "C2"],
+ )
+ for t in case_table, reflected:
+ assert "c1" in t.c.keys()
+ assert "C2" in t.c.keys()
+ reflected2 = Table(
+ "mysql_case",
+ MetaData(),
+ autoload_with=testing.db,
+ include_columns=["c1", "c2"],
+ )
+ assert "c1" in reflected2.c.keys()
+ for c in ["c2", "C2", "C3"]:
+ assert c not in reflected2.c.keys()
+ @testing.provide_metadata
def test_autoincrement(self):
- meta = MetaData(testing.db)
- try:
- Table(
- "ai_1",
- meta,
- Column("int_y", Integer, primary_key=True, autoincrement=True),
- Column("int_n", Integer, DefaultClause("0"), primary_key=True),
- mysql_engine="MyISAM",
- )
- Table(
- "ai_2",
- meta,
- Column("int_y", Integer, primary_key=True, autoincrement=True),
- Column("int_n", Integer, DefaultClause("0"), primary_key=True),
- mysql_engine="MyISAM",
- )
- Table(
- "ai_3",
- meta,
- Column(
- "int_n",
- Integer,
- DefaultClause("0"),
- primary_key=True,
- autoincrement=False,
- ),
- Column("int_y", Integer, primary_key=True, autoincrement=True),
- mysql_engine="MyISAM",
- )
- Table(
- "ai_4",
- meta,
- Column(
- "int_n",
- Integer,
- DefaultClause("0"),
- primary_key=True,
- autoincrement=False,
- ),
- Column(
- "int_n2",
- Integer,
- DefaultClause("0"),
- primary_key=True,
- autoincrement=False,
- ),
- mysql_engine="MyISAM",
- )
- Table(
- "ai_5",
- meta,
- Column("int_y", Integer, primary_key=True, autoincrement=True),
- Column(
- "int_n",
- Integer,
- DefaultClause("0"),
- primary_key=True,
- autoincrement=False,
- ),
- mysql_engine="MyISAM",
- )
- Table(
- "ai_6",
- meta,
- Column("o1", String(1), DefaultClause("x"), primary_key=True),
- Column("int_y", Integer, primary_key=True, autoincrement=True),
- mysql_engine="MyISAM",
- )
- Table(
- "ai_7",
- meta,
- Column("o1", String(1), DefaultClause("x"), primary_key=True),
- Column("o2", String(1), DefaultClause("x"), primary_key=True),
- Column("int_y", Integer, primary_key=True, autoincrement=True),
- mysql_engine="MyISAM",
- )
- Table(
- "ai_8",
- meta,
- Column("o1", String(1), DefaultClause("x"), primary_key=True),
- Column("o2", String(1), DefaultClause("x"), primary_key=True),
- mysql_engine="MyISAM",
- )
- meta.create_all()
-
- table_names = [
- "ai_1",
- "ai_2",
- "ai_3",
- "ai_4",
- "ai_5",
- "ai_6",
- "ai_7",
- "ai_8",
- ]
- mr = MetaData(testing.db)
- mr.reflect(only=table_names)
+ meta = self.metadata
+ Table(
+ "ai_1",
+ meta,
+ Column("int_y", Integer, primary_key=True, autoincrement=True),
+ Column("int_n", Integer, DefaultClause("0"), primary_key=True),
+ mysql_engine="MyISAM",
+ )
+ Table(
+ "ai_2",
+ meta,
+ Column("int_y", Integer, primary_key=True, autoincrement=True),
+ Column("int_n", Integer, DefaultClause("0"), primary_key=True),
+ mysql_engine="MyISAM",
+ )
+ Table(
+ "ai_3",
+ meta,
+ Column(
+ "int_n",
+ Integer,
+ DefaultClause("0"),
+ primary_key=True,
+ autoincrement=False,
+ ),
+ Column("int_y", Integer, primary_key=True, autoincrement=True),
+ mysql_engine="MyISAM",
+ )
+ Table(
+ "ai_4",
+ meta,
+ Column(
+ "int_n",
+ Integer,
+ DefaultClause("0"),
+ primary_key=True,
+ autoincrement=False,
+ ),
+ Column(
+ "int_n2",
+ Integer,
+ DefaultClause("0"),
+ primary_key=True,
+ autoincrement=False,
+ ),
+ mysql_engine="MyISAM",
+ )
+ Table(
+ "ai_5",
+ meta,
+ Column("int_y", Integer, primary_key=True, autoincrement=True),
+ Column(
+ "int_n",
+ Integer,
+ DefaultClause("0"),
+ primary_key=True,
+ autoincrement=False,
+ ),
+ mysql_engine="MyISAM",
+ )
+ Table(
+ "ai_6",
+ meta,
+ Column("o1", String(1), DefaultClause("x"), primary_key=True),
+ Column("int_y", Integer, primary_key=True, autoincrement=True),
+ mysql_engine="MyISAM",
+ )
+ Table(
+ "ai_7",
+ meta,
+ Column("o1", String(1), DefaultClause("x"), primary_key=True),
+ Column("o2", String(1), DefaultClause("x"), primary_key=True),
+ Column("int_y", Integer, primary_key=True, autoincrement=True),
+ mysql_engine="MyISAM",
+ )
+ Table(
+ "ai_8",
+ meta,
+ Column("o1", String(1), DefaultClause("x"), primary_key=True),
+ Column("o2", String(1), DefaultClause("x"), primary_key=True),
+ mysql_engine="MyISAM",
+ )
+ meta.create_all(testing.db)
+
+ table_names = [
+ "ai_1",
+ "ai_2",
+ "ai_3",
+ "ai_4",
+ "ai_5",
+ "ai_6",
+ "ai_7",
+ "ai_8",
+ ]
+ mr = MetaData()
+ mr.reflect(testing.db, only=table_names)
+ with testing.db.begin() as conn:
for tbl in [mr.tables[name] for name in table_names]:
for c in tbl.c:
if c.name.startswith("int_y"):
assert c.autoincrement
elif c.name.startswith("int_n"):
assert not c.autoincrement
- tbl.insert().execute()
+ conn.execute(tbl.insert())
if "int_y" in tbl.c:
- assert select(tbl.c.int_y).scalar() == 1
- assert list(tbl.select().execute().first()).count(1) == 1
+ assert conn.scalar(select(tbl.c.int_y)) == 1
+ assert (
+ list(conn.execute(tbl.select()).first()).count(1) == 1
+ )
else:
- assert 1 not in list(tbl.select().execute().first())
- finally:
- meta.drop_all()
+ assert 1 not in list(conn.execute(tbl.select()).first())
@testing.provide_metadata
def test_view_reflection(self):
@@ -734,7 +738,7 @@ class ReflectionTest(fixtures.TestBase, AssertsCompiledSQL):
# reflection here favors the unique index, as that's the
# more "official" MySQL construct
- reflected = Table("mysql_uc", MetaData(testing.db), autoload=True)
+ reflected = Table("mysql_uc", MetaData(), autoload_with=testing.db)
indexes = dict((i.name, i) for i in reflected.indexes)
constraints = set(uc.name for uc in reflected.constraints)
diff --git a/test/dialect/oracle/test_reflection.py b/test/dialect/oracle/test_reflection.py
index bd881ac1c..efa21fc1a 100644
--- a/test/dialect/oracle/test_reflection.py
+++ b/test/dialect/oracle/test_reflection.py
@@ -123,17 +123,18 @@ drop synonym %(test_schema)s.local_table;
Column("pid", Integer, ForeignKey("%s.parent.pid" % schema)),
schema=schema,
)
- meta.create_all()
- parent.insert().execute({"pid": 1})
- child.insert().execute({"cid": 1, "pid": 1})
- eq_(child.select().execute().fetchall(), [(1, 1)])
+ with testing.db.begin() as conn:
+ meta.create_all(conn)
+ conn.execute(parent.insert(), {"pid": 1})
+ conn.execute(child.insert(), {"cid": 1, "pid": 1})
+ eq_(conn.execute(child.select()).fetchall(), [(1, 1)])
def test_reflect_alt_table_owner_local_synonym(self):
- meta = MetaData(testing.db)
+ meta = MetaData()
parent = Table(
"%s_pt" % testing.config.test_schema,
meta,
- autoload=True,
+ autoload_with=testing.db,
oracle_resolve_synonyms=True,
)
self.assert_compile(
@@ -142,14 +143,13 @@ drop synonym %(test_schema)s.local_table;
"%(test_schema)s_pt.data FROM %(test_schema)s_pt"
% {"test_schema": testing.config.test_schema},
)
- select(parent).execute().fetchall()
def test_reflect_alt_synonym_owner_local_table(self):
- meta = MetaData(testing.db)
+ meta = MetaData()
parent = Table(
"local_table",
meta,
- autoload=True,
+ autoload_with=testing.db,
oracle_resolve_synonyms=True,
schema=testing.config.test_schema,
)
@@ -160,7 +160,6 @@ drop synonym %(test_schema)s.local_table;
"FROM %(test_schema)s.local_table"
% {"test_schema": testing.config.test_schema},
)
- select(parent).execute().fetchall()
@testing.provide_metadata
def test_create_same_names_implicit_schema(self):
@@ -180,12 +179,18 @@ drop synonym %(test_schema)s.local_table;
eq_(child.select().execute().fetchall(), [(1, 1)])
def test_reflect_alt_owner_explicit(self):
- meta = MetaData(testing.db)
+ meta = MetaData()
parent = Table(
- "parent", meta, autoload=True, schema=testing.config.test_schema
+ "parent",
+ meta,
+ autoload_with=testing.db,
+ schema=testing.config.test_schema,
)
child = Table(
- "child", meta, autoload=True, schema=testing.config.test_schema
+ "child",
+ meta,
+ autoload_with=testing.db,
+ schema=testing.config.test_schema,
)
self.assert_compile(
@@ -194,9 +199,10 @@ drop synonym %(test_schema)s.local_table;
"%(test_schema)s.parent.id = %(test_schema)s.child.parent_id"
% {"test_schema": testing.config.test_schema},
)
- select(parent, child).select_from(
- parent.join(child)
- ).execute().fetchall()
+ with testing.db.connect() as conn:
+ conn.execute(
+ select(parent, child).select_from(parent.join(child))
+ ).fetchall()
# check table comment (#5146)
eq_(parent.comment, "my table comment")
@@ -241,8 +247,8 @@ drop synonym %(test_schema)s.local_table;
% {"test_schema": testing.config.test_schema},
)
try:
- meta = MetaData(testing.db)
- lcl = Table("localtable", meta, autoload=True)
+ meta = MetaData()
+ lcl = Table("localtable", meta, autoload_with=testing.db)
parent = meta.tables["%s.parent" % testing.config.test_schema]
self.assert_compile(
parent.join(lcl),
@@ -251,19 +257,22 @@ drop synonym %(test_schema)s.local_table;
"localtable.parent_id"
% {"test_schema": testing.config.test_schema},
)
- select(parent, lcl).select_from(
- parent.join(lcl)
- ).execute().fetchall()
finally:
exec_sql(testing.db, "DROP TABLE localtable")
def test_reflect_alt_owner_implicit(self):
- meta = MetaData(testing.db)
+ meta = MetaData()
parent = Table(
- "parent", meta, autoload=True, schema=testing.config.test_schema
+ "parent",
+ meta,
+ autoload_with=testing.db,
+ schema=testing.config.test_schema,
)
child = Table(
- "child", meta, autoload=True, schema=testing.config.test_schema
+ "child",
+ meta,
+ autoload_with=testing.db,
+ schema=testing.config.test_schema,
)
self.assert_compile(
parent.join(child),
@@ -272,9 +281,10 @@ drop synonym %(test_schema)s.local_table;
"%(test_schema)s.child.parent_id"
% {"test_schema": testing.config.test_schema},
)
- select(parent, child).select_from(
- parent.join(child)
- ).execute().fetchall()
+ with testing.db.connect() as conn:
+ conn.execute(
+ select(parent, child).select_from(parent.join(child))
+ ).fetchall()
def test_reflect_alt_owner_synonyms(self):
exec_sql(
@@ -284,9 +294,12 @@ drop synonym %(test_schema)s.local_table;
"%s.ptable(id))" % testing.config.test_schema,
)
try:
- meta = MetaData(testing.db)
+ meta = MetaData()
lcl = Table(
- "localtable", meta, autoload=True, oracle_resolve_synonyms=True
+ "localtable",
+ meta,
+ autoload_with=testing.db,
+ oracle_resolve_synonyms=True,
)
parent = meta.tables["%s.ptable" % testing.config.test_schema]
self.assert_compile(
@@ -296,25 +309,26 @@ drop synonym %(test_schema)s.local_table;
"localtable.parent_id"
% {"test_schema": testing.config.test_schema},
)
- select(parent, lcl).select_from(
- parent.join(lcl)
- ).execute().fetchall()
+ with testing.db.connect() as conn:
+ conn.execute(
+ select(parent, lcl).select_from(parent.join(lcl))
+ ).fetchall()
finally:
exec_sql(testing.db, "DROP TABLE localtable")
def test_reflect_remote_synonyms(self):
- meta = MetaData(testing.db)
+ meta = MetaData()
parent = Table(
"ptable",
meta,
- autoload=True,
+ autoload_with=testing.db,
schema=testing.config.test_schema,
oracle_resolve_synonyms=True,
)
child = Table(
"ctable",
meta,
- autoload=True,
+ autoload_with=testing.db,
schema=testing.config.test_schema,
oracle_resolve_synonyms=True,
)
@@ -326,9 +340,6 @@ drop synonym %(test_schema)s.local_table;
"%(test_schema)s.ctable.parent_id"
% {"test_schema": testing.config.test_schema},
)
- select(parent, child).select_from(
- parent.join(child)
- ).execute().fetchall()
class ConstraintTest(fixtures.TablesTest):
@@ -488,9 +499,9 @@ class TableReflectionTest(fixtures.TestBase):
)
metadata.create_all()
- m2 = MetaData(testing.db)
+ m2 = MetaData()
- tbl = Table("test_compress", m2, autoload=True)
+ tbl = Table("test_compress", m2, autoload_with=testing.db)
# Don't hardcode the exact value, but it must be non-empty
assert tbl.dialect_options["oracle"]["compress"]
@@ -507,9 +518,9 @@ class TableReflectionTest(fixtures.TestBase):
)
metadata.create_all()
- m2 = MetaData(testing.db)
+ m2 = MetaData()
- tbl = Table("test_compress", m2, autoload=True)
+ tbl = Table("test_compress", m2, autoload_with=testing.db)
assert tbl.dialect_options["oracle"]["compress"] == "OLTP"
@@ -760,7 +771,6 @@ class DBLinkReflectionTest(fixtures.TestBase):
t = Table(
"test_table_syn",
m,
- autoload=True,
autoload_with=testing.db,
oracle_resolve_synonyms=True,
)
@@ -778,8 +788,8 @@ class TypeReflectionTest(fixtures.TestBase):
m = self.metadata
Table("oracle_types", m, *columns)
m.create_all()
- m2 = MetaData(testing.db)
- table = Table("oracle_types", m2, autoload=True)
+ m2 = MetaData()
+ table = Table("oracle_types", m2, autoload_with=testing.db)
for i, (reflected_col, spec) in enumerate(zip(table.c, specs)):
expected_spec = spec[1]
reflected_type = reflected_col.type
diff --git a/test/dialect/oracle/test_types.py b/test/dialect/oracle/test_types.py
index 0b000e89d..8fbf374ee 100644
--- a/test/dialect/oracle/test_types.py
+++ b/test/dialect/oracle/test_types.py
@@ -313,42 +313,47 @@ class TypesTest(fixtures.TestBase):
Column("numbercol2", oracle.NUMBER(9, 3)),
Column("numbercol3", oracle.NUMBER),
)
- t1.create()
- t1.insert().execute(
- intcol=1,
- numericcol=5.2,
- floatcol1=6.5,
- floatcol2=8.5,
- doubleprec=9.5,
- numbercol1=12,
- numbercol2=14.85,
- numbercol3=15.76,
- )
+ with testing.db.begin() as conn:
+ t1.create(conn)
+ conn.execute(
+ t1.insert(),
+ dict(
+ intcol=1,
+ numericcol=5.2,
+ floatcol1=6.5,
+ floatcol2=8.5,
+ doubleprec=9.5,
+ numbercol1=12,
+ numbercol2=14.85,
+ numbercol3=15.76,
+ ),
+ )
- m2 = MetaData(testing.db)
- t2 = Table("t1", m2, autoload=True)
+ m2 = MetaData()
+ t2 = Table("t1", m2, autoload_with=testing.db)
- for row in (
- t1.select().execute().first(),
- t2.select().execute().first(),
- ):
- for i, (val, type_) in enumerate(
- (
- (1, int),
- (decimal.Decimal("5.2"), decimal.Decimal),
- (6.5, float),
- (8.5, float),
- (9.5, float),
- (12, int),
- (decimal.Decimal("14.85"), decimal.Decimal),
- (15.76, float),
- )
+ with testing.db.connect() as conn:
+ for row in (
+ conn.execute(t1.select()).first(),
+ conn.execute(t2.select()).first(),
):
- eq_(row[i], val)
- assert isinstance(row[i], type_), "%r is not %r" % (
- row[i],
- type_,
- )
+ for i, (val, type_) in enumerate(
+ (
+ (1, int),
+ (decimal.Decimal("5.2"), decimal.Decimal),
+ (6.5, float),
+ (8.5, float),
+ (9.5, float),
+ (12, int),
+ (decimal.Decimal("14.85"), decimal.Decimal),
+ (15.76, float),
+ )
+ ):
+ eq_(row[i], val)
+ assert isinstance(row[i], type_), "%r is not %r" % (
+ row[i],
+ type_,
+ )
@testing.provide_metadata
def test_numeric_infinity_float(self, connection):
@@ -743,8 +748,8 @@ class TypesTest(fixtures.TestBase):
Column("d5", oracle.INTERVAL(second_precision=5)),
)
metadata.create_all()
- m = MetaData(testing.db)
- t1 = Table("date_types", m, autoload=True)
+ m = MetaData()
+ t1 = Table("date_types", m, autoload_with=testing.db)
assert isinstance(t1.c.d1.type, oracle.DATE)
assert isinstance(t1.c.d1.type, DateTime)
assert isinstance(t1.c.d2.type, oracle.DATE)
@@ -758,10 +763,10 @@ class TypesTest(fixtures.TestBase):
def _dont_test_reflect_all_types_schema(self):
types_table = Table(
"all_types",
- MetaData(testing.db),
+ MetaData(),
Column("owner", String(30), primary_key=True),
Column("type_name", String(30), primary_key=True),
- autoload=True,
+ autoload_with=testing.db,
oracle_resolve_synonyms=True,
)
for row in types_table.select().execute().fetchall():
@@ -790,8 +795,8 @@ class TypesTest(fixtures.TestBase):
Column("c_data", sqltypes.NCHAR(20)),
)
metadata.create_all()
- m2 = MetaData(testing.db)
- t2 = Table("tnv", m2, autoload=True)
+ m2 = MetaData()
+ t2 = Table("tnv", m2, autoload_with=testing.db)
assert isinstance(t2.c.nv_data.type, sqltypes.NVARCHAR)
assert isinstance(t2.c.c_data.type, sqltypes.NCHAR)
@@ -820,8 +825,8 @@ class TypesTest(fixtures.TestBase):
metadata = self.metadata
Table("tnv", metadata, Column("data", sqltypes.Unicode(255)))
metadata.create_all()
- m2 = MetaData(testing.db)
- t2 = Table("tnv", m2, autoload=True)
+ m2 = MetaData()
+ t2 = Table("tnv", m2, autoload_with=testing.db)
assert isinstance(t2.c.data.type, sqltypes.VARCHAR)
if testing.against("oracle+cx_oracle"):
@@ -831,10 +836,11 @@ class TypesTest(fixtures.TestBase):
)
data = u("m’a réveillé.")
- t2.insert().execute(data=data)
- res = t2.select().execute().first()["data"]
- eq_(res, data)
- assert isinstance(res, util.text_type)
+ with testing.db.begin() as conn:
+ conn.execute(t2.insert(), {"data": data})
+ res = conn.execute(t2.select()).first().data
+ eq_(res, data)
+ assert isinstance(res, util.text_type)
@testing.provide_metadata
def test_char_length(self):
@@ -848,8 +854,8 @@ class TypesTest(fixtures.TestBase):
Column("c4", NCHAR(180)),
)
t1.create()
- m2 = MetaData(testing.db)
- t2 = Table("t1", m2, autoload=True)
+ m2 = MetaData()
+ t2 = Table("t1", m2, autoload_with=testing.db)
eq_(t2.c.c1.type.length, 50)
eq_(t2.c.c2.type.length, 250)
eq_(t2.c.c3.type.length, 200)
diff --git a/test/dialect/postgresql/test_dialect.py b/test/dialect/postgresql/test_dialect.py
index 8ee5a6e2f..d1e9c2e6d 100644
--- a/test/dialect/postgresql/test_dialect.py
+++ b/test/dialect/postgresql/test_dialect.py
@@ -752,8 +752,8 @@ class MiscBackendTest(
Column("date2", DateTime(timezone=False)),
)
metadata.create_all()
- m2 = MetaData(testing.db)
- t2 = Table("pgdate", m2, autoload=True)
+ m2 = MetaData()
+ t2 = Table("pgdate", m2, autoload_with=testing.db)
assert t2.c.date1.type.timezone is True
assert t2.c.date2.type.timezone is False
diff --git a/test/dialect/postgresql/test_query.py b/test/dialect/postgresql/test_query.py
index 3484fa4c3..c959acf35 100644
--- a/test/dialect/postgresql/test_query.py
+++ b/test/dialect/postgresql/test_query.py
@@ -246,8 +246,8 @@ class InsertTest(fixtures.TestBase, AssertsExecutionResults):
# test the same series of events using a reflected version of
# the table
- m2 = MetaData(engine)
- table = Table(table.name, m2, autoload=True)
+ m2 = MetaData()
+ table = Table(table.name, m2, autoload_with=engine)
with self.sql_execution_asserter(engine) as asserter:
with engine.connect() as conn:
@@ -386,8 +386,8 @@ class InsertTest(fixtures.TestBase, AssertsExecutionResults):
# test the same series of events using a reflected version of
# the table
- m2 = MetaData(engine)
- table = Table(table.name, m2, autoload=True)
+ m2 = MetaData()
+ table = Table(table.name, m2, autoload_with=engine)
with self.sql_execution_asserter(engine) as asserter:
with engine.connect() as conn:
@@ -632,8 +632,8 @@ class InsertTest(fixtures.TestBase, AssertsExecutionResults):
# test the same series of events using a reflected version of
# the table
- m2 = MetaData(engine)
- table = Table(table.name, m2, autoload=True)
+ m2 = MetaData()
+ table = Table(table.name, m2, autoload_with=engine)
with engine.connect() as conn:
conn.execute(table.insert(), {"id": 30, "data": "d1"})
diff --git a/test/dialect/postgresql/test_reflection.py b/test/dialect/postgresql/test_reflection.py
index 2e79d071c..4de4d88e3 100644
--- a/test/dialect/postgresql/test_reflection.py
+++ b/test/dialect/postgresql/test_reflection.py
@@ -81,8 +81,8 @@ class ForeignTableReflectionTest(fixtures.TablesTest, AssertsExecutionResults):
sa.event.listen(metadata, "before_drop", sa.DDL(ddl))
def test_foreign_table_is_reflected(self):
- metadata = MetaData(testing.db)
- table = Table("test_foreigntable", metadata, autoload=True)
+ metadata = MetaData()
+ table = Table("test_foreigntable", metadata, autoload_with=testing.db)
eq_(
set(table.columns.keys()),
set(["id", "data"]),
@@ -221,8 +221,8 @@ class MaterializedViewReflectionTest(
)
def test_mview_is_reflected(self):
- metadata = MetaData(testing.db)
- table = Table("test_mview", metadata, autoload=True)
+ metadata = MetaData()
+ table = Table("test_mview", metadata, autoload_with=testing.db)
eq_(
set(table.columns.keys()),
set(["id", "data"]),
@@ -230,9 +230,10 @@ class MaterializedViewReflectionTest(
)
def test_mview_select(self):
- metadata = MetaData(testing.db)
- table = Table("test_mview", metadata, autoload=True)
- eq_(table.select().execute().fetchall(), [(89, "d1")])
+ metadata = MetaData()
+ table = Table("test_mview", metadata, autoload_with=testing.db)
+ with testing.db.connect() as conn:
+ eq_(conn.execute(table.select()).fetchall(), [(89, "d1")])
def test_get_view_names(self):
insp = inspect(testing.db)
@@ -349,8 +350,8 @@ class DomainReflectionTest(fixtures.TestBase, AssertsExecutionResults):
con.exec_driver_sql('DROP SCHEMA "SomeSchema"')
def test_table_is_reflected(self):
- metadata = MetaData(testing.db)
- table = Table("testtable", metadata, autoload=True)
+ metadata = MetaData()
+ table = Table("testtable", metadata, autoload_with=testing.db)
eq_(
set(table.columns.keys()),
set(["question", "answer"]),
@@ -359,8 +360,8 @@ class DomainReflectionTest(fixtures.TestBase, AssertsExecutionResults):
assert isinstance(table.c.answer.type, Integer)
def test_domain_is_reflected(self):
- metadata = MetaData(testing.db)
- table = Table("testtable", metadata, autoload=True)
+ metadata = MetaData()
+ table = Table("testtable", metadata, autoload_with=testing.db)
eq_(
str(table.columns.answer.server_default.arg),
"42",
@@ -371,25 +372,28 @@ class DomainReflectionTest(fixtures.TestBase, AssertsExecutionResults):
), "Expected reflected column to not be nullable."
def test_enum_domain_is_reflected(self):
- metadata = MetaData(testing.db)
- table = Table("enum_test", metadata, autoload=True)
+ metadata = MetaData()
+ table = Table("enum_test", metadata, autoload_with=testing.db)
eq_(table.c.data.type.enums, ["test"])
def test_array_domain_is_reflected(self):
- metadata = MetaData(testing.db)
- table = Table("array_test", metadata, autoload=True)
+ metadata = MetaData()
+ table = Table("array_test", metadata, autoload_with=testing.db)
eq_(table.c.data.type.__class__, ARRAY)
eq_(table.c.data.type.item_type.__class__, INTEGER)
def test_quoted_remote_schema_domain_is_reflected(self):
- metadata = MetaData(testing.db)
- table = Table("quote_test", metadata, autoload=True)
+ metadata = MetaData()
+ table = Table("quote_test", metadata, autoload_with=testing.db)
eq_(table.c.data.type.__class__, INTEGER)
def test_table_is_reflected_test_schema(self):
- metadata = MetaData(testing.db)
+ metadata = MetaData()
table = Table(
- "testtable", metadata, autoload=True, schema="test_schema"
+ "testtable",
+ metadata,
+ autoload_with=testing.db,
+ schema="test_schema",
)
eq_(
set(table.columns.keys()),
@@ -399,9 +403,12 @@ class DomainReflectionTest(fixtures.TestBase, AssertsExecutionResults):
assert isinstance(table.c.anything.type, Integer)
def test_schema_domain_is_reflected(self):
- metadata = MetaData(testing.db)
+ metadata = MetaData()
table = Table(
- "testtable", metadata, autoload=True, schema="test_schema"
+ "testtable",
+ metadata,
+ autoload_with=testing.db,
+ schema="test_schema",
)
eq_(
str(table.columns.answer.server_default.arg),
@@ -413,8 +420,8 @@ class DomainReflectionTest(fixtures.TestBase, AssertsExecutionResults):
), "Expected reflected column to be nullable."
def test_crosschema_domain_is_reflected(self):
- metadata = MetaData(testing.db)
- table = Table("crosschema", metadata, autoload=True)
+ metadata = MetaData()
+ table = Table("crosschema", metadata, autoload_with=testing.db)
eq_(
str(table.columns.answer.server_default.arg),
"0",
@@ -430,13 +437,15 @@ class DomainReflectionTest(fixtures.TestBase, AssertsExecutionResults):
ischema_names = base.PGDialect.ischema_names
base.PGDialect.ischema_names = {}
try:
- m2 = MetaData(testing.db)
- assert_raises(exc.SAWarning, Table, "testtable", m2, autoload=True)
+ m2 = MetaData()
+ assert_raises(
+ exc.SAWarning, Table, "testtable", m2, autoload_with=testing.db
+ )
@testing.emits_warning("Did not recognize type")
def warns():
- m3 = MetaData(testing.db)
- t3 = Table("testtable", m3, autoload=True)
+ m3 = MetaData()
+ t3 = Table("testtable", m3, autoload_with=testing.db)
assert t3.c.answer.type.__class__ == sa.types.NullType
finally:
@@ -461,8 +470,8 @@ class ReflectionTest(AssertsCompiledSQL, fixtures.TestBase):
PrimaryKeyConstraint("p2", "p1"),
)
meta1.create_all()
- meta2 = MetaData(testing.db)
- subject = Table("subject", meta2, autoload=True)
+ meta2 = MetaData()
+ subject = Table("subject", meta2, autoload_with=testing.db)
eq_(subject.primary_key.columns.keys(), ["p2", "p1"])
@testing.provide_metadata
@@ -478,9 +487,9 @@ class ReflectionTest(AssertsCompiledSQL, fixtures.TestBase):
Column("ref", Integer, ForeignKey("subject.id$")),
)
meta1.create_all()
- meta2 = MetaData(testing.db)
- subject = Table("subject", meta2, autoload=True)
- referer = Table("referer", meta2, autoload=True)
+ meta2 = MetaData()
+ subject = Table("subject", meta2, autoload_with=testing.db)
+ referer = Table("referer", meta2, autoload_with=testing.db)
self.assert_(
(subject.c["id$"] == referer.c.ref).compare(
subject.join(referer).onclause
@@ -496,7 +505,7 @@ class ReflectionTest(AssertsCompiledSQL, fixtures.TestBase):
).create(testing.db)
m = MetaData()
- t = Table("t", m, autoload=True, autoload_with=testing.db)
+ t = Table("t", m, autoload_with=testing.db)
eq_(
t.c.x.server_default.arg.text,
"'%s'::character varying" % ("abcd" * 40),
@@ -507,23 +516,25 @@ class ReflectionTest(AssertsCompiledSQL, fixtures.TestBase):
def test_renamed_sequence_reflection(self):
metadata = self.metadata
Table("t", metadata, Column("id", Integer, primary_key=True))
- metadata.create_all()
- m2 = MetaData(testing.db)
- t2 = Table("t", m2, autoload=True, implicit_returning=False)
+ metadata.create_all(testing.db)
+ m2 = MetaData()
+ t2 = Table("t", m2, autoload_with=testing.db, implicit_returning=False)
eq_(t2.c.id.server_default.arg.text, "nextval('t_id_seq'::regclass)")
- r = t2.insert().execute()
- eq_(r.inserted_primary_key, (1,))
+ with testing.db.begin() as conn:
+ r = conn.execute(t2.insert())
+ eq_(r.inserted_primary_key, (1,))
testing.db.connect().execution_options(
autocommit=True
).exec_driver_sql("alter table t_id_seq rename to foobar_id_seq")
- m3 = MetaData(testing.db)
- t3 = Table("t", m3, autoload=True, implicit_returning=False)
+ m3 = MetaData()
+ t3 = Table("t", m3, autoload_with=testing.db, implicit_returning=False)
eq_(
t3.c.id.server_default.arg.text,
"nextval('foobar_id_seq'::regclass)",
)
- r = t3.insert().execute()
- eq_(r.inserted_primary_key, (2,))
+ with testing.db.begin() as conn:
+ r = conn.execute(t3.insert())
+ eq_(r.inserted_primary_key, (2,))
@testing.provide_metadata
def test_altered_type_autoincrement_pk_reflection(self):
@@ -538,8 +549,8 @@ class ReflectionTest(AssertsCompiledSQL, fixtures.TestBase):
testing.db.connect().execution_options(
autocommit=True
).exec_driver_sql("alter table t alter column id type varchar(50)")
- m2 = MetaData(testing.db)
- t2 = Table("t", m2, autoload=True)
+ m2 = MetaData()
+ t2 = Table("t", m2, autoload_with=testing.db)
eq_(t2.c.id.autoincrement, False)
eq_(t2.c.x.autoincrement, False)
@@ -551,8 +562,8 @@ class ReflectionTest(AssertsCompiledSQL, fixtures.TestBase):
testing.db.connect().execution_options(
autocommit=True
).exec_driver_sql("alter table t rename id to t_id")
- m2 = MetaData(testing.db)
- t2 = Table("t", m2, autoload=True)
+ m2 = MetaData()
+ t2 = Table("t", m2, autoload_with=testing.db)
eq_([c.name for c in t2.primary_key], ["t_id"])
@testing.provide_metadata
@@ -589,9 +600,12 @@ class ReflectionTest(AssertsCompiledSQL, fixtures.TestBase):
schema="test_schema",
)
meta1.create_all()
- meta2 = MetaData(testing.db)
+ meta2 = MetaData()
addresses = Table(
- "email_addresses", meta2, autoload=True, schema="test_schema"
+ "email_addresses",
+ meta2,
+ autoload_with=testing.db,
+ schema="test_schema",
)
users = Table("users", meta2, must_exist=True, schema="test_schema")
j = join(users, addresses)
@@ -613,9 +627,11 @@ class ReflectionTest(AssertsCompiledSQL, fixtures.TestBase):
schema="test_schema",
)
meta1.create_all()
- meta2 = MetaData(testing.db)
- subject = Table("subject", meta2, autoload=True)
- referer = Table("referer", meta2, schema="test_schema", autoload=True)
+ meta2 = MetaData()
+ subject = Table("subject", meta2, autoload_with=testing.db)
+ referer = Table(
+ "referer", meta2, schema="test_schema", autoload_with=testing.db
+ )
self.assert_(
(subject.c.id == referer.c.ref).compare(
subject.join(referer).onclause
@@ -639,11 +655,13 @@ class ReflectionTest(AssertsCompiledSQL, fixtures.TestBase):
schema="test_schema",
)
meta1.create_all()
- meta2 = MetaData(testing.db)
+ meta2 = MetaData()
subject = Table(
- "subject", meta2, autoload=True, schema="test_schema_2"
+ "subject", meta2, autoload_with=testing.db, schema="test_schema_2"
+ )
+ referer = Table(
+ "referer", meta2, autoload_with=testing.db, schema="test_schema"
)
- referer = Table("referer", meta2, autoload=True, schema="test_schema")
self.assert_(
(subject.c.id == referer.c.ref).compare(
subject.join(referer).onclause
@@ -675,14 +693,14 @@ class ReflectionTest(AssertsCompiledSQL, fixtures.TestBase):
subject = Table(
"subject",
meta2,
- autoload=True,
+ autoload_with=testing.db,
schema="test_schema_2",
postgresql_ignore_search_path=True,
)
referer = Table(
"referer",
meta2,
- autoload=True,
+ autoload_with=testing.db,
schema="test_schema",
postgresql_ignore_search_path=True,
)
@@ -710,18 +728,18 @@ class ReflectionTest(AssertsCompiledSQL, fixtures.TestBase):
)
meta1.create_all()
- meta2 = MetaData(testing.db)
+ meta2 = MetaData()
subject = Table(
"subject",
meta2,
- autoload=True,
+ autoload_with=testing.db,
schema=default_schema,
postgresql_ignore_search_path=True,
)
referer = Table(
"referer",
meta2,
- autoload=True,
+ autoload_with=testing.db,
schema=default_schema,
postgresql_ignore_search_path=True,
)
@@ -759,30 +777,33 @@ class ReflectionTest(AssertsCompiledSQL, fixtures.TestBase):
"set search_path to test_schema_2, test_schema, public"
)
- m1 = MetaData(conn)
+ m1 = MetaData()
- Table("some_table", m1, schema="test_schema", autoload=True)
+ Table("some_table", m1, schema="test_schema", autoload_with=conn)
t2_schema = Table(
- "some_other_table", m1, schema="test_schema_2", autoload=True
+ "some_other_table",
+ m1,
+ schema="test_schema_2",
+ autoload_with=conn,
)
- t2_no_schema = Table("some_other_table", m1, autoload=True)
+ t2_no_schema = Table("some_other_table", m1, autoload_with=conn)
- t1_no_schema = Table("some_table", m1, autoload=True)
+ t1_no_schema = Table("some_table", m1, autoload_with=conn)
- m2 = MetaData(conn)
+ m2 = MetaData()
t1_schema_isp = Table(
"some_table",
m2,
schema="test_schema",
- autoload=True,
+ autoload_with=conn,
postgresql_ignore_search_path=True,
)
t2_schema_isp = Table(
"some_other_table",
m2,
schema="test_schema_2",
- autoload=True,
+ autoload_with=conn,
postgresql_ignore_search_path=True,
)
@@ -929,8 +950,8 @@ class ReflectionTest(AssertsCompiledSQL, fixtures.TestBase):
)
def go():
- m2 = MetaData(testing.db)
- t2 = Table("party", m2, autoload=True)
+ m2 = MetaData()
+ t2 = Table("party", m2, autoload_with=testing.db)
assert len(t2.indexes) == 2
# Make sure indexes are in the order we expect them in
@@ -1039,7 +1060,7 @@ class ReflectionTest(AssertsCompiledSQL, fixtures.TestBase):
# reflect data
with testing.db.connect() as conn:
m2 = MetaData(conn)
- t2 = Table("party", m2, autoload=True)
+ t2 = Table("party", m2, autoload_with=testing.db)
eq_(len(t2.indexes), 3)
@@ -1515,7 +1536,7 @@ class ReflectionTest(AssertsCompiledSQL, fixtures.TestBase):
"t", self.metadata, Column("x", postgresql.ENUM(name="empty"))
).create(testing.db)
- t = Table("t", MetaData(testing.db), autoload_with=testing.db)
+ t = Table("t", MetaData(), autoload_with=testing.db)
eq_(t.c.x.type.enums, [])
@testing.provide_metadata
@@ -1544,7 +1565,7 @@ class ReflectionTest(AssertsCompiledSQL, fixtures.TestBase):
self.assert_("uc_a" in constraints)
# reflection corrects for the dupe
- reflected = Table("pgsql_uc", MetaData(testing.db), autoload=True)
+ reflected = Table("pgsql_uc", MetaData(), autoload_with=testing.db)
indexes = set(i.name for i in reflected.indexes)
constraints = set(uc.name for uc in reflected.constraints)
@@ -1585,7 +1606,7 @@ class ReflectionTest(AssertsCompiledSQL, fixtures.TestBase):
eq_(insp.get_indexes("t"), expected)
# reflection corrects for the dupe
- reflected = Table("t", MetaData(testing.db), autoload=True)
+ reflected = Table("t", MetaData(), autoload_with=testing.db)
eq_(set(reflected.indexes), set())
@@ -1615,7 +1636,7 @@ class ReflectionTest(AssertsCompiledSQL, fixtures.TestBase):
assert indexes["ix_a"]["unique"]
self.assert_("ix_a" not in constraints)
- reflected = Table("pgsql_uc", MetaData(testing.db), autoload=True)
+ reflected = Table("pgsql_uc", MetaData(), autoload_with=testing.db)
indexes = dict((i.name, i) for i in reflected.indexes)
constraints = set(uc.name for uc in reflected.constraints)
diff --git a/test/dialect/postgresql/test_types.py b/test/dialect/postgresql/test_types.py
index 509603e1b..e7174f234 100644
--- a/test/dialect/postgresql/test_types.py
+++ b/test/dialect/postgresql/test_types.py
@@ -273,8 +273,8 @@ class EnumTest(fixtures.TestBase, AssertsExecutionResults):
(3, util.u("S’il")),
],
)
- m2 = MetaData(testing.db)
- t2 = Table("table", m2, autoload=True)
+ m2 = MetaData()
+ t2 = Table("table", m2, autoload_with=testing.db)
eq_(
t2.c.value.type.enums,
[util.u("réveillé"), util.u("drôle"), util.u("S’il")],
@@ -671,8 +671,8 @@ class EnumTest(fixtures.TestBase, AssertsExecutionResults):
Column("value2", etype),
)
metadata.create_all()
- m2 = MetaData(testing.db)
- t2 = Table("table", m2, autoload=True)
+ m2 = MetaData()
+ t2 = Table("table", m2, autoload_with=testing.db)
eq_(t2.c.value.type.enums, ["one", "two", "three"])
eq_(t2.c.value.type.name, "onetwothreetype")
eq_(t2.c.value2.type.enums, ["four", "five", "six"])
@@ -706,8 +706,8 @@ class EnumTest(fixtures.TestBase, AssertsExecutionResults):
Column("value2", etype),
)
metadata.create_all()
- m2 = MetaData(testing.db)
- t2 = Table("table", m2, autoload=True)
+ m2 = MetaData()
+ t2 = Table("table", m2, autoload_with=testing.db)
eq_(t2.c.value.type.enums, ["one", "two", "three"])
eq_(t2.c.value.type.name, "onetwothreetype")
eq_(t2.c.value2.type.enums, ["four", "five", "six"])
@@ -821,7 +821,11 @@ class OIDTest(fixtures.TestBase):
)
metadata.create_all()
m2 = MetaData()
- t2 = Table("table", m2, autoload_with=testing.db, autoload=True)
+ t2 = Table(
+ "table",
+ m2,
+ autoload_with=testing.db,
+ )
assert isinstance(t2.c.y.type, postgresql.OID)
@@ -1071,8 +1075,8 @@ class TimePrecisionTest(fixtures.TestBase):
Column("c6", postgresql.TIMESTAMP(timezone=True, precision=5)),
)
t1.create()
- m2 = MetaData(testing.db)
- t2 = Table("t1", m2, autoload=True)
+ m2 = MetaData()
+ t2 = Table("t1", m2, autoload_with=testing.db)
eq_(t2.c.c1.type.precision, None)
eq_(t2.c.c2.type.precision, 5)
eq_(t2.c.c3.type.precision, 5)
@@ -1380,8 +1384,8 @@ class ArrayRoundTripTest(object):
conn.execute(table.insert(), intarr=[4, 5, 6])
def test_reflect_array_column(self):
- metadata2 = MetaData(testing.db)
- tbl = Table("arrtable", metadata2, autoload=True)
+ metadata2 = MetaData()
+ tbl = Table("arrtable", metadata2, autoload_with=testing.db)
assert isinstance(tbl.c.intarr.type, self.ARRAY)
assert isinstance(tbl.c.strarr.type, self.ARRAY)
assert isinstance(tbl.c.intarr.type.item_type, Integer)
@@ -2156,8 +2160,8 @@ class SpecialTypesTest(fixtures.TablesTest, ComparesTables):
special_types_table.c.year_interval.type = postgresql.INTERVAL()
special_types_table.c.month_interval.type = postgresql.INTERVAL()
- m = MetaData(testing.db)
- t = Table("sometable", m, autoload=True)
+ m = MetaData()
+ t = Table("sometable", m, autoload_with=testing.db)
self.assert_tables_equal(special_types_table, t, strict_types=True)
assert t.c.plain_interval.type.precision is None
@@ -2190,8 +2194,8 @@ class SpecialTypesTest(fixtures.TablesTest, ComparesTables):
Column("bitvarying5", postgresql.BIT(5, varying=True)),
)
t1.create()
- m2 = MetaData(testing.db)
- t2 = Table("t1", m2, autoload=True)
+ m2 = MetaData()
+ t2 = Table("t1", m2, autoload_with=testing.db)
eq_(t2.c.bit1.type.length, 1)
eq_(t2.c.bit1.type.varying, False)
eq_(t2.c.bit5.type.length, 5)
diff --git a/test/dialect/test_sqlite.py b/test/dialect/test_sqlite.py
index 4f581433c..d06cd48f5 100644
--- a/test/dialect/test_sqlite.py
+++ b/test/dialect/test_sqlite.py
@@ -486,8 +486,8 @@ class DefaultsTest(fixtures.TestBase, AssertsCompiledSQL):
Table("t_defaults", m, *columns)
try:
m.create_all()
- m2 = MetaData(db)
- rt = Table("t_defaults", m2, autoload=True)
+ m2 = MetaData()
+ rt = Table("t_defaults", m2, autoload_with=db)
expected = [c[1] for c in specs]
for i, reflected in enumerate(rt.c):
eq_(str(reflected.server_default.arg), expected[i])
@@ -505,7 +505,7 @@ class DefaultsTest(fixtures.TestBase, AssertsCompiledSQL):
def test_default_reflection_2(self):
db = testing.db
- m = MetaData(db)
+ m = MetaData()
expected = ["'my_default'", "0"]
table = """CREATE TABLE r_defaults (
data VARCHAR(40) DEFAULT 'my_default',
@@ -513,7 +513,7 @@ class DefaultsTest(fixtures.TestBase, AssertsCompiledSQL):
)"""
try:
exec_sql(db, table)
- rt = Table("r_defaults", m, autoload=True)
+ rt = Table("r_defaults", m, autoload_with=db)
for i, reflected in enumerate(rt.c):
eq_(str(reflected.server_default.arg), expected[i])
finally:
@@ -527,12 +527,12 @@ class DefaultsTest(fixtures.TestBase, AssertsCompiledSQL):
)"""
try:
exec_sql(db, table)
- m1 = MetaData(db)
- t1 = Table("r_defaults", m1, autoload=True)
+ m1 = MetaData()
+ t1 = Table("r_defaults", m1, autoload_with=db)
exec_sql(db, "DROP TABLE r_defaults")
- t1.create()
- m2 = MetaData(db)
- t2 = Table("r_defaults", m2, autoload=True)
+ t1.create(db)
+ m2 = MetaData()
+ t2 = Table("r_defaults", m2, autoload_with=db)
self.assert_compile(
CreateTable(t2),
"CREATE TABLE r_defaults (data VARCHAR(40) "
@@ -672,8 +672,10 @@ class DialectTest(
)
""",
)
- table1 = Table("django_admin_log", metadata, autoload=True)
- table2 = Table("django_content_type", metadata, autoload=True)
+ table1 = Table("django_admin_log", metadata, autoload_with=testing.db)
+ table2 = Table(
+ "django_content_type", metadata, autoload_with=testing.db
+ )
j = table1.join(table2)
assert j.onclause.compare(table1.c.content_type_id == table2.c.id)
@@ -708,14 +710,9 @@ class DialectTest(
# )
# ''')
- table1 = Table(r'"a"', metadata, autoload=True)
+ table1 = Table(r'"a"', metadata, autoload_with=testing.db)
assert '"id"' in table1.c
- # table2 = Table(r'"b"', metadata, autoload=True)
- # j = table1.join(table2)
- # assert j.onclause.compare(table1.c['"id"']
- # == table2.c['"aid"'])
-
@testing.provide_metadata
def test_description_encoding(self, connection):
# amazingly, pysqlite seems to still deliver cursor.description
@@ -915,7 +912,6 @@ class AttachedDBTest(fixtures.TestBase):
alt_master = Table(
"sqlite_master",
meta,
- autoload=True,
autoload_with=self.conn,
schema="test_schema",
)
@@ -925,7 +921,7 @@ class AttachedDBTest(fixtures.TestBase):
self._fixture()
m2 = MetaData()
- c2 = Table("created", m2, autoload=True, autoload_with=self.conn)
+ c2 = Table("created", m2, autoload_with=self.conn)
eq_(len(c2.c), 2)
def test_crud(self):
@@ -1518,7 +1514,7 @@ class MatchTest(fixtures.TestBase, AssertsCompiledSQL):
@classmethod
def setup_class(cls):
global metadata, cattable, matchtable
- metadata = MetaData(testing.db)
+ metadata = MetaData()
exec_sql(
testing.db,
"""
@@ -1529,7 +1525,7 @@ class MatchTest(fixtures.TestBase, AssertsCompiledSQL):
)
""",
)
- cattable = Table("cattable", metadata, autoload=True)
+ cattable = Table("cattable", metadata, autoload_with=testing.db)
exec_sql(
testing.db,
"""
@@ -1541,39 +1537,47 @@ class MatchTest(fixtures.TestBase, AssertsCompiledSQL):
)
""",
)
- matchtable = Table("matchtable", metadata, autoload=True)
- metadata.create_all()
- cattable.insert().execute(
- [
- {"id": 1, "description": "Python"},
- {"id": 2, "description": "Ruby"},
- ]
- )
- matchtable.insert().execute(
- [
- {
- "id": 1,
- "title": "Agile Web Development with Rails",
- "category_id": 2,
- },
- {"id": 2, "title": "Dive Into Python", "category_id": 1},
- {
- "id": 3,
- "title": "Programming Matz's Ruby",
- "category_id": 2,
- },
- {
- "id": 4,
- "title": "The Definitive Guide to Django",
- "category_id": 1,
- },
- {"id": 5, "title": "Python in a Nutshell", "category_id": 1},
- ]
- )
+ matchtable = Table("matchtable", metadata, autoload_with=testing.db)
+ with testing.db.begin() as conn:
+ metadata.create_all(conn)
+
+ conn.execute(
+ cattable.insert(),
+ [
+ {"id": 1, "description": "Python"},
+ {"id": 2, "description": "Ruby"},
+ ],
+ )
+ conn.execute(
+ matchtable.insert(),
+ [
+ {
+ "id": 1,
+ "title": "Agile Web Development with Rails",
+ "category_id": 2,
+ },
+ {"id": 2, "title": "Dive Into Python", "category_id": 1},
+ {
+ "id": 3,
+ "title": "Programming Matz's Ruby",
+ "category_id": 2,
+ },
+ {
+ "id": 4,
+ "title": "The Definitive Guide to Django",
+ "category_id": 1,
+ },
+ {
+ "id": 5,
+ "title": "Python in a Nutshell",
+ "category_id": 1,
+ },
+ ],
+ )
@classmethod
def teardown_class(cls):
- metadata.drop_all()
+ metadata.drop_all(testing.db)
def test_expression(self):
self.assert_compile(
@@ -1582,46 +1586,38 @@ class MatchTest(fixtures.TestBase, AssertsCompiledSQL):
dialect=sqlite.dialect(),
)
- def test_simple_match(self):
- results = (
+ def test_simple_match(self, connection):
+ results = connection.execute(
matchtable.select()
.where(matchtable.c.title.match("python"))
.order_by(matchtable.c.id)
- .execute()
- .fetchall()
- )
+ ).fetchall()
eq_([2, 5], [r.id for r in results])
- def test_simple_prefix_match(self):
- results = (
- matchtable.select()
- .where(matchtable.c.title.match("nut*"))
- .execute()
- .fetchall()
- )
+ def test_simple_prefix_match(self, connection):
+ results = connection.execute(
+ matchtable.select().where(matchtable.c.title.match("nut*"))
+ ).fetchall()
eq_([5], [r.id for r in results])
- def test_or_match(self):
- results2 = (
+ def test_or_match(self, connection):
+ results2 = connection.execute(
matchtable.select()
.where(matchtable.c.title.match("nutshell OR ruby"))
.order_by(matchtable.c.id)
- .execute()
- .fetchall()
- )
+ ).fetchall()
eq_([3, 5], [r.id for r in results2])
- def test_and_match(self):
- results2 = (
- matchtable.select()
- .where(matchtable.c.title.match("python nutshell"))
- .execute()
- .fetchall()
- )
+ def test_and_match(self, connection):
+ results2 = connection.execute(
+ matchtable.select().where(
+ matchtable.c.title.match("python nutshell")
+ )
+ ).fetchall()
eq_([5], [r.id for r in results2])
- def test_match_across_joins(self):
- results = (
+ def test_match_across_joins(self, connection):
+ results = connection.execute(
matchtable.select()
.where(
and_(
@@ -1630,9 +1626,7 @@ class MatchTest(fixtures.TestBase, AssertsCompiledSQL):
)
)
.order_by(matchtable.c.id)
- .execute()
- .fetchall()
- )
+ ).fetchall()
eq_([1, 3], [r.id for r in results])
@@ -1718,8 +1712,8 @@ class ReflectHeadlessFKsTest(fixtures.TestBase):
def test_reflect_tables_fk_no_colref(self):
meta = MetaData()
- a = Table("a", meta, autoload=True, autoload_with=testing.db)
- b = Table("b", meta, autoload=True, autoload_with=testing.db)
+ a = Table("a", meta, autoload_with=testing.db)
+ b = Table("b", meta, autoload_with=testing.db)
assert b.c.id.references(a.c.id)
View Lorry Controller Status