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, table
from alembic.migration import MigrationContext
'target_metadata': MetaData()
}
context = MigrationContext.configure(
- dialect_name="mysql",
+ dialect=DefaultDialect(),
opts=ctx_opts
)
self.autogen_context = api.AutogenContext(context)
- context = MigrationContext.configure(
- dialect_name="postgresql",
- opts=ctx_opts
- )
- self.pg_autogen_context = api.AutogenContext(context)
-
- context = MigrationContext.configure(
- dialect=DefaultDialect(),
- opts=ctx_opts
- )
- self.default_autogen_context = api.AutogenContext(context)
-
def test_render_add_index(self):
"""
autogenerate.render._add_index
"['active', 'code'], unique=False)"
)
- 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'))
-
- op_obj = ops.CreateIndexOp.from_index(idx)
-
- if compat.sqla_08:
- eq_ignore_whitespace(
- autogenerate.render_op_text(autogen_context, op_obj),
- """op.create_index('foo_idx', 't', \
-['x', 'y'], unique=False, """
- """postgresql_where=sa.text(!U"y = 'something'"))"""
- )
- else:
- eq_ignore_whitespace(
- autogenerate.render_op_text(autogen_context, op_obj),
- """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()
eq_ignore_whitespace(
autogenerate.render_op_text(self.autogen_context, op_obj),
"op.create_index('test_lower_code_idx', 'test', "
- "[sa.text(!U'CAST(code AS CHAR)')], unique=False)"
+ "[sa.text(!U'CAST(code AS VARCHAR)')], unique=False)"
)
else:
eq_ignore_whitespace(
autogenerate.render_op_text(self.autogen_context, op_obj),
"op.create_index('test_lower_code_idx', 'test', "
- "[sa.text(!U'CAST(test.code AS CHAR)')], unique=False)"
+ "[sa.text(!U'CAST(test.code AS VARCHAR)')], unique=False)"
)
@config.requirements.fail_before_sqla_080
'nullable=False)'
)
- def test_render_server_default_native_boolean(self):
- c = Column(
- 'updated_at', Boolean(),
- server_default=false(),
- nullable=False)
- result = autogenerate.render._render_column(
- c, self.pg_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(
nullable=False)
result = autogenerate.render._render_column(
- c, self.default_autogen_context
+ c, self.autogen_context
)
eq_ignore_whitespace(
result,
from alembic.testing.fixtures import op_fixture
from alembic.testing import config
from alembic import op
+from alembic.util import compat
+from alembic.testing import eq_ignore_whitespace
+from alembic import autogenerate
+from sqlalchemy import Index
+from sqlalchemy import Boolean
+from sqlalchemy import false
class PostgresqlOpTest(TestBase):
None,
Column('x', Integer, autoincrement=False, primary_key=True)
)
+
+
+class PostgresqlAutogenRenderTest(TestBase):
+
+ def setUp(self):
+ ctx_opts = {
+ 'sqlalchemy_module_prefix': 'sa.',
+ 'alembic_module_prefix': 'op.',
+ 'target_metadata': MetaData()
+ }
+ context = MigrationContext.configure(
+ dialect_name="postgresql",
+ opts=ctx_opts
+ )
+
+ self.autogen_context = api.AutogenContext(context)
+
+ def test_render_add_index_pg_where(self):
+ autogen_context = self.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'))
+
+ op_obj = ops.CreateIndexOp.from_index(idx)
+
+ if compat.sqla_08:
+ eq_ignore_whitespace(
+ autogenerate.render_op_text(autogen_context, op_obj),
+ """op.create_index('foo_idx', 't', \
+['x', 'y'], unique=False, """
+ """postgresql_where=sa.text(!U"y = 'something'"))"""
+ )
+ else:
+ eq_ignore_whitespace(
+ autogenerate.render_op_text(autogen_context, op_obj),
+ """op.create_index('foo_idx', 't', ['x', 'y'], \
+unique=False, """
+ """postgresql_where=sa.text(!U't.y = %(y_1)s'))"""
+ )
+
+ def test_render_server_default_native_boolean(self):
+ c = Column(
+ 'updated_at', Boolean(),
+ server_default=false(),
+ nullable=False)
+ result = autogenerate.render._render_column(
+ c, self.autogen_context,
+ )
+ eq_ignore_whitespace(
+ result,
+ 'sa.Column(\'updated_at\', sa.Boolean(), '
+ 'server_default=sa.text(!U\'false\'), '
+ 'nullable=False)'
+ )
+