diff options
Diffstat (limited to 'lib/sqlalchemy/test/schema.py')
-rw-r--r-- | lib/sqlalchemy/test/schema.py | 79 |
1 files changed, 0 insertions, 79 deletions
diff --git a/lib/sqlalchemy/test/schema.py b/lib/sqlalchemy/test/schema.py deleted file mode 100644 index d33d75e2c..000000000 --- a/lib/sqlalchemy/test/schema.py +++ /dev/null @@ -1,79 +0,0 @@ -"""Enhanced versions of schema.Table and schema.Column which establish -desired state for different backends. -""" - -from sqlalchemy.test import testing -from sqlalchemy import schema - -__all__ = 'Table', 'Column', - -table_options = {} - -def Table(*args, **kw): - """A schema.Table wrapper/hook for dialect-specific tweaks.""" - - test_opts = dict([(k,kw.pop(k)) for k in kw.keys() - if k.startswith('test_')]) - - kw.update(table_options) - - if testing.against('mysql'): - if 'mysql_engine' not in kw and 'mysql_type' not in kw: - if 'test_needs_fk' in test_opts or 'test_needs_acid' in test_opts: - kw['mysql_engine'] = 'InnoDB' - - # Apply some default cascading rules for self-referential foreign keys. - # MySQL InnoDB has some issues around seleting self-refs too. - if testing.against('firebird'): - table_name = args[0] - unpack = (testing.config.db.dialect. - identifier_preparer.unformat_identifiers) - - # Only going after ForeignKeys in Columns. May need to - # expand to ForeignKeyConstraint too. - fks = [fk - for col in args if isinstance(col, schema.Column) - for fk in col.foreign_keys] - - for fk in fks: - # root around in raw spec - ref = fk._colspec - if isinstance(ref, schema.Column): - name = ref.table.name - else: - # take just the table name: on FB there cannot be - # a schema, so the first element is always the - # table name, possibly followed by the field name - name = unpack(ref)[0] - if name == table_name: - if fk.ondelete is None: - fk.ondelete = 'CASCADE' - if fk.onupdate is None: - fk.onupdate = 'CASCADE' - - return schema.Table(*args, **kw) - - -def Column(*args, **kw): - """A schema.Column wrapper/hook for dialect-specific tweaks.""" - - test_opts = dict([(k,kw.pop(k)) for k in kw.keys() - if k.startswith('test_')]) - - col = schema.Column(*args, **kw) - if 'test_needs_autoincrement' in test_opts and \ - kw.get('primary_key', False) and \ - testing.against('firebird', 'oracle'): - def add_seq(tbl, c): - c._init_items( - schema.Sequence(_truncate_name(testing.db.dialect, tbl.name + '_' + c.name + '_seq'), optional=True) - ) - col._on_table_attach(add_seq) - return col - -def _truncate_name(dialect, name): - if len(name) > dialect.max_identifier_length: - return name[0:max(dialect.max_identifier_length - 6, 0)] + "_" + hex(hash(name) % 64)[2:] - else: - return name - |