1321 lines
46 KiB
Python
1321 lines
46 KiB
Python
import re
|
|
import sys
|
|
from alembic.testing import TestBase, exclusions
|
|
|
|
from sqlalchemy import MetaData, Column, Table, String, \
|
|
Numeric, CHAR, ForeignKey, DATETIME, Integer, \
|
|
CheckConstraint, Unicode, Enum, cast,\
|
|
UniqueConstraint, Boolean, ForeignKeyConstraint,\
|
|
PrimaryKeyConstraint, Index, func, text, DefaultClause
|
|
|
|
from sqlalchemy.types import TIMESTAMP
|
|
from sqlalchemy.types import UserDefinedType
|
|
from sqlalchemy.dialects import mysql, postgresql
|
|
from sqlalchemy.engine.default import DefaultDialect
|
|
from sqlalchemy.sql import and_, column, literal_column, false
|
|
|
|
from alembic.testing.mock import patch
|
|
|
|
from alembic import autogenerate, util, compat
|
|
from alembic.testing import eq_, eq_ignore_whitespace, config
|
|
|
|
from alembic.testing.fixtures import op_fixture
|
|
from alembic import op # noqa
|
|
import sqlalchemy as sa # noqa
|
|
|
|
py3k = sys.version_info >= (3, )
|
|
|
|
|
|
class AutogenRenderTest(TestBase):
|
|
|
|
"""test individual directives"""
|
|
|
|
@classmethod
|
|
def setup_class(cls):
|
|
cls.autogen_context = {
|
|
'opts': {
|
|
'sqlalchemy_module_prefix': 'sa.',
|
|
'alembic_module_prefix': 'op.',
|
|
},
|
|
'dialect': mysql.dialect()
|
|
}
|
|
cls.pg_autogen_context = {
|
|
'opts': {
|
|
'sqlalchemy_module_prefix': 'sa.',
|
|
'alembic_module_prefix': 'op.',
|
|
},
|
|
'dialect': postgresql.dialect()
|
|
}
|
|
|
|
def test_render_add_index(self):
|
|
"""
|
|
autogenerate.render._add_index
|
|
"""
|
|
m = MetaData()
|
|
t = Table('test', m,
|
|
Column('id', Integer, primary_key=True),
|
|
Column('active', Boolean()),
|
|
Column('code', String(255)),
|
|
)
|
|
idx = Index('test_active_code_idx', t.c.active, t.c.code)
|
|
eq_ignore_whitespace(
|
|
autogenerate.render._add_index(idx, self.autogen_context),
|
|
"op.create_index('test_active_code_idx', 'test', "
|
|
"['active', 'code'], unique=False)"
|
|
)
|
|
|
|
def test_render_add_index_schema(self):
|
|
"""
|
|
autogenerate.render._add_index using schema
|
|
"""
|
|
m = MetaData()
|
|
t = Table('test', m,
|
|
Column('id', Integer, primary_key=True),
|
|
Column('active', Boolean()),
|
|
Column('code', String(255)),
|
|
schema='CamelSchema'
|
|
)
|
|
idx = Index('test_active_code_idx', t.c.active, t.c.code)
|
|
eq_ignore_whitespace(
|
|
autogenerate.render._add_index(idx, self.autogen_context),
|
|
"op.create_index('test_active_code_idx', 'test', "
|
|
"['active', 'code'], unique=False, schema='CamelSchema')"
|
|
)
|
|
|
|
def test_render_add_index_pg_where(self):
|
|
autogen_context = self.pg_autogen_context
|
|
|
|
m = MetaData()
|
|
t = Table('t', m,
|
|
Column('x', String),
|
|
Column('y', String)
|
|
)
|
|
|
|
idx = Index('foo_idx', t.c.x, t.c.y,
|
|
postgresql_where=(t.c.y == 'something'))
|
|
|
|
if compat.sqla_08:
|
|
eq_ignore_whitespace(
|
|
autogenerate.render._add_index(idx, autogen_context),
|
|
"""op.create_index('foo_idx', 't', \
|
|
['x', 'y'], unique=False, """
|
|
"""postgresql_where=sa.text(!U"t.y = 'something'"))"""
|
|
)
|
|
else:
|
|
eq_ignore_whitespace(
|
|
autogenerate.render._add_index(idx, autogen_context),
|
|
"""op.create_index('foo_idx', 't', ['x', 'y'], \
|
|
unique=False, """
|
|
"""postgresql_where=sa.text(!U't.y = %(y_1)s'))"""
|
|
)
|
|
|
|
@config.requirements.fail_before_sqla_080
|
|
def test_render_add_index_func(self):
|
|
m = MetaData()
|
|
t = Table(
|
|
'test', m,
|
|
Column('id', Integer, primary_key=True),
|
|
Column('code', String(255))
|
|
)
|
|
idx = Index('test_lower_code_idx', func.lower(t.c.code))
|
|
eq_ignore_whitespace(
|
|
autogenerate.render._add_index(idx, self.autogen_context),
|
|
"op.create_index('test_lower_code_idx', 'test', "
|
|
"[sa.text(!U'lower(test.code)')], unique=False)"
|
|
)
|
|
|
|
@config.requirements.fail_before_sqla_080
|
|
def test_render_add_index_cast(self):
|
|
m = MetaData()
|
|
t = Table(
|
|
'test', m,
|
|
Column('id', Integer, primary_key=True),
|
|
Column('code', String(255))
|
|
)
|
|
idx = Index('test_lower_code_idx', cast(t.c.code, String))
|
|
eq_ignore_whitespace(
|
|
autogenerate.render._add_index(idx, self.autogen_context),
|
|
"op.create_index('test_lower_code_idx', 'test', "
|
|
"[sa.text(!U'CAST(test.code AS CHAR)')], unique=False)"
|
|
)
|
|
|
|
@config.requirements.fail_before_sqla_080
|
|
def test_render_add_index_desc(self):
|
|
m = MetaData()
|
|
t = Table(
|
|
'test', m,
|
|
Column('id', Integer, primary_key=True),
|
|
Column('code', String(255))
|
|
)
|
|
idx = Index('test_desc_code_idx', t.c.code.desc())
|
|
eq_ignore_whitespace(
|
|
autogenerate.render._add_index(idx, self.autogen_context),
|
|
"op.create_index('test_desc_code_idx', 'test', "
|
|
"[sa.text(!U'test.code DESC')], unique=False)"
|
|
)
|
|
|
|
def test_drop_index(self):
|
|
"""
|
|
autogenerate.render._drop_index
|
|
"""
|
|
m = MetaData()
|
|
t = Table('test', m,
|
|
Column('id', Integer, primary_key=True),
|
|
Column('active', Boolean()),
|
|
Column('code', String(255)),
|
|
)
|
|
idx = Index('test_active_code_idx', t.c.active, t.c.code)
|
|
eq_ignore_whitespace(
|
|
autogenerate.render._drop_index(idx, self.autogen_context),
|
|
"op.drop_index('test_active_code_idx', table_name='test')"
|
|
)
|
|
|
|
def test_drop_index_schema(self):
|
|
"""
|
|
autogenerate.render._drop_index using schema
|
|
"""
|
|
m = MetaData()
|
|
t = Table('test', m,
|
|
Column('id', Integer, primary_key=True),
|
|
Column('active', Boolean()),
|
|
Column('code', String(255)),
|
|
schema='CamelSchema'
|
|
)
|
|
idx = Index('test_active_code_idx', t.c.active, t.c.code)
|
|
eq_ignore_whitespace(
|
|
autogenerate.render._drop_index(idx, self.autogen_context),
|
|
"op.drop_index('test_active_code_idx', " +
|
|
"table_name='test', schema='CamelSchema')"
|
|
)
|
|
|
|
def test_add_unique_constraint(self):
|
|
"""
|
|
autogenerate.render._add_unique_constraint
|
|
"""
|
|
m = MetaData()
|
|
t = Table('test', m,
|
|
Column('id', Integer, primary_key=True),
|
|
Column('active', Boolean()),
|
|
Column('code', String(255)),
|
|
)
|
|
uq = UniqueConstraint(t.c.code, name='uq_test_code')
|
|
eq_ignore_whitespace(
|
|
autogenerate.render._add_unique_constraint(
|
|
uq, self.autogen_context),
|
|
"op.create_unique_constraint('uq_test_code', 'test', ['code'])"
|
|
)
|
|
|
|
def test_add_unique_constraint_schema(self):
|
|
"""
|
|
autogenerate.render._add_unique_constraint using schema
|
|
"""
|
|
m = MetaData()
|
|
t = Table('test', m,
|
|
Column('id', Integer, primary_key=True),
|
|
Column('active', Boolean()),
|
|
Column('code', String(255)),
|
|
schema='CamelSchema'
|
|
)
|
|
uq = UniqueConstraint(t.c.code, name='uq_test_code')
|
|
eq_ignore_whitespace(
|
|
autogenerate.render._add_unique_constraint(
|
|
uq, self.autogen_context),
|
|
"op.create_unique_constraint('uq_test_code', 'test', "
|
|
"['code'], schema='CamelSchema')"
|
|
)
|
|
|
|
def test_drop_unique_constraint(self):
|
|
"""
|
|
autogenerate.render._drop_constraint
|
|
"""
|
|
m = MetaData()
|
|
t = Table('test', m,
|
|
Column('id', Integer, primary_key=True),
|
|
Column('active', Boolean()),
|
|
Column('code', String(255)),
|
|
)
|
|
uq = UniqueConstraint(t.c.code, name='uq_test_code')
|
|
eq_ignore_whitespace(
|
|
autogenerate.render._drop_constraint(uq, self.autogen_context),
|
|
"op.drop_constraint('uq_test_code', 'test', type_='unique')"
|
|
)
|
|
|
|
def test_drop_unique_constraint_schema(self):
|
|
"""
|
|
autogenerate.render._drop_constraint using schema
|
|
"""
|
|
m = MetaData()
|
|
t = Table('test', m,
|
|
Column('id', Integer, primary_key=True),
|
|
Column('active', Boolean()),
|
|
Column('code', String(255)),
|
|
schema='CamelSchema'
|
|
)
|
|
uq = UniqueConstraint(t.c.code, name='uq_test_code')
|
|
eq_ignore_whitespace(
|
|
autogenerate.render._drop_constraint(uq, self.autogen_context),
|
|
"op.drop_constraint('uq_test_code', 'test', "
|
|
"schema='CamelSchema', type_='unique')"
|
|
)
|
|
|
|
def test_add_fk_constraint(self):
|
|
m = MetaData()
|
|
Table('a', m, Column('id', Integer, primary_key=True))
|
|
b = Table('b', m, Column('a_id', Integer, ForeignKey('a.id')))
|
|
fk = ForeignKeyConstraint(['a_id'], ['a.id'], name='fk_a_id')
|
|
b.append_constraint(fk)
|
|
eq_ignore_whitespace(
|
|
autogenerate.render._add_fk_constraint(fk, self.autogen_context),
|
|
"op.create_foreign_key('fk_a_id', 'b', 'a', ['a_id'], ['id'])"
|
|
)
|
|
|
|
def test_add_fk_constraint_inline_colkeys(self):
|
|
m = MetaData()
|
|
Table('a', m, Column('id', Integer, key='aid', primary_key=True))
|
|
b = Table(
|
|
'b', m,
|
|
Column('a_id', Integer, ForeignKey('a.aid'), key='baid'))
|
|
|
|
py_code = autogenerate.render._add_table(b, self.autogen_context)
|
|
|
|
eq_ignore_whitespace(
|
|
py_code,
|
|
"op.create_table('b',"
|
|
"sa.Column('a_id', sa.Integer(), nullable=True),"
|
|
"sa.ForeignKeyConstraint(['a_id'], ['a.id'], ))"
|
|
)
|
|
|
|
context = op_fixture()
|
|
eval(py_code)
|
|
context.assert_(
|
|
"CREATE TABLE b (a_id INTEGER, "
|
|
"FOREIGN KEY(a_id) REFERENCES a (id))")
|
|
|
|
def test_add_fk_constraint_separate_colkeys(self):
|
|
m = MetaData()
|
|
Table('a', m, Column('id', Integer, key='aid', primary_key=True))
|
|
b = Table('b', m, Column('a_id', Integer, key='baid'))
|
|
fk = ForeignKeyConstraint(['baid'], ['a.aid'], name='fk_a_id')
|
|
b.append_constraint(fk)
|
|
|
|
py_code = autogenerate.render._add_table(b, self.autogen_context)
|
|
|
|
eq_ignore_whitespace(
|
|
py_code,
|
|
"op.create_table('b',"
|
|
"sa.Column('a_id', sa.Integer(), nullable=True),"
|
|
"sa.ForeignKeyConstraint(['a_id'], ['a.id'], name='fk_a_id'))"
|
|
)
|
|
|
|
context = op_fixture()
|
|
eval(py_code)
|
|
context.assert_(
|
|
"CREATE TABLE b (a_id INTEGER, CONSTRAINT "
|
|
"fk_a_id FOREIGN KEY(a_id) REFERENCES a (id))")
|
|
|
|
context = op_fixture()
|
|
py_code = autogenerate.render._add_fk_constraint(
|
|
fk, self.autogen_context)
|
|
|
|
eq_ignore_whitespace(
|
|
autogenerate.render._add_fk_constraint(fk, self.autogen_context),
|
|
"op.create_foreign_key('fk_a_id', 'b', 'a', ['a_id'], ['id'])"
|
|
)
|
|
|
|
eval(py_code)
|
|
context.assert_(
|
|
"ALTER TABLE b ADD CONSTRAINT fk_a_id "
|
|
"FOREIGN KEY(a_id) REFERENCES a (id)")
|
|
|
|
def test_add_fk_constraint_schema(self):
|
|
m = MetaData()
|
|
Table(
|
|
'a', m, Column('id', Integer, primary_key=True),
|
|
schema="CamelSchemaTwo")
|
|
b = Table(
|
|
'b', m, Column('a_id', Integer, ForeignKey('a.id')),
|
|
schema="CamelSchemaOne")
|
|
fk = ForeignKeyConstraint(
|
|
["a_id"],
|
|
["CamelSchemaTwo.a.id"], name='fk_a_id')
|
|
b.append_constraint(fk)
|
|
eq_ignore_whitespace(
|
|
autogenerate.render._add_fk_constraint(fk, self.autogen_context),
|
|
"op.create_foreign_key('fk_a_id', 'b', 'a', ['a_id'], ['id'],"
|
|
" source_schema='CamelSchemaOne', "
|
|
"referent_schema='CamelSchemaTwo')"
|
|
)
|
|
|
|
def test_drop_fk_constraint(self):
|
|
m = MetaData()
|
|
Table('a', m, Column('id', Integer, primary_key=True))
|
|
b = Table('b', m, Column('a_id', Integer, ForeignKey('a.id')))
|
|
fk = ForeignKeyConstraint(['a_id'], ['a.id'], name='fk_a_id')
|
|
b.append_constraint(fk)
|
|
eq_ignore_whitespace(
|
|
autogenerate.render._drop_constraint(fk, self.autogen_context),
|
|
"op.drop_constraint('fk_a_id', 'b', type_='foreignkey')"
|
|
)
|
|
|
|
def test_drop_fk_constraint_schema(self):
|
|
m = MetaData()
|
|
Table(
|
|
'a', m, Column('id', Integer, primary_key=True),
|
|
schema="CamelSchemaTwo")
|
|
b = Table(
|
|
'b', m, Column('a_id', Integer, ForeignKey('a.id')),
|
|
schema="CamelSchemaOne")
|
|
fk = ForeignKeyConstraint(
|
|
["a_id"],
|
|
["CamelSchemaTwo.a.id"], name='fk_a_id')
|
|
b.append_constraint(fk)
|
|
|
|
eq_ignore_whitespace(
|
|
autogenerate.render._drop_constraint(fk, self.autogen_context),
|
|
"op.drop_constraint('fk_a_id', 'b', schema='CamelSchemaOne', "
|
|
"type_='foreignkey')"
|
|
)
|
|
|
|
def test_render_table_upgrade(self):
|
|
m = MetaData()
|
|
t = Table('test', m,
|
|
Column('id', Integer, primary_key=True),
|
|
Column('name', Unicode(255)),
|
|
Column("address_id", Integer, ForeignKey("address.id")),
|
|
Column("timestamp", DATETIME, server_default="NOW()"),
|
|
Column("amount", Numeric(5, 2)),
|
|
UniqueConstraint("name", name="uq_name"),
|
|
UniqueConstraint("timestamp"),
|
|
)
|
|
eq_ignore_whitespace(
|
|
autogenerate.render._add_table(t, self.autogen_context),
|
|
"op.create_table('test',"
|
|
"sa.Column('id', sa.Integer(), nullable=False),"
|
|
"sa.Column('name', sa.Unicode(length=255), nullable=True),"
|
|
"sa.Column('address_id', sa.Integer(), nullable=True),"
|
|
"sa.Column('timestamp', sa.DATETIME(), "
|
|
"server_default='NOW()', "
|
|
"nullable=True),"
|
|
"sa.Column('amount', sa.Numeric(precision=5, scale=2), "
|
|
"nullable=True),"
|
|
"sa.ForeignKeyConstraint(['address_id'], ['address.id'], ),"
|
|
"sa.PrimaryKeyConstraint('id'),"
|
|
"sa.UniqueConstraint('name', name='uq_name'),"
|
|
"sa.UniqueConstraint('timestamp')"
|
|
")"
|
|
)
|
|
|
|
def test_render_table_w_schema(self):
|
|
m = MetaData()
|
|
t = Table('test', m,
|
|
Column('id', Integer, primary_key=True),
|
|
Column('q', Integer, ForeignKey('address.id')),
|
|
schema='foo'
|
|
)
|
|
eq_ignore_whitespace(
|
|
autogenerate.render._add_table(t, self.autogen_context),
|
|
"op.create_table('test',"
|
|
"sa.Column('id', sa.Integer(), nullable=False),"
|
|
"sa.Column('q', sa.Integer(), nullable=True),"
|
|
"sa.ForeignKeyConstraint(['q'], ['address.id'], ),"
|
|
"sa.PrimaryKeyConstraint('id'),"
|
|
"schema='foo'"
|
|
")"
|
|
)
|
|
|
|
def test_render_table_w_unicode_name(self):
|
|
m = MetaData()
|
|
t = Table(compat.ue('\u0411\u0435\u0437'), m,
|
|
Column('id', Integer, primary_key=True),
|
|
)
|
|
eq_ignore_whitespace(
|
|
autogenerate.render._add_table(t, self.autogen_context),
|
|
"op.create_table(%r,"
|
|
"sa.Column('id', sa.Integer(), nullable=False),"
|
|
"sa.PrimaryKeyConstraint('id'))" % compat.ue('\u0411\u0435\u0437')
|
|
)
|
|
|
|
def test_render_table_w_unicode_schema(self):
|
|
m = MetaData()
|
|
t = Table('test', m,
|
|
Column('id', Integer, primary_key=True),
|
|
schema=compat.ue('\u0411\u0435\u0437')
|
|
)
|
|
eq_ignore_whitespace(
|
|
autogenerate.render._add_table(t, self.autogen_context),
|
|
"op.create_table('test',"
|
|
"sa.Column('id', sa.Integer(), nullable=False),"
|
|
"sa.PrimaryKeyConstraint('id'),"
|
|
"schema=%r)" % compat.ue('\u0411\u0435\u0437')
|
|
)
|
|
|
|
@patch("alembic.autogenerate.render.MAX_PYTHON_ARGS", 3)
|
|
def test_render_table_max_cols(self):
|
|
m = MetaData()
|
|
t = Table(
|
|
'test', m,
|
|
Column('a', Integer),
|
|
Column('b', Integer),
|
|
Column('c', Integer),
|
|
Column('d', Integer),
|
|
)
|
|
eq_ignore_whitespace(
|
|
autogenerate.render._add_table(t, self.autogen_context),
|
|
"op.create_table('test',"
|
|
"*[sa.Column('a', sa.Integer(), nullable=True),"
|
|
"sa.Column('b', sa.Integer(), nullable=True),"
|
|
"sa.Column('c', sa.Integer(), nullable=True),"
|
|
"sa.Column('d', sa.Integer(), nullable=True)])"
|
|
)
|
|
|
|
t2 = Table(
|
|
'test2', m,
|
|
Column('a', Integer),
|
|
Column('b', Integer),
|
|
Column('c', Integer),
|
|
)
|
|
|
|
eq_ignore_whitespace(
|
|
autogenerate.render._add_table(t2, self.autogen_context),
|
|
"op.create_table('test2',"
|
|
"sa.Column('a', sa.Integer(), nullable=True),"
|
|
"sa.Column('b', sa.Integer(), nullable=True),"
|
|
"sa.Column('c', sa.Integer(), nullable=True))"
|
|
)
|
|
|
|
def test_render_table_w_fk_schema(self):
|
|
m = MetaData()
|
|
t = Table('test', m,
|
|
Column('id', Integer, primary_key=True),
|
|
Column('q', Integer, ForeignKey('foo.address.id')),
|
|
)
|
|
eq_ignore_whitespace(
|
|
autogenerate.render._add_table(t, self.autogen_context),
|
|
"op.create_table('test',"
|
|
"sa.Column('id', sa.Integer(), nullable=False),"
|
|
"sa.Column('q', sa.Integer(), nullable=True),"
|
|
"sa.ForeignKeyConstraint(['q'], ['foo.address.id'], ),"
|
|
"sa.PrimaryKeyConstraint('id')"
|
|
")"
|
|
)
|
|
|
|
def test_render_table_w_metadata_schema(self):
|
|
m = MetaData(schema="foo")
|
|
t = Table('test', m,
|
|
Column('id', Integer, primary_key=True),
|
|
Column('q', Integer, ForeignKey('address.id')),
|
|
)
|
|
eq_ignore_whitespace(
|
|
re.sub(
|
|
r"u'", "'",
|
|
autogenerate.render._add_table(t, self.autogen_context)
|
|
),
|
|
"op.create_table('test',"
|
|
"sa.Column('id', sa.Integer(), nullable=False),"
|
|
"sa.Column('q', sa.Integer(), nullable=True),"
|
|
"sa.ForeignKeyConstraint(['q'], ['foo.address.id'], ),"
|
|
"sa.PrimaryKeyConstraint('id'),"
|
|
"schema='foo'"
|
|
")"
|
|
)
|
|
|
|
def test_render_table_w_metadata_schema_override(self):
|
|
m = MetaData(schema="foo")
|
|
t = Table('test', m,
|
|
Column('id', Integer, primary_key=True),
|
|
Column('q', Integer, ForeignKey('bar.address.id')),
|
|
)
|
|
eq_ignore_whitespace(
|
|
autogenerate.render._add_table(t, self.autogen_context),
|
|
"op.create_table('test',"
|
|
"sa.Column('id', sa.Integer(), nullable=False),"
|
|
"sa.Column('q', sa.Integer(), nullable=True),"
|
|
"sa.ForeignKeyConstraint(['q'], ['bar.address.id'], ),"
|
|
"sa.PrimaryKeyConstraint('id'),"
|
|
"schema='foo'"
|
|
")"
|
|
)
|
|
|
|
def test_render_addtl_args(self):
|
|
m = MetaData()
|
|
t = Table('test', m,
|
|
Column('id', Integer, primary_key=True),
|
|
Column('q', Integer, ForeignKey('bar.address.id')),
|
|
sqlite_autoincrement=True, mysql_engine="InnoDB"
|
|
)
|
|
eq_ignore_whitespace(
|
|
autogenerate.render._add_table(t, self.autogen_context),
|
|
"op.create_table('test',"
|
|
"sa.Column('id', sa.Integer(), nullable=False),"
|
|
"sa.Column('q', sa.Integer(), nullable=True),"
|
|
"sa.ForeignKeyConstraint(['q'], ['bar.address.id'], ),"
|
|
"sa.PrimaryKeyConstraint('id'),"
|
|
"mysql_engine='InnoDB',sqlite_autoincrement=True)"
|
|
)
|
|
|
|
def test_render_drop_table(self):
|
|
eq_ignore_whitespace(
|
|
autogenerate.render._drop_table(Table("sometable", MetaData()),
|
|
self.autogen_context),
|
|
"op.drop_table('sometable')"
|
|
)
|
|
|
|
def test_render_drop_table_w_schema(self):
|
|
eq_ignore_whitespace(
|
|
autogenerate.render._drop_table(
|
|
Table("sometable", MetaData(), schema='foo'),
|
|
self.autogen_context),
|
|
"op.drop_table('sometable', schema='foo')"
|
|
)
|
|
|
|
def test_render_table_no_implicit_check(self):
|
|
m = MetaData()
|
|
t = Table('test', m, Column('x', Boolean()))
|
|
|
|
eq_ignore_whitespace(
|
|
autogenerate.render._add_table(t, self.autogen_context),
|
|
"op.create_table('test',"
|
|
"sa.Column('x', sa.Boolean(), nullable=True))"
|
|
)
|
|
|
|
def test_render_empty_pk_vs_nonempty_pk(self):
|
|
m = MetaData()
|
|
t1 = Table('t1', m, Column('x', Integer))
|
|
t2 = Table('t2', m, Column('x', Integer, primary_key=True))
|
|
|
|
eq_ignore_whitespace(
|
|
autogenerate.render._add_table(t1, self.autogen_context),
|
|
"op.create_table('t1',"
|
|
"sa.Column('x', sa.Integer(), nullable=True))"
|
|
)
|
|
|
|
eq_ignore_whitespace(
|
|
autogenerate.render._add_table(t2, self.autogen_context),
|
|
"op.create_table('t2',"
|
|
"sa.Column('x', sa.Integer(), nullable=False),"
|
|
"sa.PrimaryKeyConstraint('x'))"
|
|
)
|
|
|
|
def test_render_add_column(self):
|
|
eq_ignore_whitespace(
|
|
autogenerate.render._add_column(
|
|
None, "foo", Column("x", Integer, server_default="5"),
|
|
self.autogen_context),
|
|
"op.add_column('foo', sa.Column('x', sa.Integer(), "
|
|
"server_default='5', nullable=True))"
|
|
)
|
|
|
|
def test_render_add_column_w_schema(self):
|
|
eq_ignore_whitespace(
|
|
autogenerate.render._add_column(
|
|
"foo", "bar", Column("x", Integer, server_default="5"),
|
|
self.autogen_context),
|
|
"op.add_column('bar', sa.Column('x', sa.Integer(), "
|
|
"server_default='5', nullable=True), schema='foo')"
|
|
)
|
|
|
|
def test_render_drop_column(self):
|
|
eq_ignore_whitespace(
|
|
autogenerate.render._drop_column(
|
|
None, "foo", Column("x", Integer, server_default="5"),
|
|
self.autogen_context),
|
|
|
|
"op.drop_column('foo', 'x')"
|
|
)
|
|
|
|
def test_render_drop_column_w_schema(self):
|
|
eq_ignore_whitespace(
|
|
autogenerate.render._drop_column(
|
|
"foo", "bar", Column("x", Integer, server_default="5"),
|
|
self.autogen_context),
|
|
|
|
"op.drop_column('bar', 'x', schema='foo')"
|
|
)
|
|
|
|
def test_render_quoted_server_default(self):
|
|
eq_(
|
|
autogenerate.render._render_server_default(
|
|
"nextval('group_to_perm_group_to_perm_id_seq'::regclass)",
|
|
self.autogen_context),
|
|
'"nextval(\'group_to_perm_group_to_perm_id_seq\'::regclass)"'
|
|
)
|
|
|
|
def test_render_unicode_server_default(self):
|
|
default = compat.ue(
|
|
'\u0411\u0435\u0437 '
|
|
'\u043d\u0430\u0437\u0432\u0430\u043d\u0438\u044f'
|
|
)
|
|
|
|
c = Column(
|
|
'x', Unicode,
|
|
server_default=text(default)
|
|
)
|
|
|
|
eq_ignore_whitespace(
|
|
autogenerate.render._render_server_default(
|
|
c.server_default, self.autogen_context
|
|
),
|
|
"sa.text(%r)" % default
|
|
)
|
|
|
|
def test_render_col_with_server_default(self):
|
|
c = Column('updated_at', TIMESTAMP(),
|
|
server_default='TIMEZONE("utc", CURRENT_TIMESTAMP)',
|
|
nullable=False)
|
|
result = autogenerate.render._render_column(
|
|
c, self.autogen_context
|
|
)
|
|
eq_ignore_whitespace(
|
|
result,
|
|
'sa.Column(\'updated_at\', sa.TIMESTAMP(), '
|
|
'server_default=\'TIMEZONE("utc", CURRENT_TIMESTAMP)\', '
|
|
'nullable=False)'
|
|
)
|
|
|
|
def test_render_col_autoinc_false_mysql(self):
|
|
c = Column('some_key', Integer, primary_key=True, autoincrement=False)
|
|
Table('some_table', MetaData(), c)
|
|
result = autogenerate.render._render_column(
|
|
c, self.autogen_context
|
|
)
|
|
eq_ignore_whitespace(
|
|
result,
|
|
'sa.Column(\'some_key\', sa.Integer(), '
|
|
'autoincrement=False, '
|
|
'nullable=False)'
|
|
)
|
|
|
|
def test_render_custom(self):
|
|
|
|
def render(type_, obj, context):
|
|
if type_ == "foreign_key":
|
|
return None
|
|
if type_ == "column":
|
|
if obj.name == "y":
|
|
return None
|
|
else:
|
|
return "col(%s)" % obj.name
|
|
return "render:%s" % type_
|
|
|
|
autogen_context = {"opts": {
|
|
'render_item': render,
|
|
'alembic_module_prefix': 'sa.'
|
|
}}
|
|
|
|
t = Table('t', MetaData(),
|
|
Column('x', Integer),
|
|
Column('y', Integer),
|
|
PrimaryKeyConstraint('x'),
|
|
ForeignKeyConstraint(['x'], ['y'])
|
|
)
|
|
result = autogenerate.render._add_table(
|
|
t, autogen_context
|
|
)
|
|
eq_ignore_whitespace(
|
|
result,
|
|
"sa.create_table('t',"
|
|
"col(x),"
|
|
"render:primary_key)"
|
|
)
|
|
|
|
def test_render_modify_type(self):
|
|
eq_ignore_whitespace(
|
|
autogenerate.render._modify_col(
|
|
"sometable", "somecolumn",
|
|
self.autogen_context,
|
|
type_=CHAR(10), existing_type=CHAR(20)),
|
|
"op.alter_column('sometable', 'somecolumn', "
|
|
"existing_type=sa.CHAR(length=20), type_=sa.CHAR(length=10))"
|
|
)
|
|
|
|
def test_render_modify_type_w_schema(self):
|
|
eq_ignore_whitespace(
|
|
autogenerate.render._modify_col(
|
|
"sometable", "somecolumn",
|
|
self.autogen_context,
|
|
type_=CHAR(10), existing_type=CHAR(20),
|
|
schema='foo'),
|
|
"op.alter_column('sometable', 'somecolumn', "
|
|
"existing_type=sa.CHAR(length=20), type_=sa.CHAR(length=10), "
|
|
"schema='foo')"
|
|
)
|
|
|
|
def test_render_modify_nullable(self):
|
|
eq_ignore_whitespace(
|
|
autogenerate.render._modify_col(
|
|
"sometable", "somecolumn",
|
|
self.autogen_context,
|
|
existing_type=Integer(),
|
|
nullable=True),
|
|
"op.alter_column('sometable', 'somecolumn', "
|
|
"existing_type=sa.Integer(), nullable=True)"
|
|
)
|
|
|
|
def test_render_modify_nullable_w_schema(self):
|
|
eq_ignore_whitespace(
|
|
autogenerate.render._modify_col(
|
|
"sometable", "somecolumn",
|
|
self.autogen_context,
|
|
existing_type=Integer(),
|
|
nullable=True, schema='foo'),
|
|
"op.alter_column('sometable', 'somecolumn', "
|
|
"existing_type=sa.Integer(), nullable=True, schema='foo')"
|
|
)
|
|
|
|
def test_render_fk_constraint_kwarg(self):
|
|
m = MetaData()
|
|
t1 = Table('t', m, Column('c', Integer))
|
|
t2 = Table('t2', m, Column('c_rem', Integer))
|
|
|
|
fk = ForeignKeyConstraint([t1.c.c], [t2.c.c_rem], onupdate="CASCADE")
|
|
if not util.sqla_08:
|
|
t1.append_constraint(fk)
|
|
|
|
# SQLA 0.9 generates a u'' here for remote cols while 0.8 does not,
|
|
# so just whack out "'u" here from the generated
|
|
|
|
eq_ignore_whitespace(
|
|
re.sub(
|
|
r"u'", "'",
|
|
autogenerate.render._render_constraint(
|
|
fk, self.autogen_context)),
|
|
"sa.ForeignKeyConstraint(['c'], ['t2.c_rem'], onupdate='CASCADE')"
|
|
)
|
|
|
|
fk = ForeignKeyConstraint([t1.c.c], [t2.c.c_rem], ondelete="CASCADE")
|
|
if not util.sqla_08:
|
|
t1.append_constraint(fk)
|
|
|
|
eq_ignore_whitespace(
|
|
re.sub(
|
|
r"u'", "'",
|
|
autogenerate.render._render_constraint(
|
|
fk, self.autogen_context)),
|
|
"sa.ForeignKeyConstraint(['c'], ['t2.c_rem'], ondelete='CASCADE')"
|
|
)
|
|
|
|
fk = ForeignKeyConstraint([t1.c.c], [t2.c.c_rem], deferrable=True)
|
|
if not util.sqla_08:
|
|
t1.append_constraint(fk)
|
|
eq_ignore_whitespace(
|
|
re.sub(
|
|
r"u'", "'",
|
|
autogenerate.render._render_constraint(
|
|
fk, self.autogen_context),
|
|
),
|
|
"sa.ForeignKeyConstraint(['c'], ['t2.c_rem'], deferrable=True)"
|
|
)
|
|
|
|
fk = ForeignKeyConstraint([t1.c.c], [t2.c.c_rem], initially="XYZ")
|
|
if not util.sqla_08:
|
|
t1.append_constraint(fk)
|
|
eq_ignore_whitespace(
|
|
re.sub(
|
|
r"u'", "'",
|
|
autogenerate.render._render_constraint(
|
|
fk, self.autogen_context)
|
|
),
|
|
"sa.ForeignKeyConstraint(['c'], ['t2.c_rem'], initially='XYZ')"
|
|
)
|
|
|
|
def test_render_fk_constraint_use_alter(self):
|
|
m = MetaData()
|
|
Table('t', m, Column('c', Integer))
|
|
t2 = Table(
|
|
't2', m,
|
|
Column(
|
|
'c_rem', Integer,
|
|
ForeignKey('t.c', name="fk1", use_alter=True)))
|
|
const = list(t2.foreign_keys)[0].constraint
|
|
|
|
eq_ignore_whitespace(
|
|
autogenerate.render._render_constraint(
|
|
const, self.autogen_context),
|
|
"sa.ForeignKeyConstraint(['c_rem'], ['t.c'], "
|
|
"name='fk1', use_alter=True)"
|
|
)
|
|
|
|
def test_render_fk_constraint_w_metadata_schema(self):
|
|
m = MetaData(schema="foo")
|
|
t1 = Table('t', m, Column('c', Integer))
|
|
t2 = Table('t2', m, Column('c_rem', Integer))
|
|
|
|
fk = ForeignKeyConstraint([t1.c.c], [t2.c.c_rem], onupdate="CASCADE")
|
|
if not util.sqla_08:
|
|
t1.append_constraint(fk)
|
|
|
|
eq_ignore_whitespace(
|
|
re.sub(
|
|
r"u'", "'",
|
|
autogenerate.render._render_constraint(
|
|
fk, self.autogen_context)
|
|
),
|
|
"sa.ForeignKeyConstraint(['c'], ['foo.t2.c_rem'], "
|
|
"onupdate='CASCADE')"
|
|
)
|
|
|
|
def test_render_check_constraint_literal(self):
|
|
eq_ignore_whitespace(
|
|
autogenerate.render._render_check_constraint(
|
|
CheckConstraint("im a constraint", name='cc1'),
|
|
self.autogen_context
|
|
),
|
|
"sa.CheckConstraint(!U'im a constraint', name='cc1')"
|
|
)
|
|
|
|
def test_render_check_constraint_sqlexpr(self):
|
|
c = column('c')
|
|
five = literal_column('5')
|
|
ten = literal_column('10')
|
|
eq_ignore_whitespace(
|
|
autogenerate.render._render_check_constraint(
|
|
CheckConstraint(and_(c > five, c < ten)),
|
|
self.autogen_context
|
|
),
|
|
"sa.CheckConstraint(!U'c > 5 AND c < 10')"
|
|
)
|
|
|
|
@config.requirements.fail_before_sqla_080
|
|
def test_render_check_constraint_literal_binds(self):
|
|
c = column('c')
|
|
eq_ignore_whitespace(
|
|
autogenerate.render._render_check_constraint(
|
|
CheckConstraint(and_(c > 5, c < 10)),
|
|
self.autogen_context
|
|
),
|
|
"sa.CheckConstraint(!U'c > 5 AND c < 10')"
|
|
)
|
|
|
|
def test_render_unique_constraint_opts(self):
|
|
m = MetaData()
|
|
t = Table('t', m, Column('c', Integer))
|
|
eq_ignore_whitespace(
|
|
autogenerate.render._render_unique_constraint(
|
|
UniqueConstraint(t.c.c, name='uq_1', deferrable='XYZ'),
|
|
self.autogen_context
|
|
),
|
|
"sa.UniqueConstraint('c', deferrable='XYZ', name='uq_1')"
|
|
)
|
|
|
|
def test_add_unique_constraint_unicode_schema(self):
|
|
m = MetaData()
|
|
t = Table(
|
|
't', m, Column('c', Integer),
|
|
schema=compat.ue('\u0411\u0435\u0437')
|
|
)
|
|
eq_ignore_whitespace(
|
|
autogenerate.render._add_unique_constraint(
|
|
UniqueConstraint(t.c.c),
|
|
self.autogen_context
|
|
),
|
|
"op.create_unique_constraint(None, 't', ['c'], "
|
|
"schema=%r)" % compat.ue('\u0411\u0435\u0437')
|
|
)
|
|
|
|
def test_render_modify_nullable_w_default(self):
|
|
eq_ignore_whitespace(
|
|
autogenerate.render._modify_col(
|
|
"sometable", "somecolumn",
|
|
self.autogen_context,
|
|
existing_type=Integer(),
|
|
existing_server_default="5",
|
|
nullable=True),
|
|
"op.alter_column('sometable', 'somecolumn', "
|
|
"existing_type=sa.Integer(), nullable=True, "
|
|
"existing_server_default='5')"
|
|
)
|
|
|
|
@config.requirements.fail_before_sqla_079
|
|
def test_render_enum(self):
|
|
eq_ignore_whitespace(
|
|
autogenerate.render._repr_type(
|
|
Enum("one", "two", "three", name="myenum"),
|
|
self.autogen_context),
|
|
"sa.Enum('one', 'two', 'three', name='myenum')"
|
|
)
|
|
eq_ignore_whitespace(
|
|
autogenerate.render._repr_type(
|
|
Enum("one", "two", "three"),
|
|
self.autogen_context),
|
|
"sa.Enum('one', 'two', 'three')"
|
|
)
|
|
|
|
@exclusions.fails_if(
|
|
lambda config: (util.sqla_09 and not util.sqla_099) or not util.sqla_079,
|
|
"Fails on SQLAlchemy <0.7.9, 0.9.0-0.9.8"
|
|
)
|
|
def test_render_non_native_enum(self):
|
|
eq_ignore_whitespace(
|
|
autogenerate.render._repr_type(
|
|
Enum("one", "two", "three", native_enum=False),
|
|
self.autogen_context),
|
|
"sa.Enum('one', 'two', 'three', native_enum=False)"
|
|
)
|
|
|
|
def test_repr_plain_sqla_type(self):
|
|
type_ = Integer()
|
|
autogen_context = {
|
|
'opts': {
|
|
'sqlalchemy_module_prefix': 'sa.',
|
|
'alembic_module_prefix': 'op.',
|
|
},
|
|
'dialect': mysql.dialect()
|
|
}
|
|
|
|
eq_ignore_whitespace(
|
|
autogenerate.render._repr_type(type_, autogen_context),
|
|
"sa.Integer()"
|
|
)
|
|
|
|
def test_repr_custom_type_w_sqla_prefix(self):
|
|
autogen_context = {
|
|
'opts': {
|
|
'sqlalchemy_module_prefix': 'sa.',
|
|
'alembic_module_prefix': 'op.',
|
|
'user_module_prefix': None
|
|
},
|
|
'dialect': mysql.dialect()
|
|
}
|
|
|
|
class MyType(UserDefinedType):
|
|
pass
|
|
|
|
MyType.__module__ = "sqlalchemy_util.types"
|
|
|
|
type_ = MyType()
|
|
|
|
eq_ignore_whitespace(
|
|
autogenerate.render._repr_type(type_, autogen_context),
|
|
"sqlalchemy_util.types.MyType()"
|
|
)
|
|
|
|
def test_repr_user_type_user_prefix_None(self):
|
|
class MyType(UserDefinedType):
|
|
|
|
def get_col_spec(self):
|
|
return "MYTYPE"
|
|
|
|
type_ = MyType()
|
|
autogen_context = {
|
|
'opts': {
|
|
'sqlalchemy_module_prefix': 'sa.',
|
|
'alembic_module_prefix': 'op.',
|
|
'user_module_prefix': None
|
|
},
|
|
'dialect': mysql.dialect()
|
|
}
|
|
|
|
eq_ignore_whitespace(
|
|
autogenerate.render._repr_type(type_, autogen_context),
|
|
"tests.test_autogen_render.MyType()"
|
|
)
|
|
|
|
def test_repr_user_type_user_prefix_present(self):
|
|
from sqlalchemy.types import UserDefinedType
|
|
|
|
class MyType(UserDefinedType):
|
|
|
|
def get_col_spec(self):
|
|
return "MYTYPE"
|
|
|
|
type_ = MyType()
|
|
autogen_context = {
|
|
'opts': {
|
|
'sqlalchemy_module_prefix': 'sa.',
|
|
'alembic_module_prefix': 'op.',
|
|
'user_module_prefix': 'user.',
|
|
},
|
|
'dialect': mysql.dialect()
|
|
}
|
|
|
|
eq_ignore_whitespace(
|
|
autogenerate.render._repr_type(type_, autogen_context),
|
|
"user.MyType()"
|
|
)
|
|
|
|
@config.requirements.sqlalchemy_09
|
|
def test_repr_dialect_type(self):
|
|
from sqlalchemy.dialects.mysql import VARCHAR
|
|
|
|
type_ = VARCHAR(20, charset='utf8', national=True)
|
|
autogen_context = {
|
|
'opts': {
|
|
'sqlalchemy_module_prefix': 'sa.',
|
|
'alembic_module_prefix': 'op.',
|
|
'user_module_prefix': None,
|
|
},
|
|
'imports': set(),
|
|
'dialect': mysql.dialect()
|
|
}
|
|
eq_ignore_whitespace(
|
|
autogenerate.render._repr_type(type_, autogen_context),
|
|
"mysql.VARCHAR(charset='utf8', national=True, length=20)"
|
|
)
|
|
eq_(autogen_context['imports'],
|
|
set(['from sqlalchemy.dialects import mysql'])
|
|
)
|
|
|
|
def test_render_server_default_text(self):
|
|
c = Column(
|
|
'updated_at', TIMESTAMP(),
|
|
server_default=text('now()'),
|
|
nullable=False)
|
|
result = autogenerate.render._render_column(
|
|
c, self.autogen_context
|
|
)
|
|
eq_ignore_whitespace(
|
|
result,
|
|
'sa.Column(\'updated_at\', sa.TIMESTAMP(), '
|
|
'server_default=sa.text(!U\'now()\'), '
|
|
'nullable=False)'
|
|
)
|
|
|
|
def test_render_server_default_native_boolean(self):
|
|
autogen_context = {
|
|
'opts': {
|
|
'sqlalchemy_module_prefix': 'sa.',
|
|
'alembic_module_prefix': 'op.',
|
|
},
|
|
'dialect': postgresql.dialect()
|
|
}
|
|
c = Column(
|
|
'updated_at', Boolean(),
|
|
server_default=false(),
|
|
nullable=False)
|
|
result = autogenerate.render._render_column(
|
|
c, autogen_context,
|
|
)
|
|
eq_ignore_whitespace(
|
|
result,
|
|
'sa.Column(\'updated_at\', sa.Boolean(), '
|
|
'server_default=sa.text(!U\'false\'), '
|
|
'nullable=False)'
|
|
)
|
|
|
|
@config.requirements.fail_before_sqla_09
|
|
def test_render_server_default_non_native_boolean(self):
|
|
c = Column(
|
|
'updated_at', Boolean(),
|
|
server_default=false(),
|
|
nullable=False)
|
|
dialect = DefaultDialect()
|
|
autogen_context = {
|
|
'opts': {
|
|
'sqlalchemy_module_prefix': 'sa.',
|
|
'alembic_module_prefix': 'op.',
|
|
},
|
|
'dialect': dialect
|
|
}
|
|
|
|
result = autogenerate.render._render_column(
|
|
c, autogen_context
|
|
)
|
|
eq_ignore_whitespace(
|
|
result,
|
|
'sa.Column(\'updated_at\', sa.Boolean(), '
|
|
'server_default=sa.text(!U\'0\'), '
|
|
'nullable=False)'
|
|
)
|
|
|
|
def test_render_server_default_func(self):
|
|
c = Column(
|
|
'updated_at', TIMESTAMP(),
|
|
server_default=func.now(),
|
|
nullable=False)
|
|
result = autogenerate.render._render_column(
|
|
c, self.autogen_context
|
|
)
|
|
eq_ignore_whitespace(
|
|
result,
|
|
'sa.Column(\'updated_at\', sa.TIMESTAMP(), '
|
|
'server_default=sa.text(!U\'now()\'), '
|
|
'nullable=False)'
|
|
)
|
|
|
|
def test_render_server_default_int(self):
|
|
c = Column(
|
|
'value', Integer,
|
|
server_default="0")
|
|
result = autogenerate.render._render_column(
|
|
c, self.autogen_context
|
|
)
|
|
eq_(
|
|
result,
|
|
"sa.Column('value', sa.Integer(), "
|
|
"server_default='0', nullable=True)"
|
|
)
|
|
|
|
def test_render_modify_reflected_int_server_default(self):
|
|
eq_ignore_whitespace(
|
|
autogenerate.render._modify_col(
|
|
"sometable", "somecolumn",
|
|
self.autogen_context,
|
|
existing_type=Integer(),
|
|
existing_server_default=DefaultClause(text("5")),
|
|
nullable=True),
|
|
"op.alter_column('sometable', 'somecolumn', "
|
|
"existing_type=sa.Integer(), nullable=True, "
|
|
"existing_server_default=sa.text(!U'5'))"
|
|
)
|
|
|
|
|
|
class RenderNamingConventionTest(TestBase):
|
|
__requires__ = ('sqlalchemy_094',)
|
|
|
|
@classmethod
|
|
def setup_class(cls):
|
|
cls.autogen_context = {
|
|
'opts': {
|
|
'sqlalchemy_module_prefix': 'sa.',
|
|
'alembic_module_prefix': 'op.',
|
|
},
|
|
'dialect': postgresql.dialect()
|
|
}
|
|
|
|
def setUp(self):
|
|
|
|
convention = {
|
|
"ix": 'ix_%(custom)s_%(column_0_label)s',
|
|
"uq": "uq_%(custom)s_%(table_name)s_%(column_0_name)s",
|
|
"ck": "ck_%(custom)s_%(table_name)s",
|
|
"fk": "fk_%(custom)s_%(table_name)s_"
|
|
"%(column_0_name)s_%(referred_table_name)s",
|
|
"pk": "pk_%(custom)s_%(table_name)s",
|
|
"custom": lambda const, table: "ct"
|
|
}
|
|
|
|
self.metadata = MetaData(
|
|
naming_convention=convention
|
|
)
|
|
|
|
def test_schema_type_boolean(self):
|
|
t = Table('t', self.metadata, Column('c', Boolean(name='xyz')))
|
|
eq_ignore_whitespace(
|
|
autogenerate.render._add_column(
|
|
None, "t", t.c.c,
|
|
self.autogen_context),
|
|
"op.add_column('t', "
|
|
"sa.Column('c', sa.Boolean(name='xyz'), nullable=True))"
|
|
)
|
|
|
|
def test_explicit_unique_constraint(self):
|
|
t = Table('t', self.metadata, Column('c', Integer))
|
|
eq_ignore_whitespace(
|
|
autogenerate.render._render_unique_constraint(
|
|
UniqueConstraint(t.c.c, deferrable='XYZ'),
|
|
self.autogen_context
|
|
),
|
|
"sa.UniqueConstraint('c', deferrable='XYZ', "
|
|
"name=op.f('uq_ct_t_c'))"
|
|
)
|
|
|
|
def test_explicit_named_unique_constraint(self):
|
|
t = Table('t', self.metadata, Column('c', Integer))
|
|
eq_ignore_whitespace(
|
|
autogenerate.render._render_unique_constraint(
|
|
UniqueConstraint(t.c.c, name='q'),
|
|
self.autogen_context
|
|
),
|
|
"sa.UniqueConstraint('c', name='q')"
|
|
)
|
|
|
|
def test_render_add_index(self):
|
|
t = Table('test', self.metadata,
|
|
Column('id', Integer, primary_key=True),
|
|
Column('active', Boolean()),
|
|
Column('code', String(255)),
|
|
)
|
|
idx = Index(None, t.c.active, t.c.code)
|
|
eq_ignore_whitespace(
|
|
autogenerate.render._add_index(idx, self.autogen_context),
|
|
"op.create_index(op.f('ix_ct_test_active'), 'test', "
|
|
"['active', 'code'], unique=False)"
|
|
)
|
|
|
|
def test_render_drop_index(self):
|
|
t = Table('test', self.metadata,
|
|
Column('id', Integer, primary_key=True),
|
|
Column('active', Boolean()),
|
|
Column('code', String(255)),
|
|
)
|
|
idx = Index(None, t.c.active, t.c.code)
|
|
eq_ignore_whitespace(
|
|
autogenerate.render._drop_index(idx, self.autogen_context),
|
|
"op.drop_index(op.f('ix_ct_test_active'), table_name='test')"
|
|
)
|
|
|
|
def test_render_add_index_schema(self):
|
|
t = Table('test', self.metadata,
|
|
Column('id', Integer, primary_key=True),
|
|
Column('active', Boolean()),
|
|
Column('code', String(255)),
|
|
schema='CamelSchema'
|
|
)
|
|
idx = Index(None, t.c.active, t.c.code)
|
|
eq_ignore_whitespace(
|
|
autogenerate.render._add_index(idx, self.autogen_context),
|
|
"op.create_index(op.f('ix_ct_CamelSchema_test_active'), 'test', "
|
|
"['active', 'code'], unique=False, schema='CamelSchema')"
|
|
)
|
|
|
|
def test_implicit_unique_constraint(self):
|
|
t = Table('t', self.metadata, Column('c', Integer, unique=True))
|
|
uq = [c for c in t.constraints if isinstance(c, UniqueConstraint)][0]
|
|
eq_ignore_whitespace(
|
|
autogenerate.render._render_unique_constraint(uq,
|
|
self.autogen_context
|
|
),
|
|
"sa.UniqueConstraint('c', name=op.f('uq_ct_t_c'))"
|
|
)
|
|
|
|
def test_inline_pk_constraint(self):
|
|
t = Table('t', self.metadata, Column('c', Integer, primary_key=True))
|
|
eq_ignore_whitespace(
|
|
autogenerate.render._add_table(t, self.autogen_context),
|
|
"op.create_table('t',sa.Column('c', sa.Integer(), nullable=False),"
|
|
"sa.PrimaryKeyConstraint('c', name=op.f('pk_ct_t')))"
|
|
)
|
|
|
|
def test_inline_ck_constraint(self):
|
|
t = Table(
|
|
't', self.metadata, Column('c', Integer), CheckConstraint("c > 5"))
|
|
eq_ignore_whitespace(
|
|
autogenerate.render._add_table(t, self.autogen_context),
|
|
"op.create_table('t',sa.Column('c', sa.Integer(), nullable=True),"
|
|
"sa.CheckConstraint(!U'c > 5', name=op.f('ck_ct_t')))"
|
|
)
|
|
|
|
def test_inline_fk(self):
|
|
t = Table('t', self.metadata, Column('c', Integer, ForeignKey('q.id')))
|
|
eq_ignore_whitespace(
|
|
autogenerate.render._add_table(t, self.autogen_context),
|
|
"op.create_table('t',sa.Column('c', sa.Integer(), nullable=True),"
|
|
"sa.ForeignKeyConstraint(['c'], ['q.id'], "
|
|
"name=op.f('fk_ct_t_c_q')))"
|
|
)
|
|
|
|
def test_render_check_constraint_renamed(self):
|
|
"""test that constraints from autogenerate render with
|
|
the naming convention name explicitly. These names should
|
|
be frozen into the migration scripts so that they remain
|
|
the same if the application's naming convention changes.
|
|
|
|
However, op.create_table() and others need to be careful that
|
|
these don't double up when the "%(constraint_name)s" token is
|
|
used.
|
|
|
|
"""
|
|
m1 = MetaData(naming_convention={
|
|
"ck": "ck_%(table_name)s_%(constraint_name)s"})
|
|
ck = CheckConstraint("im a constraint", name="cc1")
|
|
Table('t', m1, Column('x'), ck)
|
|
|
|
eq_ignore_whitespace(
|
|
autogenerate.render._render_check_constraint(
|
|
ck,
|
|
self.autogen_context
|
|
),
|
|
"sa.CheckConstraint(!U'im a constraint', name=op.f('ck_t_cc1'))"
|
|
)
|
|
|