summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
-rw-r--r--doc/build/changelog/changelog_09.rst24
-rw-r--r--lib/sqlalchemy/sql/schema.py85
-rw-r--r--test/sql/test_metadata.py476
3 files changed, 399 insertions, 186 deletions
diff --git a/doc/build/changelog/changelog_09.rst b/doc/build/changelog/changelog_09.rst
index 573548cfa..dbd26bd1f 100644
--- a/doc/build/changelog/changelog_09.rst
+++ b/doc/build/changelog/changelog_09.rst
@@ -15,6 +15,30 @@
:version: 0.9.2
.. change::
+ :tags: bug, sql
+ :tickets: 2913
+
+ The behavior of :meth:`.Table.tometadata` has been adjusted such that
+ the schema target of a :class:`.ForeignKey` will not be changed unless
+ that schema matches that of the parent table. That is, if
+ a table "schema_a.user" has a foreign key to "schema_b.order.id",
+ the "schema_b" target will be maintained whether or not the
+ "schema" argument is passed to :meth:`.Table.tometadata`. However
+ if a table "schema_a.user" refers to "schema_a.order.id", the presence
+ of "schema_a" will be updated on both the parent and referred tables.
+ This is a behavioral change hence isn't likely to be backported to
+ 0.8; it is assumed that the previous behavior is pretty buggy
+ however and that it's unlikely anyone was relying upon it.
+
+ Additionally, a new parameter has been added
+ :paramref:`.Table.tometadata.referred_schema_fn`. This refers to a
+ callable function which will be used to determine the new referred
+ schema for any :class:`.ForeignKeyConstraint` encountered in the
+ tometadata operation. This callable can be used to revert to the
+ previous behavior or to customize how referred schemas are treated
+ on a per-constraint basis.
+
+ .. change::
:tags: bug, orm
:tickets: 2932
diff --git a/lib/sqlalchemy/sql/schema.py b/lib/sqlalchemy/sql/schema.py
index 621ac20e8..ef4b4cfa5 100644
--- a/lib/sqlalchemy/sql/schema.py
+++ b/lib/sqlalchemy/sql/schema.py
@@ -661,31 +661,61 @@ class Table(DialectKWArgs, SchemaItem, TableClause):
self,
checkfirst=checkfirst)
- def tometadata(self, metadata, schema=RETAIN_SCHEMA):
+ def tometadata(self, metadata, schema=RETAIN_SCHEMA, referred_schema_fn=None):
"""Return a copy of this :class:`.Table` associated with a different
:class:`.MetaData`.
E.g.::
- some_engine = create_engine("sqlite:///some.db")
+ m1 = MetaData()
- # create two metadata
- meta1 = MetaData()
- meta2 = MetaData()
+ user = Table('user', m1, Column('id', Integer, priamry_key=True))
- # load 'users' from the sqlite engine
- users_table = Table('users', meta1, autoload=True,
- autoload_with=some_engine)
+ m2 = MetaData()
+ user_copy = user.tometadata(m2)
- # create the same Table object for the plain metadata
- users_table_2 = users_table.tometadata(meta2)
+ :param metadata: Target :class:`.MetaData` object, into which the
+ new :class:`.Table` object will be created.
- :param metadata: Target :class:`.MetaData` object.
- :param schema: Optional string name of a target schema, or
- ``None`` for no schema. The :class:`.Table` object will be
- given this schema name upon copy. Defaults to the special
- symbol :attr:`.RETAIN_SCHEMA` which indicates no change should be
- made to the schema name of the resulting :class:`.Table`.
+ :param schema: optional string name indicating the target schema.
+ Defaults to the special symbol :attr:`.RETAIN_SCHEMA` which indicates
+ that no change to the schema name should be made in the new
+ :class:`.Table`. If set to a string name, the new :class:`.Table`
+ will have this new name as the ``.schema``. If set to ``None``, the
+ schema will be set to that of the schema set on the target
+ :class:`.MetaData`, which is typically ``None`` as well, unless
+ set explicitly::
+
+ m2 = MetaData(schema='newschema')
+
+ # user_copy_one will have "newschema" as the schema name
+ user_copy_one = user.tometadata(m2, schema=None)
+
+ m3 = MetaData() # schema defaults to None
+
+ # user_copy_two will have None as the schema name
+ user_copy_two = user.tometadata(m3, schema=None)
+
+ :param referred_schema_fn: optional callable which can be supplied
+ in order to provide for the schema name that should be assigned
+ to the referenced table of a :class:`.ForeignKeyConstraint`.
+ The callable accepts this parent :class:`.Table`, the
+ target schema that we are changing to, the :class:`.ForeignKeyConstraint`
+ object, and the existing "target schema" of that constraint. The
+ function should return the string schema name that should be applied.
+ E.g.::
+
+ def referred_schema_fn(table, to_schema,
+ constraint, referred_schema):
+ if referred_schema == 'base_tables':
+ return referred_schema
+ else:
+ return to_schema
+
+ new_table = table.tometadata(m2, schema="alt_schema",
+ referred_schema_fn=referred_schema_fn)
+
+ .. versionadded:: 0.9.2
"""
@@ -707,8 +737,16 @@ class Table(DialectKWArgs, SchemaItem, TableClause):
*args, **self.kwargs
)
for c in self.constraints:
- table.append_constraint(c.copy(schema=schema, target_table=table))
+ if isinstance(c, ForeignKeyConstraint):
+ referred_schema = c._referred_schema
+ if referred_schema_fn:
+ fk_constraint_schema = referred_schema_fn(self, schema, c, referred_schema)
+ else:
+ fk_constraint_schema = schema if referred_schema == self.schema else None
+ table.append_constraint(c.copy(schema=fk_constraint_schema, target_table=table))
+ else:
+ table.append_constraint(c.copy(schema=schema, target_table=table))
for index in self.indexes:
# skip indexes that would be generated
# by the 'index' flag on Column
@@ -1429,6 +1467,10 @@ class ForeignKey(DialectKWArgs, SchemaItem):
else:
return self._colspec
+ @property
+ def _referred_schema(self):
+ return self._column_tokens[0]
+
def _table_key(self):
if self._table_column is not None:
@@ -1465,7 +1507,7 @@ class ForeignKey(DialectKWArgs, SchemaItem):
def _column_tokens(self):
"""parse a string-based _colspec into its component parts."""
- m = self._colspec.split('.')
+ m = self._get_colspec().split('.')
if m is None:
raise exc.ArgumentError(
"Invalid foreign key column specification: %s" %
@@ -2447,6 +2489,13 @@ class ForeignKeyConstraint(Constraint):
columns[0].table is not None:
self._set_parent_with_dispatch(columns[0].table)
+ @property
+ def _referred_schema(self):
+ for elem in self._elements.values():
+ return elem._referred_schema
+ else:
+ return None
+
def _validate_dest_table(self, table):
table_keys = set([elem._table_key() for elem in self._elements.values()])
if None not in table_keys and len(table_keys) > 1:
diff --git a/test/sql/test_metadata.py b/test/sql/test_metadata.py
index 36c777c9a..dd3f4fc3d 100644
--- a/test/sql/test_metadata.py
+++ b/test/sql/test_metadata.py
@@ -306,8 +306,168 @@ class MetaDataTest(fixtures.TestBase, ComparesTables):
)
- @testing.exclude('mysql', '<', (4, 1, 1), 'early types are squirrely')
- def test_to_metadata(self):
+
+ def test_pickle_metadata_sequence_restated(self):
+ m1 = MetaData()
+ Table('a', m1,
+ Column('id', Integer, primary_key=True),
+ Column('x', Integer, Sequence("x_seq")))
+
+ m2 = pickle.loads(pickle.dumps(m1))
+
+ s2 = Sequence("x_seq")
+ t2 = Table('a', m2,
+ Column('id', Integer, primary_key=True),
+ Column('x', Integer, s2),
+ extend_existing=True)
+
+ assert m2._sequences['x_seq'] is t2.c.x.default
+ assert m2._sequences['x_seq'] is s2
+
+
+ def test_sequence_restated_replaced(self):
+ """Test restatement of Sequence replaces."""
+
+ m1 = MetaData()
+ s1 = Sequence("x_seq")
+ t = Table('a', m1,
+ Column('x', Integer, s1)
+ )
+ assert m1._sequences['x_seq'] is s1
+
+ s2 = Sequence('x_seq')
+ Table('a', m1,
+ Column('x', Integer, s2),
+ extend_existing=True
+ )
+ assert t.c.x.default is s2
+ assert m1._sequences['x_seq'] is s2
+
+
+ def test_pickle_metadata_sequence_implicit(self):
+ m1 = MetaData()
+ Table('a', m1,
+ Column('id', Integer, primary_key=True),
+ Column('x', Integer, Sequence("x_seq")))
+
+ m2 = pickle.loads(pickle.dumps(m1))
+
+ t2 = Table('a', m2, extend_existing=True)
+
+ eq_(m2._sequences, {'x_seq': t2.c.x.default})
+
+ def test_pickle_metadata_schema(self):
+ m1 = MetaData()
+ Table('a', m1,
+ Column('id', Integer, primary_key=True),
+ Column('x', Integer, Sequence("x_seq")),
+ schema='y')
+
+ m2 = pickle.loads(pickle.dumps(m1))
+
+ Table('a', m2, schema='y',
+ extend_existing=True)
+
+ eq_(m2._schemas, m1._schemas)
+
+ def test_metadata_schema_arg(self):
+ m1 = MetaData(schema='sch1')
+ m2 = MetaData(schema='sch1', quote_schema=True)
+ m3 = MetaData(schema='sch1', quote_schema=False)
+ m4 = MetaData()
+
+ for i, (name, metadata, schema, quote_schema,
+ exp_schema, exp_quote_schema) in enumerate([
+ ('t1', m1, None, None, 'sch1', None),
+ ('t2', m1, 'sch2', None, 'sch2', None),
+ ('t3', m1, 'sch2', True, 'sch2', True),
+ ('t4', m1, 'sch1', None, 'sch1', None),
+ ('t1', m2, None, None, 'sch1', True),
+ ('t2', m2, 'sch2', None, 'sch2', None),
+ ('t3', m2, 'sch2', True, 'sch2', True),
+ ('t4', m2, 'sch1', None, 'sch1', None),
+ ('t1', m3, None, None, 'sch1', False),
+ ('t2', m3, 'sch2', None, 'sch2', None),
+ ('t3', m3, 'sch2', True, 'sch2', True),
+ ('t4', m3, 'sch1', None, 'sch1', None),
+ ('t1', m4, None, None, None, None),
+ ('t2', m4, 'sch2', None, 'sch2', None),
+ ('t3', m4, 'sch2', True, 'sch2', True),
+ ('t4', m4, 'sch1', None, 'sch1', None),
+ ]):
+ kw = {}
+ if schema is not None:
+ kw['schema'] = schema
+ if quote_schema is not None:
+ kw['quote_schema'] = quote_schema
+ t = Table(name, metadata, **kw)
+ eq_(t.schema, exp_schema, "test %d, table schema" % i)
+ eq_(t.schema.quote if t.schema is not None else None,
+ exp_quote_schema,
+ "test %d, table quote_schema" % i)
+ seq = Sequence(name, metadata=metadata, **kw)
+ eq_(seq.schema, exp_schema, "test %d, seq schema" % i)
+ eq_(seq.schema.quote if seq.schema is not None else None,
+ exp_quote_schema,
+ "test %d, seq quote_schema" % i)
+
+ def test_manual_dependencies(self):
+ meta = MetaData()
+ a = Table('a', meta, Column('foo', Integer))
+ b = Table('b', meta, Column('foo', Integer))
+ c = Table('c', meta, Column('foo', Integer))
+ d = Table('d', meta, Column('foo', Integer))
+ e = Table('e', meta, Column('foo', Integer))
+
+ e.add_is_dependent_on(c)
+ a.add_is_dependent_on(b)
+ b.add_is_dependent_on(d)
+ e.add_is_dependent_on(b)
+ c.add_is_dependent_on(a)
+ eq_(
+ meta.sorted_tables,
+ [d, b, a, c, e]
+ )
+
+ def test_nonexistent(self):
+ assert_raises(tsa.exc.NoSuchTableError, Table,
+ 'fake_table',
+ MetaData(testing.db), autoload=True)
+
+ def test_assorted_repr(self):
+ t1 = Table("foo", MetaData(), Column("x", Integer))
+ i1 = Index("bar", t1.c.x)
+ ck = schema.CheckConstraint("x > y", name="someconstraint")
+
+ for const, exp in (
+ (Sequence("my_seq"),
+ "Sequence('my_seq')"),
+ (Sequence("my_seq", start=5),
+ "Sequence('my_seq', start=5)"),
+ (Column("foo", Integer),
+ "Column('foo', Integer(), table=None)"),
+ (Table("bar", MetaData(), Column("x", String)),
+ "Table('bar', MetaData(bind=None), "
+ "Column('x', String(), table=<bar>), schema=None)"),
+ (schema.DefaultGenerator(for_update=True),
+ "DefaultGenerator(for_update=True)"),
+ (schema.Index("bar", "c"), "Index('bar')"),
+ (i1, "Index('bar', Column('x', Integer(), table=<foo>))"),
+ (schema.FetchedValue(), "FetchedValue()"),
+ (ck,
+ "CheckConstraint("
+ "%s"
+ ", name='someconstraint')" % repr(ck.sqltext)),
+ ):
+ eq_(
+ repr(const),
+ exp
+ )
+
+
+class ToMetaDataTest(fixtures.TestBase, ComparesTables):
+
+ def test_copy(self):
from sqlalchemy.testing.schema import Table
meta = MetaData()
@@ -403,7 +563,7 @@ class MetaDataTest(fixtures.TestBase, ComparesTables):
finally:
meta.drop_all(testing.db)
- def test_col_key_fk_parent_tometadata(self):
+ def test_col_key_fk_parent(self):
# test #2643
m1 = MetaData()
a = Table('a', m1, Column('x', Integer))
@@ -416,70 +576,7 @@ class MetaDataTest(fixtures.TestBase, ComparesTables):
assert b2.c.y.references(a2.c.x)
- def test_pickle_metadata_sequence_restated(self):
- m1 = MetaData()
- Table('a', m1,
- Column('id', Integer, primary_key=True),
- Column('x', Integer, Sequence("x_seq")))
-
- m2 = pickle.loads(pickle.dumps(m1))
-
- s2 = Sequence("x_seq")
- t2 = Table('a', m2,
- Column('id', Integer, primary_key=True),
- Column('x', Integer, s2),
- extend_existing=True)
-
- assert m2._sequences['x_seq'] is t2.c.x.default
- assert m2._sequences['x_seq'] is s2
-
-
- def test_sequence_restated_replaced(self):
- """Test restatement of Sequence replaces."""
-
- m1 = MetaData()
- s1 = Sequence("x_seq")
- t = Table('a', m1,
- Column('x', Integer, s1)
- )
- assert m1._sequences['x_seq'] is s1
-
- s2 = Sequence('x_seq')
- Table('a', m1,
- Column('x', Integer, s2),
- extend_existing=True
- )
- assert t.c.x.default is s2
- assert m1._sequences['x_seq'] is s2
-
-
- def test_pickle_metadata_sequence_implicit(self):
- m1 = MetaData()
- Table('a', m1,
- Column('id', Integer, primary_key=True),
- Column('x', Integer, Sequence("x_seq")))
-
- m2 = pickle.loads(pickle.dumps(m1))
-
- t2 = Table('a', m2, extend_existing=True)
-
- eq_(m2._sequences, {'x_seq': t2.c.x.default})
-
- def test_pickle_metadata_schema(self):
- m1 = MetaData()
- Table('a', m1,
- Column('id', Integer, primary_key=True),
- Column('x', Integer, Sequence("x_seq")),
- schema='y')
-
- m2 = pickle.loads(pickle.dumps(m1))
-
- Table('a', m2, schema='y',
- extend_existing=True)
-
- eq_(m2._schemas, m1._schemas)
-
- def test_tometadata_with_schema(self):
+ def test_change_schema(self):
meta = MetaData()
table = Table('mytable', meta,
@@ -504,7 +601,7 @@ class MetaDataTest(fixtures.TestBase, ComparesTables):
eq_(str(table_c.join(table2_c).onclause),
'someschema.mytable.myid = someschema.othertable.myid')
- def test_tometadata_with_default_schema(self):
+ def test_retain_table_schema(self):
meta = MetaData()
table = Table('mytable', meta,
@@ -531,7 +628,144 @@ class MetaDataTest(fixtures.TestBase, ComparesTables):
eq_(str(table_c.join(table2_c).onclause),
'myschema.mytable.myid = myschema.othertable.myid')
- def test_tometadata_copy_info(self):
+ def _assert_fk(self, t2, schema, expected, referred_schema_fn=None):
+ m2 = MetaData()
+ existing_schema = t2.schema
+ if schema:
+ t2c = t2.tometadata(m2, schema=schema,
+ referred_schema_fn=referred_schema_fn)
+ eq_(t2c.schema, schema)
+ else:
+ t2c = t2.tometadata(m2, referred_schema_fn=referred_schema_fn)
+ eq_(t2c.schema, existing_schema)
+ eq_(list(t2c.c.y.foreign_keys)[0]._get_colspec(), expected)
+
+ def test_fk_has_schema_string_retain_schema(self):
+ m = MetaData()
+
+ t2 = Table('t2', m, Column('y', Integer, ForeignKey('q.t1.x')))
+ self._assert_fk(t2, None, "q.t1.x")
+
+ Table('t1', m, Column('x', Integer), schema='q')
+ self._assert_fk(t2, None, "q.t1.x")
+
+ def test_fk_has_schema_string_new_schema(self):
+ m = MetaData()
+
+ t2 = Table('t2', m, Column('y', Integer, ForeignKey('q.t1.x')))
+ self._assert_fk(t2, "z", "q.t1.x")
+
+ Table('t1', m, Column('x', Integer), schema='q')
+ self._assert_fk(t2, "z", "q.t1.x")
+
+ def test_fk_has_schema_col_retain_schema(self):
+ m = MetaData()
+
+ t1 = Table('t1', m, Column('x', Integer), schema='q')
+ t2 = Table('t2', m, Column('y', Integer, ForeignKey(t1.c.x)))
+
+ self._assert_fk(t2, "z", "q.t1.x")
+
+ def test_fk_has_schema_col_new_schema(self):
+ m = MetaData()
+
+ t1 = Table('t1', m, Column('x', Integer), schema='q')
+ t2 = Table('t2', m, Column('y', Integer, ForeignKey(t1.c.x)))
+
+ self._assert_fk(t2, "z", "q.t1.x")
+
+ def test_fk_and_referent_has_same_schema_string_retain_schema(self):
+ m = MetaData()
+
+ t2 = Table('t2', m, Column('y', Integer,
+ ForeignKey('q.t1.x')), schema="q")
+
+ self._assert_fk(t2, None, "q.t1.x")
+
+ Table('t1', m, Column('x', Integer), schema='q')
+ self._assert_fk(t2, None, "q.t1.x")
+
+ def test_fk_and_referent_has_same_schema_string_new_schema(self):
+ m = MetaData()
+
+ t2 = Table('t2', m, Column('y', Integer,
+ ForeignKey('q.t1.x')), schema="q")
+
+ self._assert_fk(t2, "z", "z.t1.x")
+
+ Table('t1', m, Column('x', Integer), schema='q')
+ self._assert_fk(t2, "z", "z.t1.x")
+
+ def test_fk_and_referent_has_same_schema_col_retain_schema(self):
+ m = MetaData()
+
+ t1 = Table('t1', m, Column('x', Integer), schema='q')
+ t2 = Table('t2', m, Column('y', Integer,
+ ForeignKey(t1.c.x)), schema='q')
+ self._assert_fk(t2, None, "q.t1.x")
+
+
+ def test_fk_and_referent_has_same_schema_col_new_schema(self):
+ m = MetaData()
+
+ t1 = Table('t1', m, Column('x', Integer), schema='q')
+ t2 = Table('t2', m, Column('y', Integer,
+ ForeignKey(t1.c.x)), schema='q')
+ self._assert_fk(t2, 'z', "z.t1.x")
+
+ def test_fk_and_referent_has_diff_schema_string_retain_schema(self):
+ m = MetaData()
+
+ t2 = Table('t2', m, Column('y', Integer,
+ ForeignKey('p.t1.x')), schema="q")
+
+ self._assert_fk(t2, None, "p.t1.x")
+
+ Table('t1', m, Column('x', Integer), schema='p')
+ self._assert_fk(t2, None, "p.t1.x")
+
+ def test_fk_and_referent_has_diff_schema_string_new_schema(self):
+ m = MetaData()
+
+ t2 = Table('t2', m, Column('y', Integer,
+ ForeignKey('p.t1.x')), schema="q")
+
+ self._assert_fk(t2, "z", "p.t1.x")
+
+ Table('t1', m, Column('x', Integer), schema='p')
+ self._assert_fk(t2, "z", "p.t1.x")
+
+ def test_fk_and_referent_has_diff_schema_col_retain_schema(self):
+ m = MetaData()
+
+ t1 = Table('t1', m, Column('x', Integer), schema='p')
+ t2 = Table('t2', m, Column('y', Integer,
+ ForeignKey(t1.c.x)), schema='q')
+ self._assert_fk(t2, None, "p.t1.x")
+
+
+ def test_fk_and_referent_has_diff_schema_col_new_schema(self):
+ m = MetaData()
+
+ t1 = Table('t1', m, Column('x', Integer), schema='p')
+ t2 = Table('t2', m, Column('y', Integer,
+ ForeignKey(t1.c.x)), schema='q')
+ self._assert_fk(t2, 'z', "p.t1.x")
+
+ def test_fk_custom_system(self):
+ m = MetaData()
+ t2 = Table('t2', m, Column('y', Integer,
+ ForeignKey('p.t1.x')), schema='q')
+
+ def ref_fn(table, to_schema, constraint, referred_schema):
+ assert table is t2
+ eq_(to_schema, "z")
+ eq_(referred_schema, "p")
+ return "h"
+ self._assert_fk(t2, 'z', "h.t1.x", referred_schema_fn=ref_fn)
+
+
+ def test_copy_info(self):
m = MetaData()
fk = ForeignKey('t2.id')
c = Column('c', Integer, fk)
@@ -573,7 +807,7 @@ class MetaDataTest(fixtures.TestBase, ComparesTables):
eq_(ck2.info, {"ckinfo": True})
- def test_tometadata_kwargs(self):
+ def test_dialect_kwargs(self):
meta = MetaData()
table = Table('mytable', meta,
@@ -588,7 +822,7 @@ class MetaDataTest(fixtures.TestBase, ComparesTables):
eq_(table.kwargs, table_c.kwargs)
- def test_tometadata_indexes(self):
+ def test_indexes(self):
meta = MetaData()
table = Table('mytable', meta,
@@ -612,7 +846,7 @@ class MetaDataTest(fixtures.TestBase, ComparesTables):
)
@emits_warning("Table '.+' already exists within the given MetaData")
- def test_tometadata_already_there(self):
+ def test_already_exists(self):
meta1 = MetaData()
table1 = Table('mytable', meta1,
@@ -629,66 +863,7 @@ class MetaDataTest(fixtures.TestBase, ComparesTables):
# d'oh!
assert table_c is table_d
- def test_metadata_schema_arg(self):
- m1 = MetaData(schema='sch1')
- m2 = MetaData(schema='sch1', quote_schema=True)
- m3 = MetaData(schema='sch1', quote_schema=False)
- m4 = MetaData()
-
- for i, (name, metadata, schema, quote_schema,
- exp_schema, exp_quote_schema) in enumerate([
- ('t1', m1, None, None, 'sch1', None),
- ('t2', m1, 'sch2', None, 'sch2', None),
- ('t3', m1, 'sch2', True, 'sch2', True),
- ('t4', m1, 'sch1', None, 'sch1', None),
- ('t1', m2, None, None, 'sch1', True),
- ('t2', m2, 'sch2', None, 'sch2', None),
- ('t3', m2, 'sch2', True, 'sch2', True),
- ('t4', m2, 'sch1', None, 'sch1', None),
- ('t1', m3, None, None, 'sch1', False),
- ('t2', m3, 'sch2', None, 'sch2', None),
- ('t3', m3, 'sch2', True, 'sch2', True),
- ('t4', m3, 'sch1', None, 'sch1', None),
- ('t1', m4, None, None, None, None),
- ('t2', m4, 'sch2', None, 'sch2', None),
- ('t3', m4, 'sch2', True, 'sch2', True),
- ('t4', m4, 'sch1', None, 'sch1', None),
- ]):
- kw = {}
- if schema is not None:
- kw['schema'] = schema
- if quote_schema is not None:
- kw['quote_schema'] = quote_schema
- t = Table(name, metadata, **kw)
- eq_(t.schema, exp_schema, "test %d, table schema" % i)
- eq_(t.schema.quote if t.schema is not None else None,
- exp_quote_schema,
- "test %d, table quote_schema" % i)
- seq = Sequence(name, metadata=metadata, **kw)
- eq_(seq.schema, exp_schema, "test %d, seq schema" % i)
- eq_(seq.schema.quote if seq.schema is not None else None,
- exp_quote_schema,
- "test %d, seq quote_schema" % i)
-
- def test_manual_dependencies(self):
- meta = MetaData()
- a = Table('a', meta, Column('foo', Integer))
- b = Table('b', meta, Column('foo', Integer))
- c = Table('c', meta, Column('foo', Integer))
- d = Table('d', meta, Column('foo', Integer))
- e = Table('e', meta, Column('foo', Integer))
-
- e.add_is_dependent_on(c)
- a.add_is_dependent_on(b)
- b.add_is_dependent_on(d)
- e.add_is_dependent_on(b)
- c.add_is_dependent_on(a)
- eq_(
- meta.sorted_tables,
- [d, b, a, c, e]
- )
-
- def test_tometadata_default_schema_metadata(self):
+ def test_default_schema_metadata(self):
meta = MetaData(schema='myschema')
table = Table('mytable', meta,
@@ -712,7 +887,7 @@ class MetaDataTest(fixtures.TestBase, ComparesTables):
eq_(str(table_c.join(table2_c).onclause),
"someschema.mytable.myid = someschema.othertable.myid")
- def test_tometadata_strip_schema(self):
+ def test_strip_schema(self):
meta = MetaData()
table = Table('mytable', meta,
@@ -737,41 +912,6 @@ class MetaDataTest(fixtures.TestBase, ComparesTables):
eq_(str(table_c.join(table2_c).onclause),
'mytable.myid = othertable.myid')
- def test_nonexistent(self):
- assert_raises(tsa.exc.NoSuchTableError, Table,
- 'fake_table',
- MetaData(testing.db), autoload=True)
-
- def test_assorted_repr(self):
- t1 = Table("foo", MetaData(), Column("x", Integer))
- i1 = Index("bar", t1.c.x)
- ck = schema.CheckConstraint("x > y", name="someconstraint")
-
- for const, exp in (
- (Sequence("my_seq"),
- "Sequence('my_seq')"),
- (Sequence("my_seq", start=5),
- "Sequence('my_seq', start=5)"),
- (Column("foo", Integer),
- "Column('foo', Integer(), table=None)"),
- (Table("bar", MetaData(), Column("x", String)),
- "Table('bar', MetaData(bind=None), "
- "Column('x', String(), table=<bar>), schema=None)"),
- (schema.DefaultGenerator(for_update=True),
- "DefaultGenerator(for_update=True)"),
- (schema.Index("bar", "c"), "Index('bar')"),
- (i1, "Index('bar', Column('x', Integer(), table=<foo>))"),
- (schema.FetchedValue(), "FetchedValue()"),
- (ck,
- "CheckConstraint("
- "%s"
- ", name='someconstraint')" % repr(ck.sqltext)),
- ):
- eq_(
- repr(const),
- exp
- )
-
class TableTest(fixtures.TestBase, AssertsCompiledSQL):
@testing.skip_if('mssql', 'different col format')
def test_prefixes(self):