summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
-rw-r--r--alembic/testing/__init__.py5
-rw-r--r--alembic/testing/fixtures.py27
-rw-r--r--alembic/testing/requirements.py29
-rw-r--r--tests/test_autogen_render.py24
-rw-r--r--tests/test_config.py58
-rw-r--r--tests/test_environment.py11
-rw-r--r--tests/test_mssql.py14
-rw-r--r--tests/test_mysql.py18
-rw-r--r--tests/test_offline_environment.py16
-rw-r--r--tests/test_op.py1525
-rw-r--r--tests/test_op_naming_convention.py327
-rw-r--r--tox.ini3
12 files changed, 990 insertions, 1067 deletions
diff --git a/alembic/testing/__init__.py b/alembic/testing/__init__.py
index 9e1a01a..0e0fd68 100644
--- a/alembic/testing/__init__.py
+++ b/alembic/testing/__init__.py
@@ -1,4 +1,7 @@
from .fixtures import TestBase
-from .assertions import eq_, ne_, assert_raises_message
+from .assertions import eq_, ne_, is_, assert_raises_message, \
+ eq_ignore_whitespace
from sqlalchemy.testing import config
+
+from sqlalchemy.testing.config import requirements as requires
diff --git a/alembic/testing/fixtures.py b/alembic/testing/fixtures.py
index 67db86c..001c402 100644
--- a/alembic/testing/fixtures.py
+++ b/alembic/testing/fixtures.py
@@ -5,7 +5,6 @@ import re
import shutil
import textwrap
-from nose import SkipTest
from sqlalchemy.engine import default
from sqlalchemy import create_engine, text, MetaData
from sqlalchemy.exc import SQLAlchemyError
@@ -41,32 +40,6 @@ def capture_db():
_engs = {}
-@decorator
-def requires_08(fn, *arg, **kw):
- if not util.sqla_08:
- raise SkipTest("SQLAlchemy 0.8.0b2 or greater required")
- return fn(*arg, **kw)
-
-
-@decorator
-def requires_09(fn, *arg, **kw):
- if not util.sqla_09:
- raise SkipTest("SQLAlchemy 0.9 or greater required")
- return fn(*arg, **kw)
-
-
-@decorator
-def requires_092(fn, *arg, **kw):
- if not util.sqla_092:
- raise SkipTest("SQLAlchemy 0.9.2 or greater required")
- return fn(*arg, **kw)
-
-
-@decorator
-def requires_094(fn, *arg, **kw):
- if not util.sqla_094:
- raise SkipTest("SQLAlchemy 0.9.4 or greater required")
- return fn(*arg, **kw)
@contextmanager
diff --git a/alembic/testing/requirements.py b/alembic/testing/requirements.py
index 2d2b678..ed79fb8 100644
--- a/alembic/testing/requirements.py
+++ b/alembic/testing/requirements.py
@@ -1,5 +1,6 @@
from sqlalchemy.testing.requirements import Requirements
from sqlalchemy.testing import exclusions
+from alembic import util
class SuiteRequirements(Requirements):
@@ -10,3 +11,31 @@ class SuiteRequirements(Requirements):
return exclusions.open()
+ @property
+ def sqlalchemy_08(self):
+
+ return exclusions.skip_if(
+ lambda config: not util.sqla_08,
+ "SQLAlchemy 0.8.0b2 or greater required"
+ )
+
+ @property
+ def sqlalchemy_09(self):
+ return exclusions.skip_if(
+ lambda config: not util.sqla_09,
+ "SQLAlchemy 0.9.0 or greater required"
+ )
+
+ @property
+ def sqlalchemy_092(self):
+ return exclusions.skip_if(
+ lambda config: not util.sqla_092,
+ "SQLAlchemy 0.9.2 or greater required"
+ )
+
+ @property
+ def sqlalchemy_094(self):
+ return exclusions.skip_if(
+ lambda config: not util.sqla_094,
+ "SQLAlchemy 0.9.4 or greater required"
+ )
diff --git a/tests/test_autogen_render.py b/tests/test_autogen_render.py
index cfb0f90..ece1617 100644
--- a/tests/test_autogen_render.py
+++ b/tests/test_autogen_render.py
@@ -1,26 +1,26 @@
import re
import sys
-from unittest import TestCase
+from alembic.testing import TestBase
-from sqlalchemy import MetaData, Column, Table, Integer, String, Text, \
- Numeric, CHAR, ForeignKey, DATETIME, INTEGER, \
- TypeDecorator, CheckConstraint, Unicode, Enum,\
+from sqlalchemy import MetaData, Column, Table, String, \
+ Numeric, CHAR, ForeignKey, DATETIME, Integer, \
+ CheckConstraint, Unicode, Enum,\
UniqueConstraint, Boolean, ForeignKeyConstraint,\
- PrimaryKeyConstraint, Index, func
+ PrimaryKeyConstraint, Index
from sqlalchemy.types import TIMESTAMP
from sqlalchemy.dialects import mysql, postgresql
from sqlalchemy.sql import and_, column, literal_column
-from . import patch
+from alembic.testing.mock import patch
from alembic import autogenerate, util, compat
-from . import eq_, eq_ignore_whitespace, requires_092, \
- requires_09, requires_094
+from alembic.testing import eq_, eq_ignore_whitespace, config
+
py3k = sys.version_info >= (3, )
-class AutogenRenderTest(TestCase):
+class AutogenRenderTest(TestBase):
"""test individual directives"""
@@ -769,7 +769,7 @@ render:primary_key\n)"""
"user.MyType()"
)
- @requires_09
+ @config.requirements.sqlalchemy_09
def test_repr_dialect_type(self):
from sqlalchemy.dialects.mysql import VARCHAR
@@ -792,10 +792,10 @@ render:primary_key\n)"""
)
-class RenderNamingConventionTest(TestCase):
+class RenderNamingConventionTest(TestBase):
+ __requires__ = ('sqlalchemy_094',)
@classmethod
- @requires_094
def setup_class(cls):
cls.autogen_context = {
'opts': {
diff --git a/tests/test_config.py b/tests/test_config.py
index cd56d13..1d03b18 100644
--- a/tests/test_config.py
+++ b/tests/test_config.py
@@ -4,49 +4,49 @@ from alembic import config, util, compat
from alembic.migration import MigrationContext
from alembic.operations import Operations
from alembic.script import ScriptDirectory
-import unittest
-from . import Mock, call
+from alembic.testing.fixtures import TestBase
+from alembic.testing.mock import Mock, call
-from . import eq_, capture_db, assert_raises_message
+from alembic.testing import eq_, assert_raises_message
+from alembic.testing.fixtures import capture_db
-def test_config_no_file_main_option():
- cfg = config.Config()
- cfg.set_main_option("url", "postgresql://foo/bar")
+class ConfigTest(TestBase):
- eq_(cfg.get_main_option("url"), "postgresql://foo/bar")
+ def test_config_no_file_main_option(self):
+ cfg = config.Config()
+ cfg.set_main_option("url", "postgresql://foo/bar")
+ eq_(cfg.get_main_option("url"), "postgresql://foo/bar")
-def test_config_no_file_section_option():
- cfg = config.Config()
- cfg.set_section_option("foo", "url", "postgresql://foo/bar")
+ def test_config_no_file_section_option(self):
+ cfg = config.Config()
+ cfg.set_section_option("foo", "url", "postgresql://foo/bar")
- eq_(cfg.get_section_option("foo", "url"), "postgresql://foo/bar")
+ eq_(cfg.get_section_option("foo", "url"), "postgresql://foo/bar")
- cfg.set_section_option("foo", "echo", "True")
- eq_(cfg.get_section_option("foo", "echo"), "True")
+ cfg.set_section_option("foo", "echo", "True")
+ eq_(cfg.get_section_option("foo", "echo"), "True")
+ def test_standalone_op(self):
+ eng, buf = capture_db()
-def test_standalone_op():
- eng, buf = capture_db()
+ env = MigrationContext.configure(eng)
+ op = Operations(env)
- env = MigrationContext.configure(eng)
- op = Operations(env)
+ op.alter_column("t", "c", nullable=True)
+ eq_(buf, ['ALTER TABLE t ALTER COLUMN c DROP NOT NULL'])
- op.alter_column("t", "c", nullable=True)
- eq_(buf, ['ALTER TABLE t ALTER COLUMN c DROP NOT NULL'])
-
-
-def test_no_script_error():
- cfg = config.Config()
- assert_raises_message(
- util.CommandError,
- "No 'script_location' key found in configuration.",
- ScriptDirectory.from_config, cfg
- )
+ def test_no_script_error(self):
+ cfg = config.Config()
+ assert_raises_message(
+ util.CommandError,
+ "No 'script_location' key found in configuration.",
+ ScriptDirectory.from_config, cfg
+ )
-class OutputEncodingTest(unittest.TestCase):
+class OutputEncodingTest(TestBase):
def test_plain(self):
stdout = Mock(encoding='latin-1')
diff --git a/tests/test_environment.py b/tests/test_environment.py
index c877960..cb8a0a2 100644
--- a/tests/test_environment.py
+++ b/tests/test_environment.py
@@ -3,14 +3,15 @@
from alembic.script import ScriptDirectory
from alembic.environment import EnvironmentContext
from alembic.migration import MigrationContext
-import unittest
-from . import Mock, call, _no_sql_testing_config, staging_env, \
- clear_staging_env
+from alembic.testing.fixtures import TestBase
+from alembic.testing.mock import Mock, call
+from alembic.testing.env import _no_sql_testing_config, \
+ staging_env, clear_staging_env
-from . import eq_, is_
+from alembic.testing import eq_, is_
-class EnvironmentTest(unittest.TestCase):
+class EnvironmentTest(TestBase):
def setUp(self):
staging_env()
diff --git a/tests/test_mssql.py b/tests/test_mssql.py
index 92a9546..b87a434 100644
--- a/tests/test_mssql.py
+++ b/tests/test_mssql.py
@@ -1,16 +1,18 @@
"""Test op functions against MSSQL."""
-from unittest import TestCase
+from alembic.testing.fixtures import TestBase
from sqlalchemy import Integer, Column
from alembic import op, command, util
-from . import op_fixture, capture_context_buffer, \
- _no_sql_testing_config, assert_raises_message, staging_env, \
- three_rev_fixture, clear_staging_env, eq_
+from alembic.testing import eq_, assert_raises_message
+from alembic.testing.fixtures import capture_context_buffer, op_fixture
+from alembic.testing.env import staging_env, _no_sql_testing_config, \
+ three_rev_fixture, clear_staging_env
-class FullEnvironmentTests(TestCase):
+
+class FullEnvironmentTests(TestBase):
@classmethod
def setup_class(cls):
@@ -46,7 +48,7 @@ class FullEnvironmentTests(TestCase):
assert "BYE" in buf.getvalue()
-class OpTest(TestCase):
+class OpTest(TestBase):
def test_add_column(self):
context = op_fixture('mssql')
diff --git a/tests/test_mysql.py b/tests/test_mysql.py
index ada6779..7ce71a9 100644
--- a/tests/test_mysql.py
+++ b/tests/test_mysql.py
@@ -1,14 +1,19 @@
from sqlalchemy import Integer, func
-from unittest import TestCase
+from alembic.testing.fixtures import TestBase
+from alembic.testing import config
from sqlalchemy import TIMESTAMP, MetaData, Table, Column, text
from sqlalchemy.engine.reflection import Inspector
from alembic import op, util
-from . import op_fixture, assert_raises_message, db_for_dialect, \
- staging_env, clear_staging_env
+
+from alembic.testing import eq_, assert_raises_message
+from alembic.testing.fixtures import capture_context_buffer, op_fixture
+from alembic.testing.env import staging_env, _no_sql_testing_config, \
+ three_rev_fixture, clear_staging_env
+
from alembic.migration import MigrationContext
-class MySQLOpTest(TestCase):
+class MySQLOpTest(TestBase):
def test_rename_column(self):
context = op_fixture('mysql')
@@ -199,11 +204,12 @@ class MySQLOpTest(TestCase):
)
-class MySQLDefaultCompareTest(TestCase):
+class MySQLDefaultCompareTest(TestBase):
+ __only_on__ = 'mysql'
@classmethod
def setup_class(cls):
- cls.bind = db_for_dialect("mysql")
+ cls.bind = config.db
staging_env()
context = MigrationContext.configure(
connection=cls.bind.connect(),
diff --git a/tests/test_offline_environment.py b/tests/test_offline_environment.py
index 192f87f..b870dc4 100644
--- a/tests/test_offline_environment.py
+++ b/tests/test_offline_environment.py
@@ -1,19 +1,19 @@
-import io
-from unittest import TestCase
+from alembic.testing.fixtures import TestBase
from alembic import command, util
-from . import clear_staging_env, staging_env, \
- _no_sql_testing_config, \
- three_rev_fixture, env_file_fixture,\
- assert_raises_message
+
+from alembic.testing import assert_raises_message
+from alembic.testing.env import staging_env, _no_sql_testing_config, \
+ three_rev_fixture, clear_staging_env, env_file_fixture
+
a = b = c = None
-class OfflineEnvironmentTest(TestCase):
+class OfflineEnvironmentTest(TestBase):
def setUp(self):
- env = staging_env()
+ staging_env()
self.cfg = _no_sql_testing_config()
global a, b, c
diff --git a/tests/test_op.py b/tests/test_op.py
index c353039..58e1cb3 100644
--- a/tests/test_op.py
+++ b/tests/test_op.py
@@ -1,13 +1,15 @@
"""Test against the builders in the op.* module."""
from sqlalchemy import Integer, Column, ForeignKey, \
- Table, String, Boolean, MetaData, CheckConstraint
+ Table, String, Boolean
from sqlalchemy.sql import column, func, text
from sqlalchemy import event
from alembic import op
-from . import op_fixture, assert_raises_message, requires_094, eq_
-from . import mock
+from alembic.testing.fixtures import op_fixture
+from alembic.testing import eq_, assert_raises_message
+from alembic.testing import mock
+from alembic.testing.fixtures import TestBase
@event.listens_for(Table, "after_parent_attach")
@@ -16,802 +18,725 @@ def _add_cols(table, metadata):
table.append_column(Column('bat', Integer))
-def test_rename_table():
- context = op_fixture()
- op.rename_table('t1', 't2')
- context.assert_("ALTER TABLE t1 RENAME TO t2")
-
-
-def test_rename_table_schema():
- context = op_fixture()
- op.rename_table('t1', 't2', schema="foo")
- context.assert_("ALTER TABLE foo.t1 RENAME TO foo.t2")
-
-
-def test_rename_table_postgresql():
- context = op_fixture("postgresql")
- op.rename_table('t1', 't2')
- context.assert_("ALTER TABLE t1 RENAME TO t2")
-
-
-def test_rename_table_schema_postgresql():
- context = op_fixture("postgresql")
- op.rename_table('t1', 't2', schema="foo")
- context.assert_("ALTER TABLE foo.t1 RENAME TO t2")
-
-
-def test_create_index_no_expr_allowed():
- op_fixture()
- assert_raises_message(
- ValueError,
- "String or text\(\) construct expected",
- op.create_index, 'name', 'tname', [func.foo(column('x'))]
- )
-
-
-def test_create_index_quoting():
- context = op_fixture("postgresql")
- op.create_index(
- 'geocoded',
- 'locations',
- ["IShouldBeQuoted"])
- context.assert_(
- 'CREATE INDEX geocoded ON locations ("IShouldBeQuoted")')
-
-
-def test_create_index_expressions():
- context = op_fixture()
- op.create_index(
- 'geocoded',
- 'locations',
- [text('lower(coordinates)')])
- context.assert_(
- "CREATE INDEX geocoded ON locations (lower(coordinates))")
-
-
-def test_create_index_postgresql_expressions():
- context = op_fixture("postgresql")
- op.create_index(
- 'geocoded',
- 'locations',
- [text('lower(coordinates)')],
- postgresql_where=text("locations.coordinates != Null"))
- context.assert_(
- "CREATE INDEX geocoded ON locations (lower(coordinates)) "
- "WHERE locations.coordinates != Null")
-
-
-def test_create_index_postgresql_where():
- context = op_fixture("postgresql")
- op.create_index(
- 'geocoded',
- 'locations',
- ['coordinates'],
- postgresql_where=text("locations.coordinates != Null"))
- context.assert_(
- "CREATE INDEX geocoded ON locations (coordinates) "
- "WHERE locations.coordinates != Null")
-
-
-def test_add_column():
- context = op_fixture()
- op.add_column('t1', Column('c1', Integer, nullable=False))
- context.assert_("ALTER TABLE t1 ADD COLUMN c1 INTEGER NOT NULL")
-
-
-def test_add_column_schema():
- context = op_fixture()
- op.add_column('t1', Column('c1', Integer, nullable=False), schema="foo")
- context.assert_("ALTER TABLE foo.t1 ADD COLUMN c1 INTEGER NOT NULL")
-
-
-def test_add_column_with_default():
- context = op_fixture()
- op.add_column(
- 't1', Column('c1', Integer, nullable=False, server_default="12"))
- context.assert_(
- "ALTER TABLE t1 ADD COLUMN c1 INTEGER DEFAULT '12' NOT NULL")
-
-
-def test_add_column_schema_with_default():
- context = op_fixture()
- op.add_column('t1',
- Column('c1', Integer, nullable=False, server_default="12"),
- schema='foo')
- context.assert_(
- "ALTER TABLE foo.t1 ADD COLUMN c1 INTEGER DEFAULT '12' NOT NULL")
-
-
-def test_add_column_fk():
- context = op_fixture()
- op.add_column(
- 't1', Column('c1', Integer, ForeignKey('c2.id'), nullable=False))
- context.assert_(
- "ALTER TABLE t1 ADD COLUMN c1 INTEGER NOT NULL",
- "ALTER TABLE t1 ADD FOREIGN KEY(c1) REFERENCES c2 (id)"
- )
-
-
-def test_add_column_schema_fk():
- context = op_fixture()
- op.add_column('t1',
- Column('c1', Integer, ForeignKey('c2.id'), nullable=False),
- schema='foo')
- context.assert_(
- "ALTER TABLE foo.t1 ADD COLUMN c1 INTEGER NOT NULL",
- "ALTER TABLE foo.t1 ADD FOREIGN KEY(c1) REFERENCES c2 (id)"
- )
-
-
-def test_add_column_schema_type():
- """Test that a schema type generates its constraints...."""
- context = op_fixture()
- op.add_column('t1', Column('c1', Boolean, nullable=False))
- context.assert_(
- 'ALTER TABLE t1 ADD COLUMN c1 BOOLEAN NOT NULL',
- 'ALTER TABLE t1 ADD CHECK (c1 IN (0, 1))'
- )
-
-
-def test_add_column_schema_schema_type():
- """Test that a schema type generates its constraints...."""
- context = op_fixture()
- op.add_column('t1', Column('c1', Boolean, nullable=False), schema='foo')
- context.assert_(
- 'ALTER TABLE foo.t1 ADD COLUMN c1 BOOLEAN NOT NULL',
- 'ALTER TABLE foo.t1 ADD CHECK (c1 IN (0, 1))'
- )
-
-
-def test_add_column_schema_type_checks_rule():
- """Test that a schema type doesn't generate a
- constraint based on check rule."""
- context = op_fixture('postgresql')
- op.add_column('t1', Column('c1', Boolean, nullable=False))
- context.assert_(
- 'ALTER TABLE t1 ADD COLUMN c1 BOOLEAN NOT NULL',
- )
-
-
-def test_add_column_fk_self_referential():
- context = op_fixture()
- op.add_column(
- 't1', Column('c1', Integer, ForeignKey('t1.c2'), nullable=False))
- context.assert_(
- "ALTER TABLE t1 ADD COLUMN c1 INTEGER NOT NULL",
- "ALTER TABLE t1 ADD FOREIGN KEY(c1) REFERENCES t1 (c2)"
- )
-
-
-def test_add_column_schema_fk_self_referential():
- context = op_fixture()
- op.add_column(
- 't1',
- Column('c1', Integer, ForeignKey('foo.t1.c2'), nullable=False),
- schema='foo')
- context.assert_(
- "ALTER TABLE foo.t1 ADD COLUMN c1 INTEGER NOT NULL",
- "ALTER TABLE foo.t1 ADD FOREIGN KEY(c1) REFERENCES foo.t1 (c2)"
- )
-
-
-def test_add_column_fk_schema():
- context = op_fixture()
- op.add_column(
- 't1',
- Column('c1', Integer, ForeignKey('remote.t2.c2'), nullable=False))
- context.assert_(
- 'ALTER TABLE t1 ADD COLUMN c1 INTEGER NOT NULL',
- 'ALTER TABLE t1 ADD FOREIGN KEY(c1) REFERENCES remote.t2 (c2)'
- )
-
-
-def test_add_column_schema_fk_schema():
- context = op_fixture()
- op.add_column(
- 't1',
- Column('c1', Integer, ForeignKey('remote.t2.c2'), nullable=False),
- schema='foo')
- context.assert_(
- 'ALTER TABLE foo.t1 ADD COLUMN c1 INTEGER NOT NULL',
- 'ALTER TABLE foo.t1 ADD FOREIGN KEY(c1) REFERENCES remote.t2 (c2)'
- )
-
-
-def test_drop_column():
- context = op_fixture()
- op.drop_column('t1', 'c1')
- context.assert_("ALTER TABLE t1 DROP COLUMN c1")
-
-
-def test_drop_column_schema():
- context = op_fixture()
- op.drop_column('t1', 'c1', schema='foo')
- context.assert_("ALTER TABLE foo.t1 DROP COLUMN c1")
-
-
-def test_alter_column_nullable():
- context = op_fixture()
- op.alter_column("t", "c", nullable=True)
- context.assert_(
- # TODO: not sure if this is PG only or standard
- # SQL
- "ALTER TABLE t ALTER COLUMN c DROP NOT NULL"
- )
-
-
-def test_alter_column_schema_nullable():
- context = op_fixture()
- op.alter_column("t", "c", nullable=True, schema='foo')
- context.assert_(
- # TODO: not sure if this is PG only or standard
- # SQL
- "ALTER TABLE foo.t ALTER COLUMN c DROP NOT NULL"
- )
-
-
-def test_alter_column_not_nullable():
- context = op_fixture()
- op.alter_column("t", "c", nullable=False)
- context.assert_(
- # TODO: not sure if this is PG only or standard
- # SQL
- "ALTER TABLE t ALTER COLUMN c SET NOT NULL"
- )
-
-
-def test_alter_column_schema_not_nullable():
- context = op_fixture()
- op.alter_column("t", "c", nullable=False, schema='foo')
- context.assert_(
- # TODO: not sure if this is PG only or standard
- # SQL
- "ALTER TABLE foo.t ALTER COLUMN c SET NOT NULL"
- )
-
-
-def test_alter_column_rename():
- context = op_fixture()
- op.alter_column("t", "c", new_column_name="x")
- context.assert_(
- "ALTER TABLE t RENAME c TO x"
- )
-
-
-def test_alter_column_schema_rename():
- context = op_fixture()
- op.alter_column("t", "c", new_column_name="x", schema='foo')
- context.assert_(
- "ALTER TABLE foo.t RENAME c TO x"
- )
-
-
-def test_alter_column_type():
- context = op_fixture()
- op.alter_column("t", "c", type_=String(50))
- context.assert_(
- 'ALTER TABLE t ALTER COLUMN c TYPE VARCHAR(50)'
- )
-
-
-def test_alter_column_schema_type():
- context = op_fixture()
- op.alter_column("t", "c", type_=String(50), schema='foo')
- context.assert_(
- 'ALTER TABLE foo.t ALTER COLUMN c TYPE VARCHAR(50)'
- )
-
-
-def test_alter_column_set_default():
- context = op_fixture()
- op.alter_column("t", "c", server_default="q")
- context.assert_(
- "ALTER TABLE t ALTER COLUMN c SET DEFAULT 'q'"
- )
-
-
-def test_alter_column_schema_set_default():
- context = op_fixture()
- op.alter_column("t", "c", server_default="q", schema='foo')
- context.assert_(
- "ALTER TABLE foo.t ALTER COLUMN c SET DEFAULT 'q'"
- )
-
-
-def test_alter_column_set_compiled_default():
- context = op_fixture()
- op.alter_column("t", "c",
- server_default=func.utc_thing(func.current_timestamp()))
- context.assert_(
- "ALTER TABLE t ALTER COLUMN c SET DEFAULT utc_thing(CURRENT_TIMESTAMP)"
- )
-
-
-def test_alter_column_schema_set_compiled_default():
- context = op_fixture()
- op.alter_column("t", "c",
- server_default=func.utc_thing(func.current_timestamp()),
- schema='foo')
- context.assert_(
- "ALTER TABLE foo.t ALTER COLUMN c "
- "SET DEFAULT utc_thing(CURRENT_TIMESTAMP)"
- )
-
-
-def test_alter_column_drop_default():
- context = op_fixture()
- op.alter_column("t", "c", server_default=None)
- context.assert_(
- 'ALTER TABLE t ALTER COLUMN c DROP DEFAULT'
- )
-
-
-def test_alter_column_schema_drop_default():
- context = op_fixture()
- op.alter_column("t", "c", server_default=None, schema='foo')
- context.assert_(
- 'ALTER TABLE foo.t ALTER COLUMN c DROP DEFAULT'
- )
-
-
-def test_alter_column_schema_type_unnamed():
- context = op_fixture('mssql')
- op.alter_column("t", "c", type_=Boolean())
- context.assert_(
- 'ALTER TABLE t ALTER COLUMN c BIT',
- 'ALTER TABLE t ADD CHECK (c IN (0, 1))'
- )
-
-
-def test_alter_column_schema_schema_type_unnamed():
- context = op_fixture('mssql')
- op.alter_column("t", "c", type_=Boolean(), schema='foo')
- context.assert_(
- 'ALTER TABLE foo.t ALTER COLUMN c BIT',
- 'ALTER TABLE foo.t ADD CHECK (c IN (0, 1))'
- )
-
-
-def test_alter_column_schema_type_named():
- context = op_fixture('mssql')
- op.alter_column("t", "c", type_=Boolean(name="xyz"))
- context.assert_(
- 'ALTER TABLE t ALTER COLUMN c BIT',
- 'ALTER TABLE t ADD CONSTRAINT xyz CHECK (c IN (0, 1))'
- )
-
-
-def test_alter_column_schema_schema_type_named():
- context = op_fixture('mssql')
- op.alter_column("t", "c", type_=Boolean(name="xyz"), schema='foo')
- context.assert_(
- 'ALTER TABLE foo.t ALTER COLUMN c BIT',
- 'ALTER TABLE foo.t ADD CONSTRAINT xyz CHECK (c IN (0, 1))'
- )
-
-
-def test_alter_column_schema_type_existing_type():
- context = op_fixture('mssql')
- op.alter_column(
- "t", "c", type_=String(10), existing_type=Boolean(name="xyz"))
- context.assert_(
- 'ALTER TABLE t DROP CONSTRAINT xyz',
- 'ALTER TABLE t ALTER COLUMN c VARCHAR(10)'
- )
-
-
-def test_alter_column_schema_schema_type_existing_type():
- context = op_fixture('mssql')
- op.alter_column("t", "c", type_=String(10),
- existing_type=Boolean(name="xyz"), schema='foo')
- context.assert_(
- 'ALTER TABLE foo.t DROP CONSTRAINT xyz',
- 'ALTER TABLE foo.t ALTER COLUMN c VARCHAR(10)'
- )
-
-
-def test_alter_column_schema_type_existing_type_no_const():
- context = op_fixture('postgresql')
- op.alter_column("t", "c", type_=String(10), existing_type=Boolean())
- context.assert_(
- 'ALTER TABLE t ALTER COLUMN c TYPE VARCHAR(10)'
- )
-
-
-def test_alter_column_schema_schema_type_existing_type_no_const():
- context = op_fixture('postgresql')
- op.alter_column("t", "c", type_=String(10), existing_type=Boolean(),
- schema='foo')
- context.assert_(
- 'ALTER TABLE foo.t ALTER COLUMN c TYPE VARCHAR(10)'
- )
-
-
-def test_alter_column_schema_type_existing_type_no_new_type():
- context = op_fixture('postgresql')
- op.alter_column("t", "c", nullable=False, existing_type=Boolean())
- context.assert_(
- 'ALTER TABLE t ALTER COLUMN c SET NOT NULL'
- )
-
-
-def test_alter_column_schema_schema_type_existing_type_no_new_type():
- context = op_fixture('postgresql')
- op.alter_column("t", "c", nullable=False, existing_type=Boolean(),
- schema='foo')
- context.assert_(
- 'ALTER TABLE foo.t ALTER COLUMN c SET NOT NULL'
- )
-
-
-def test_add_foreign_key():
- context = op_fixture()
- op.create_foreign_key('fk_test', 't1', 't2',
- ['foo', 'bar'], ['bat', 'hoho'])
- context.assert_(
- "ALTER TABLE t1 ADD CONSTRAINT fk_test FOREIGN KEY(foo, bar) "
- "REFERENCES t2 (bat, hoho)"
- )
-
-
-def test_add_foreign_key_schema():
- context = op_fixture()
- op.create_foreign_key('fk_test', 't1', 't2',
- ['foo', 'bar'], ['bat', 'hoho'],
- source_schema='foo2', referent_schema='bar2')
- context.assert_(
- "ALTER TABLE foo2.t1 ADD CONSTRAINT fk_test FOREIGN KEY(foo, bar) "
- "REFERENCES bar2.t2 (bat, hoho)"
- )
-
-
-def test_add_foreign_key_onupdate():
- context = op_fixture()
- op.create_foreign_key('fk_test', 't1', 't2',
- ['foo', 'bar'], ['bat', 'hoho'],
- onupdate='CASCADE')
- context.assert_(
- "ALTER TABLE t1 ADD CONSTRAINT fk_test FOREIGN KEY(foo, bar) "
- "REFERENCES t2 (bat, hoho) ON UPDATE CASCADE"
- )
-
-
-def test_add_foreign_key_ondelete():
- context = op_fixture()
- op.create_foreign_key('fk_test', 't1', 't2',
- ['foo', 'bar'], ['bat', 'hoho'],
- ondelete='CASCADE')
- context.assert_(
- "ALTER TABLE t1 ADD CONSTRAINT fk_test FOREIGN KEY(foo, bar) "
- "REFERENCES t2 (bat, hoho) ON DELETE CASCADE"
- )
-
-
-def test_add_foreign_key_deferrable():
- context = op_fixture()
- op.create_foreign_key('fk_test', 't1', 't2',
- ['foo', 'bar'], ['bat', 'hoho'],
- deferrable=True)
- context.assert_(
- "ALTER TABLE t1 ADD CONSTRAINT fk_test FOREIGN KEY(foo, bar) "
- "REFERENCES t2 (bat, hoho) DEFERRABLE"
- )
-
-
-def test_add_foreign_key_initially():
- context = op_fixture()
- op.create_foreign_key('fk_test', 't1', 't2',
- ['foo', 'bar'], ['bat', 'hoho'],
- initially='INITIAL')
- context.assert_(
- "ALTER TABLE t1 ADD CONSTRAINT fk_test FOREIGN KEY(foo, bar) "
- "REFERENCES t2 (bat, hoho) INITIALLY INITIAL"
- )
-
-
-def test_add_foreign_key_match():
- context = op_fixture()
- op.create_foreign_key('fk_test', 't1', 't2',
- ['foo', 'bar'], ['bat', 'hoho'],
- match='SIMPLE')
- context.assert_(
- "ALTER TABLE t1 ADD CONSTRAINT fk_test FOREIGN KEY(foo, bar) "
- "REFERENCES t2 (bat, hoho) MATCH SIMPLE"
- )
-
-
-def test_add_foreign_key_dialect_kw():
- context = op_fixture()
- with mock.patch(
- "alembic.operations.sa_schema.ForeignKeyConstraint") as fkc:
+class OpTest(TestBase):
+
+ def test_rename_table(self):
+ context = op_fixture()
+ op.rename_table('t1', 't2')
+ context.assert_("ALTER TABLE t1 RENAME TO t2")
+
+ def test_rename_table_schema(self):
+ context = op_fixture()
+ op.rename_table('t1', 't2', schema="foo")
+ context.assert_("ALTER TABLE foo.t1 RENAME TO foo.t2")
+
+ def test_rename_table_postgresql(self):
+ context = op_fixture("postgresql")
+ op.rename_table('t1', 't2')
+ context.assert_("ALTER TABLE t1 RENAME TO t2")
+
+ def test_rename_table_schema_postgresql(self):
+ context = op_fixture("postgresql")
+ op.rename_table('t1', 't2', schema="foo")
+ context.assert_("ALTER TABLE foo.t1 RENAME TO t2")
+
+ def test_create_index_no_expr_allowed(self):
+ op_fixture()
+ assert_raises_message(
+ ValueError,
+ "String or text\(\) construct expected",
+ op.create_index, 'name', 'tname', [func.foo(column('x'))]
+ )
+
+ def test_create_index_quoting(self):
+ context = op_fixture("postgresql")
+ op.create_index(
+ 'geocoded',
+ 'locations',
+ ["IShouldBeQuoted"])
+ context.assert_(
+ 'CREATE INDEX geocoded ON locations ("IShouldBeQuoted")')
+
+ def test_create_index_expressions(self):
+ context = op_fixture()
+ op.create_index(
+ 'geocoded',
+ 'locations',
+ [text('lower(coordinates)')])
+ context.assert_(
+ "CREATE INDEX geocoded ON locations (lower(coordinates))")
+
+ def test_create_index_postgresql_expressions(self):
+ context = op_fixture("postgresql")
+ op.create_index(
+ 'geocoded',
+ 'locations',
+ [text('lower(coordinates)')],
+ postgresql_where=text("locations.coordinates != Null"))
+ context.assert_(
+ "CREATE INDEX geocoded ON locations (lower(coordinates)) "
+ "WHERE locations.coordinates != Null")
+
+ def test_create_index_postgresql_where(self):
+ context = op_fixture("postgresql")
+ op.create_index(
+ 'geocoded',
+ 'locations',
+ ['coordinates'],
+ postgresql_where=text("locations.coordinates != Null"))
+ context.assert_(
+ "CREATE INDEX geocoded ON locations (coordinates) "
+ "WHERE locations.coordinates != Null")
+
+ def test_add_column(self):
+ context = op_fixture()
+ op.add_column('t1', Column('c1', Integer, nullable=False))
+ context.assert_("ALTER TABLE t1 ADD COLUMN c1 INTEGER NOT NULL")
+
+ def test_add_column_schema(self):
+ context = op_fixture()
+ op.add_column('t1', Column('c1', Integer, nullable=False), schema="foo")
+ context.assert_("ALTER TABLE foo.t1 ADD COLUMN c1 INTEGER NOT NULL")
+
+ def test_add_column_with_default(self):
+ context = op_fixture()
+ op.add_column(
+ 't1', Column('c1', Integer, nullable=False, server_default="12"))
+ context.assert_(
+ "ALTER TABLE t1 ADD COLUMN c1 INTEGER DEFAULT '12' NOT NULL")
+
+ def test_add_column_schema_with_default(self):
+ context = op_fixture()
+ op.add_column('t1',
+ Column('c1', Integer, nullable=False, server_default="12"),
+ schema='foo')
+ context.assert_(
+ "ALTER TABLE foo.t1 ADD COLUMN c1 INTEGER DEFAULT '12' NOT NULL")
+
+ def test_add_column_fk(self):
+ context = op_fixture()
+ op.add_column(
+ 't1', Column('c1', Integer, ForeignKey('c2.id'), nullable=False))
+ context.assert_(
+ "ALTER TABLE t1 ADD COLUMN c1 INTEGER NOT NULL",
+ "ALTER TABLE t1 ADD FOREIGN KEY(c1) REFERENCES c2 (id)"
+ )
+
+ def test_add_column_schema_fk(self):
+ context = op_fixture()
+ op.add_column('t1',
+ Column('c1', Integer, ForeignKey('c2.id'), nullable=False),
+ schema='foo')
+ context.assert_(
+ "ALTER TABLE foo.t1 ADD COLUMN c1 INTEGER NOT NULL",
+ "ALTER TABLE foo.t1 ADD FOREIGN KEY(c1) REFERENCES c2 (id)"
+ )
+
+ def test_add_column_schema_type(self):
+ """Test that a schema type generates its constraints...."""
+ context = op_fixture()
+ op.add_column('t1', Column('c1', Boolean, nullable=False))
+ context.assert_(
+ 'ALTER TABLE t1 ADD COLUMN c1 BOOLEAN NOT NULL',
+ 'ALTER TABLE t1 ADD CHECK (c1 IN (0, 1))'
+ )
+
+ def test_add_column_schema_schema_type(self):
+ """Test that a schema type generates its constraints...."""
+ context = op_fixture()
+ op.add_column('t1', Column('c1', Boolean, nullable=False), schema='foo')
+ context.assert_(
+ 'ALTER TABLE foo.t1 ADD COLUMN c1 BOOLEAN NOT NULL',
+ 'ALTER TABLE foo.t1 ADD CHECK (c1 IN (0, 1))'
+ )
+
+ def test_add_column_schema_type_checks_rule(self):
+ """Test that a schema type doesn't generate a
+ constraint based on check rule."""
+ context = op_fixture('postgresql')
+ op.add_column('t1', Column('c1', Boolean, nullable=False))
+ context.assert_(
+ 'ALTER TABLE t1 ADD COLUMN c1 BOOLEAN NOT NULL',
+ )
+
+ def test_add_column_fk_self_referential(self):
+ context = op_fixture()
+ op.add_column(
+ 't1', Column('c1', Integer, ForeignKey('t1.c2'), nullable=False))
+ context.assert_(
+ "ALTER TABLE t1 ADD COLUMN c1 INTEGER NOT NULL",
+ "ALTER TABLE t1 ADD FOREIGN KEY(c1) REFERENCES t1 (c2)"
+ )
+
+ def test_add_column_schema_fk_self_referential(self):
+ context = op_fixture()
+ op.add_column(
+ 't1',
+ Column('c1', Integer, ForeignKey('foo.t1.c2'), nullable=False),
+ schema='foo')
+ context.assert_(
+ "ALTER TABLE foo.t1 ADD COLUMN c1 INTEGER NOT NULL",
+ "ALTER TABLE foo.t1 ADD FOREIGN KEY(c1) REFERENCES foo.t1 (c2)"
+ )
+
+ def test_add_column_fk_schema(self):
+ context = op_fixture()
+ op.add_column(
+ 't1',
+ Column('c1', Integer, ForeignKey('remote.t2.c2'), nullable=False))
+ context.assert_(
+ 'ALTER TABLE t1 ADD COLUMN c1 INTEGER NOT NULL',
+ 'ALTER TABLE t1 ADD FOREIGN KEY(c1) REFERENCES remote.t2 (c2)'
+ )
+
+ def test_add_column_schema_fk_schema(self):
+ context = op_fixture()
+ op.add_column(
+ 't1',
+ Column('c1', Integer, ForeignKey('remote.t2.c2'), nullable=False),
+ schema='foo')
+ context.assert_(
+ 'ALTER TABLE foo.t1 ADD COLUMN c1 INTEGER NOT NULL',
+ 'ALTER TABLE foo.t1 ADD FOREIGN KEY(c1) REFERENCES remote.t2 (c2)'
+ )
+
+ def test_drop_column(self):
+ context = op_fixture()
+ op.drop_column('t1', 'c1')
+ context.assert_("ALTER TABLE t1 DROP COLUMN c1")
+
+ def test_drop_column_schema(self):
+ context = op_fixture()
+ op.drop_column('t1', 'c1', schema='foo')
+ context.assert_("ALTER TABLE foo.t1 DROP COLUMN c1")
+
+ def test_alter_column_nullable(self):
+ context = op_fixture()
+ op.alter_column("t", "c", nullable=True)
+ context.assert_(
+ # TODO: not sure if this is PG only or standard
+ # SQL
+ "ALTER TABLE t ALTER COLUMN c DROP NOT NULL"
+ )
+
+ def test_alter_column_schema_nullable(self):
+ context = op_fixture()
+ op.alter_column("t", "c", nullable=True, schema='foo')
+ context.assert_(
+ # TODO: not sure if this is PG only or standard
+ # SQL
+ "ALTER TABLE foo.t ALTER COLUMN c DROP NOT NULL"
+ )
+
+ def test_alter_column_not_nullable(self):
+ context = op_fixture()
+ op.alter_column("t", "c", nullable=False)
+ context.assert_(
+ # TODO: not sure if this is PG only or standard
+ # SQL
+ "ALTER TABLE t ALTER COLUMN c SET NOT NULL"
+ )
+
+ def test_alter_column_schema_not_nullable(self):
+ context = op_fixture()
+ op.alter_column("t", "c", nullable=False, schema='foo')
+ context.assert_(
+ # TODO: not sure if this is PG only or standard
+ # SQL
+ "ALTER TABLE foo.t ALTER COLUMN c SET NOT NULL"
+ )
+
+ def test_alter_column_rename(self):
+ context = op_fixture()
+ op.alter_column("t", "c", new_column_name="x")
+ context.assert_(
+ "ALTER TABLE t RENAME c TO x"
+ )
+
+ def test_alter_column_schema_rename(self):
+ context = op_fixture()
+ op.alter_column("t", "c", new_column_name="x", schema='foo')
+ context.assert_(
+ "ALTER TABLE foo.t RENAME c TO x"
+ )
+
+ def test_alter_column_type(self):
+ context = op_fixture()
+ op.alter_column("t", "c", type_=String(50))
+ context.assert_(
+ 'ALTER TABLE t ALTER COLUMN c TYPE VARCHAR(50)'
+ )
+
+ def test_alter_column_schema_type(self):
+ context = op_fixture()
+ op.alter_column("t", "c", type_=String(50), schema='foo')
+ context.assert_(
+ 'ALTER TABLE foo.t ALTER COLUMN c TYPE VARCHAR(50)'
+ )
+
+ def test_alter_column_set_default(self):
+ context = op_fixture()
+ op.alter_column("t", "c", server_default="q")
+ context.assert_(
+ "ALTER TABLE t ALTER COLUMN c SET DEFAULT 'q'"
+ )
+
+ def test_alter_column_schema_set_default(self):
+ context = op_fixture()
+ op.alter_column("t", "c", server_default="q", schema='foo')
+ context.assert_(
+ "ALTER TABLE foo.t ALTER COLUMN c SET DEFAULT 'q'"
+ )
+
+ def test_alter_column_set_compiled_default(self):
+ context = op_fixture()
+ op.alter_column("t", "c",
+ server_default=func.utc_thing(func.current_timestamp()))
+ context.assert_(
+ "ALTER TABLE t ALTER COLUMN c SET DEFAULT utc_thing(CURRENT_TIMESTAMP)"
+ )
+
+ def test_alter_column_schema_set_compiled_default(self):
+ context = op_fixture()
+ op.alter_column("t", "c",
+ server_default=func.utc_thing(func.current_timestamp()),
+ schema='foo')
+ context.assert_(
+ "ALTER TABLE foo.t ALTER COLUMN c "
+ "SET DEFAULT utc_thing(CURRENT_TIMESTAMP)"
+ )
+
+ def test_alter_column_drop_default(self):
+ context = op_fixture()
+ op.alter_column("t", "c", server_default=None)
+ context.assert_(
+ 'ALTER TABLE t ALTER COLUMN c DROP DEFAULT'
+ )
+
+ def test_alter_column_schema_drop_default(self):
+ context = op_fixture()
+ op.alter_column("t", "c", server_default=None, schema='foo')
+ context.assert_(
+ 'ALTER TABLE foo.t ALTER COLUMN c DROP DEFAULT'
+ )
+
+ def test_alter_column_schema_type_unnamed(self):
+ context = op_fixture('mssql')
+ op.alter_column("t", "c", type_=Boolean())
+ context.assert_(
+ 'ALTER TABLE t ALTER COLUMN c BIT',
+ 'ALTER TABLE t ADD CHECK (c IN (0, 1))'
+ )
+
+ def test_alter_column_schema_schema_type_unnamed(self):
+ context = op_fixture('mssql')
+ op.alter_column("t", "c", type_=Boolean(), schema='foo')
+ context.assert_(
+ 'ALTER TABLE foo.t ALTER COLUMN c BIT',
+ 'ALTER TABLE foo.t ADD CHECK (c IN (0, 1))'
+ )
+
+ def test_alter_column_schema_type_named(self):
+ context = op_fixture('mssql')
+ op.alter_column("t", "c", type_=Boolean(name="xyz"))
+ context.assert_(
+ 'ALTER TABLE t ALTER COLUMN c BIT',
+ 'ALTER TABLE t ADD CONSTRAINT xyz CHECK (c IN (0, 1))'
+ )
+
+ def test_alter_column_schema_schema_type_named(self):
+ context = op_fixture('mssql')
+ op.alter_column("t", "c", type_=Boolean(name="xyz"), schema='foo')
+ context.assert_(
+ 'ALTER TABLE foo.t ALTER COLUMN c BIT',
+ 'ALTER TABLE foo.t ADD CONSTRAINT xyz CHECK (c IN (0, 1))'
+ )
+
+ def test_alter_column_schema_type_existing_type(self):
+ context = op_fixture('mssql')
+ op.alter_column(
+ "t", "c", type_=String(10), existing_type=Boolean(name="xyz"))
+ context.assert_(
+ 'ALTER TABLE t DROP CONSTRAINT xyz',
+ 'ALTER TABLE t ALTER COLUMN c VARCHAR(10)'
+ )
+
+ def test_alter_column_schema_schema_type_existing_type(self):
+ context = op_fixture('mssql')
+ op.alter_column("t", "c", type_=String(10),
+ existing_type=Boolean(name="xyz"), schema='foo')
+ context.assert_(
+ 'ALTER TABLE foo.t DROP CONSTRAINT xyz',
+ 'ALTER TABLE foo.t ALTER COLUMN c VARCHAR(10)'
+ )
+
+ def test_alter_column_schema_type_existing_type_no_const(self):
+ context = op_fixture('postgresql')
+ op.alter_column("t", "c", type_=String(10), existing_type=Boolean())
+ context.assert_(
+ 'ALTER TABLE t ALTER COLUMN c TYPE VARCHAR(10)'
+ )
+
+ def test_alter_column_schema_schema_type_existing_type_no_const(self):
+ context = op_fixture('postgresql')
+ op.alter_column("t", "c", type_=String(10), existing_type=Boolean(),
+ schema='foo')
+ context.assert_(
+ 'ALTER TABLE foo.t ALTER COLUMN c TYPE VARCHAR(10)'
+ )
+
+ def test_alter_column_schema_type_existing_type_no_new_type(self):
+ context = op_fixture('postgresql')
+ op.alter_column("t", "c", nullable=False, existing_type=Boolean())
+ context.assert_(
+ 'ALTER TABLE t ALTER COLUMN c SET NOT NULL'
+ )
+
+ def test_alter_column_schema_schema_type_existing_type_no_new_type(self):
+ context = op_fixture('postgresql')
+ op.alter_column("t", "c", nullable=False, existing_type=Boolean(),
+ schema='foo')
+ context.assert_(
+ 'ALTER TABLE foo.t ALTER COLUMN c SET NOT NULL'
+ )
+
+ def test_add_foreign_key(self):
+ context = op_fixture()
+ op.create_foreign_key('fk_test', 't1', 't2',
+ ['foo', 'bar'], ['bat', 'hoho'])
+ context.assert_(
+ "ALTER TABLE t1 ADD CONSTRAINT fk_test FOREIGN KEY(foo, bar) "
+ "REFERENCES t2 (bat, hoho)"
+ )
+
+ def test_add_foreign_key_schema(self):
+ context = op_fixture()
+ op.create_foreign_key('fk_test', 't1', 't2',
+ ['foo', 'bar'], ['bat', 'hoho'],
+ source_schema='foo2', referent_schema='bar2')
+ context.assert_(
+ "ALTER TABLE foo2.t1 ADD CONSTRAINT fk_test FOREIGN KEY(foo, bar) "
+ "REFERENCES bar2.t2 (bat, hoho)"
+ )
+
+ def test_add_foreign_key_onupdate(self):
+ context = op_fixture()
+ op.create_foreign_key('fk_test', 't1', 't2',
+ ['foo', 'bar'], ['bat', 'hoho'],
+ onupdate='CASCADE')
+ context.assert_(
+ "ALTER TABLE t1 ADD CONSTRAINT fk_test FOREIGN KEY(foo, bar) "
+ "REFERENCES t2 (bat, hoho) ON UPDATE CASCADE"
+ )
+
+ def test_add_foreign_key_ondelete(self):
+ context = op_fixture()
+ op.create_foreign_key('fk_test', 't1', 't2',
+ ['foo', 'bar'], ['bat', 'hoho'],
+ ondelete='CASCADE')
+ context.assert_(
+ "ALTER TABLE t1 ADD CONSTRAINT fk_test FOREIGN KEY(foo, bar) "
+ "REFERENCES t2 (bat, hoho) ON DELETE CASCADE"
+ )
+
+ def test_add_foreign_key_deferrable(self):
+ context = op_fixture()
+ op.create_foreign_key('fk_test', 't1', 't2',
+ ['foo', 'bar'], ['bat', 'hoho'],
+ deferrable=True)
+ context.assert_(
+ "ALTER TABLE t1 ADD CONSTRAINT fk_test FOREIGN KEY(foo, bar) "
+ "REFERENCES t2 (bat, hoho) DEFERRABLE"
+ )
+
+ def test_add_foreign_key_initially(self):
+ context = op_fixture()
+ op.create_foreign_key('fk_test', 't1', 't2',
+ ['foo', 'bar'], ['bat', 'hoho'],
+ initially='INITIAL')
+ context.assert_(
+ "ALTER TABLE t1 ADD CONSTRAINT fk_test FOREIGN KEY(foo, bar) "
+ "REFERENCES t2 (bat, hoho) INITIALLY INITIAL"
+ )
+
+ def test_add_foreign_key_match(self):
+ context = op_fixture()
op.create_foreign_key('fk_test', 't1', 't2',
['foo', 'bar'], ['bat', 'hoho'],
- foobar_arg='xyz')
- eq_(fkc.mock_calls[0],
- mock.call(['foo', 'bar'], ['t2.bat', 't2.hoho'],
- onupdate=None, ondelete=None, name='fk_test',
- foobar_arg='xyz',
- deferrable=None, initially=None, match=None))
-
-
-def test_add_foreign_key_self_referential():
- context = op_fixture()
- op.create_foreign_key("fk_test", "t1", "t1", ["foo"], ["bar"])
- context.assert_(
- "ALTER TABLE t1 ADD CONSTRAINT fk_test "
- "FOREIGN KEY(foo) REFERENCES t1 (bar)"
- )
-
-
-def test_add_primary_key_constraint():
- context = op_fixture()
- op.create_primary_key("pk_test", "t1", ["foo", "bar"])
- context.assert_(
- "ALTER TABLE t1 ADD CONSTRAINT pk_test PRIMARY KEY (foo, bar)"
- )
-
-
-def test_add_primary_key_constraint_schema():
- context = op_fixture()
- op.create_primary_key("pk_test", "t1", ["foo"], schema="bar")
- context.assert_(
- "ALTER TABLE bar.t1 ADD CONSTRAINT pk_test PRIMARY KEY (foo)"
- )
-
-
-def test_add_check_constraint():
- context = op_fixture()
- op.create_check_constraint(
- "ck_user_name_len",
- "user_table",
- func.len(column('name')) > 5
- )
- context.assert_(
- "ALTER TABLE user_table ADD CONSTRAINT ck_user_name_len "
- "CHECK (len(name) > 5)"
- )
-
-
-def test_add_check_constraint_schema():
- context = op_fixture()
- op.create_check_constraint(
- "ck_user_name_len",
- "user_table",
- func.len(column('name')) > 5,
- schema='foo'
- )
- context.assert_(
- "ALTER TABLE foo.user_table ADD CONSTRAINT ck_user_name_len "
- "CHECK (len(name) > 5)"
- )
-
-
-def test_add_unique_constraint():
- context = op_fixture()
- op.create_unique_constraint('uk_test', 't1', ['foo', 'bar'])
- context.assert_(
- "ALTER TABLE t1 ADD CONSTRAINT uk_test UNIQUE (foo, bar)"
- )
-
-
-def test_add_unique_constraint_schema():
- context = op_fixture()
- op.create_unique_constraint('uk_test', 't1', ['foo', 'bar'], schema='foo')
- context.assert_(
- "ALTER TABLE foo.t1 ADD CONSTRAINT uk_test UNIQUE (foo, bar)"
- )
-
-
-def test_drop_constraint():
- context = op_fixture()
- op.drop_constraint('foo_bar_bat', 't1')
- context.assert_(
- "ALTER TABLE t1 DROP CONSTRAINT foo_bar_bat"
- )
-
-
-def test_drop_constraint_schema():
- context = op_fixture()
- op.drop_constraint('foo_bar_bat', 't1', schema='foo')
- context.assert_(
- "ALTER TABLE foo.t1 DROP CONSTRAINT foo_bar_bat"
- )
-
-
-def test_create_index():
- context = op_fixture()
- op.create_index('ik_test', 't1', ['foo', 'bar'])
- context.assert_(
- "CREATE INDEX ik_test ON t1 (foo, bar)"
- )
-
-
-def test_create_index_table_col_event():
- context = op_fixture()
-
- op.create_index('ik_test', 'tbl_with_auto_appended_column', ['foo', 'bar'])
- context.assert_(
- "CREATE INDEX ik_test ON tbl_with_auto_appended_column (foo, bar)"
- )
-
-
-def test_add_unique_constraint_col_event():
- context = op_fixture()
- op.create_unique_constraint(
- 'ik_test',
- 'tbl_with_auto_appended_column', ['foo', 'bar'])
- context.assert_(
- "ALTER TABLE tbl_with_auto_appended_column "
- "ADD CONSTRAINT ik_test UNIQUE (foo, bar)"
- )
-
-
-def test_create_index_schema():
- context = op_fixture()
- op.create_index('ik_test', 't1', ['foo', 'bar'], schema='foo')
- context.assert_(
- "CREATE INDEX ik_test ON foo.t1 (foo, bar)"
- )
-
-
-def test_drop_index():
- context = op_fixture()
- op.drop_index('ik_test')
- context.assert_(
- "DROP INDEX ik_test"
- )
-
-
-def test_drop_index_schema():
- context = op_fixture()
- op.drop_index('ik_test', schema='foo')
- context.assert_(
- "DROP INDEX foo.ik_test"
- )
-
-
-def test_drop_table():
- context = op_fixture()
- op.drop_table('tb_test')
- context.assert_(
- "DROP TABLE tb_test"
- )
-
-
-def test_drop_table_schema():
- context = op_fixture()
- op.drop_table('tb_test', schema='foo')
- context.assert_(
- "DROP TABLE foo.tb_test"
- )
-
-
-def test_create_table_selfref():
- context = op_fixture()
- op.create_table(
- "some_table",
- Column('id', Integer, primary_key=True),
- Column('st_id', Integer, ForeignKey('some_table.id'))
- )
- context.assert_(
- "CREATE TABLE some_table ("
- "id INTEGER NOT NULL, "
- "st_id INTEGER, "
- "PRIMARY KEY (id), "
- "FOREIGN KEY(st_id) REFERENCES some_table (id))"
- )
-
-
-def test_create_table_fk_and_schema():
- context = op_fixture()
- op.create_table(
- "some_table",
- Column('id', Integer, primary_key=True),
- Column('foo_id', Integer, ForeignKey('foo.id')),
- schema='schema'
- )
- context.assert_(
- "CREATE TABLE schema.some_table ("
- "id INTEGER NOT NULL, "
- "foo_id INTEGER, "
- "PRIMARY KEY (id), "
- "FOREIGN KEY(foo_id) REFERENCES foo (id))"
- )
-
-
-def test_create_table_no_pk():
- context = op_fixture()
- op.create_table(
- "some_table",
- Column('x', Integer),
- Column('y', Integer),
- Column('z', Integer),
- )
- context.assert_(
- "CREATE TABLE some_table (x INTEGER, y INTEGER, z INTEGER)"
- )
-
-
-def test_create_table_two_fk():
- context = op_fixture()
- op.create_table(
- "some_table",
- Column('id', Integer, primary_key=True),
- Column('foo_id', Integer, ForeignKey('foo.id')),
- Column('foo_bar', Integer, ForeignKey('foo.bar')),
- )
- context.assert_(
- "CREATE TABLE some_table ("
- "id INTEGER NOT NULL, "
- "foo_id INTEGER, "
- "foo_bar INTEGER, "
- "PRIMARY KEY (id), "
- "FOREIGN KEY(foo_id) REFERENCES foo (id), "
- "FOREIGN KEY(foo_bar) REFERENCES foo (bar))"
- )
-
-
-def test_inline_literal():
- context = op_fixture()
- from sqlalchemy.sql import table, column
- from sqlalchemy import String, Integer
-
- account = table('account',
- column('name', String),
- column('id', Integer)
- )
- op.execute(
- account.update().
- where(account.c.name == op.inline_literal('account 1')).
- values({'name': op.inline_literal('account 2')})
- )
- op.execute(
- account.update().
- where(account.c.id == op.inline_literal(1)).
- values({'id': op.inline_literal(2)})
- )
- context.assert_(
- "UPDATE account SET name='account 2' WHERE account.name = 'account 1'",
- "UPDATE account SET id=2 WHERE account.id = 1"
- )
-
-
-def test_cant_op():
- if hasattr(op, '_proxy'):
- del op._proxy
- assert_raises_message(
- NameError,
- "Can't invoke function 'inline_literal', as the "
- "proxy object has not yet been established "
- "for the Alembic 'Operations' class. "
- "Try placing this code inside a callable.",
- op.inline_literal, "asdf"
- )
-
-
-def test_naming_changes():
- context = op_fixture()
- op.alter_column("t", "c", name="x")
- context.assert_("ALTER TABLE t RENAME c TO x")
-
- context = op_fixture()
- op.alter_column("t", "c", new_column_name="x")
- context.assert_("ALTER TABLE t RENAME c TO x")
-
- context = op_fixture('mssql')
- op.drop_index('ik_test', tablename='t1')
- context.assert_("DROP INDEX ik_test ON t1")
-
- context = op_fixture('mysql')
- op.drop_constraint("f1", "t1", type="foreignkey")
- context.assert_("ALTER TABLE t1 DROP FOREIGN KEY f1")
-
- context = op_fixture('mysql')
- op.drop_constraint("f1", "t1", type_="foreignkey")
- context.assert_("ALTER TABLE t1 DROP FOREIGN KEY f1")
-
- assert_raises_message(
- TypeError,
- r"Unknown arguments: badarg\d, badarg\d",
- op.alter_column, "t", "c", badarg1="x", badarg2="y"
- )
+ match='SIMPLE')
+ context.assert_(
+ "ALTER TABLE t1 ADD CONSTRAINT fk_test FOREIGN KEY(foo, bar) "
+ "REFERENCES t2 (bat, hoho) MATCH SIMPLE"
+ )
+
+ def test_add_foreign_key_dialect_kw(self):
+ context = op_fixture()
+ with mock.patch(
+ "alembic.operations.sa_schema.ForeignKeyConstraint") as fkc:
+ op.create_foreign_key('fk_test', 't1', 't2',
+ ['foo', 'bar'], ['bat', 'hoho'],
+ foobar_arg='xyz')
+ eq_(fkc.mock_calls[0],
+ mock.call(['foo', 'bar'], ['t2.bat', 't2.hoho'],
+ onupdate=None, ondelete=None, name='fk_test',
+ foobar_arg='xyz',
+ deferrable=None, initially=None, match=None))
+
+ def test_add_foreign_key_self_referential(self):
+ context = op_fixture()
+ op.create_foreign_key("fk_test", "t1", "t1", ["foo"], ["bar"])
+ context.assert_(
+ "ALTER TABLE t1 ADD CONSTRAINT fk_test "
+ "FOREIGN KEY(foo) REFERENCES t1 (bar)"
+ )
+
+ def test_add_primary_key_constraint(self):
+ context = op_fixture()
+ op.create_primary_key("pk_test", "t1", ["foo", "bar"])
+ context.assert_(
+ "ALTER TABLE t1 ADD CONSTRAINT pk_test PRIMARY KEY (foo, bar)"
+ )
+
+ def test_add_primary_key_constraint_schema(self):
+ context = op_fixture()
+ op.create_primary_key("pk_test", "t1", ["foo"], schema="bar")
+ context.assert_(
+ "ALTER TABLE bar.t1 ADD CONSTRAINT pk_test PRIMARY KEY (foo)"
+ )
+
+ def test_add_check_constraint(self):
+ context = op_fixture()
+ op.create_check_constraint(
+ "ck_user_name_len",
+ "user_table",
+ func.len(column('name')) > 5
+ )
+ context.assert_(
+ "ALTER TABLE user_table ADD CONSTRAINT ck_user_name_len "
+ "CHECK (len(name) > 5)"
+ )
+
+ def test_add_check_constraint_schema(self):
+ context = op_fixture()
+ op.create_check_constraint(
+ "ck_user_name_len",
+ "user_table",
+ func.len(column('name')) > 5,
+ schema='foo'
+ )
+ context.assert_(
+ "ALTER TABLE foo.user_table ADD CONSTRAINT ck_user_name_len "
+ "CHECK (len(name) > 5)"
+ )
+
+ def test_add_unique_constraint(self):
+ context = op_fixture()
+ op.create_unique_constraint('uk_test', 't1', ['foo', 'bar'])
+ context.assert_(
+ "ALTER TABLE t1 ADD CONSTRAINT uk_test UNIQUE (foo, bar)"
+ )
+
+ def test_add_unique_constraint_schema(self):
+ context = op_fixture()
+ op.create_unique_constraint('uk_test', 't1', ['foo', 'bar'], schema='foo')
+ context.assert_(
+ "ALTER TABLE foo.t1 ADD CONSTRAINT uk_test UNIQUE (foo, bar)"
+ )
+
+ def test_drop_constraint(self):
+ context = op_fixture()
+ op.drop_constraint('foo_bar_bat', 't1')
+ context.assert_(
+ "ALTER TABLE t1 DROP CONSTRAINT foo_bar_bat"
+ )
+
+ def test_drop_constraint_schema(self):
+ context = op_fixture()
+ op.drop_constraint('foo_bar_bat', 't1', schema='foo')
+ context.assert_(
+ "ALTER TABLE foo.t1 DROP CONSTRAINT foo_bar_bat"
+ )
+
+ def test_create_index(self):
+ context = op_fixture()
+ op.create_index('ik_test', 't1', ['foo', 'bar'])
+ context.assert_(
+ "CREATE INDEX ik_test ON t1 (foo, bar)"
+ )
+
+ def test_create_index_table_col_event(self):
+ context = op_fixture()
+
+ op.create_index('ik_test', 'tbl_with_auto_appended_column', ['foo', 'bar'])
+ context.assert_(
+ "CREATE INDEX ik_test ON tbl_with_auto_appended_column (foo, bar)"
+ )
+
+ def test_add_unique_constraint_col_event(self):
+ context = op_fixture()
+ op.create_unique_constraint(
+ 'ik_test',
+ 'tbl_with_auto_appended_column', ['foo', 'bar'])
+ context.assert_(
+ "ALTER TABLE tbl_with_auto_appended_column "
+ "ADD CONSTRAINT ik_test UNIQUE (foo, bar)"
+ )
+
+ def test_create_index_schema(self):
+ context = op_fixture()
+ op.create_index('ik_test', 't1', ['foo', 'bar'], schema='foo')
+ context.assert_(
+ "CREATE INDEX ik_test ON foo.t1 (foo, bar)"
+ )
+
+ def test_drop_index(self):
+ context = op_fixture()
+ op.drop_index('ik_test')
+ context.assert_(
+ "DROP INDEX ik_test"
+ )
+
+ def test_drop_index_schema(self):
+ context = op_fixture()
+ op.drop_index('ik_test', schema='foo')
+ context.assert_(
+ "DROP INDEX foo.ik_test"
+ )
+
+ def test_drop_table(self):
+ context = op_fixture()
+ op.drop_table('tb_test')
+ context.assert_(
+ "DROP TABLE tb_test"
+ )
+
+ def test_drop_table_schema(self):
+ context = op_fixture()
+ op.drop_table('tb_test', schema='foo')
+ context.assert_(
+ "DROP TABLE foo.tb_test"
+ )
+
+ def test_create_table_selfref(self):
+ context = op_fixture()
+ op.create_table(
+ "some_table",
+ Column('id', Integer, primary_key=True),
+ Column('st_id', Integer, ForeignKey('some_table.id'))
+ )
+ context.assert_(
+ "CREATE TABLE some_table ("
+ "id INTEGER NOT NULL, "
+ "st_id INTEGER, "
+ "PRIMARY KEY (id), "
+ "FOREIGN KEY(st_id) REFERENCES some_table (id))"
+ )
+
+ def test_create_table_fk_and_schema(self):
+ context = op_fixture()
+ op.create_table(
+ "some_table",
+ Column('id', Integer, primary_key=True),
+ Column('foo_id', Integer, ForeignKey('foo.id')),
+ schema='schema'
+ )
+ context.assert_(
+ "CREATE TABLE schema.some_table ("
+ "id INTEGER NOT NULL, "
+ "foo_id INTEGER, "
+ "PRIMARY KEY (id), "
+ "FOREIGN KEY(foo_id) REFERENCES foo (id))"
+ )
+
+ def test_create_table_no_pk(self):
+ context = op_fixture()
+ op.create_table(
+ "some_table",
+ Column('x', Integer),
+ Column('y', Integer),
+ Column('z', Integer),
+ )
+ context.assert_(
+ "CREATE TABLE some_table (x INTEGER, y INTEGER, z INTEGER)"
+ )
+
+ def test_create_table_two_fk(self):
+ context = op_fixture()
+ op.create_table(
+ "some_table",
+ Column('id', Integer, primary_key=True),
+ Column('foo_id', Integer, ForeignKey('foo.id')),
+ Column('foo_bar', Integer, ForeignKey('foo.bar')),
+ )
+ context.assert_(
+ "CREATE TABLE some_table ("
+ "id INTEGER NOT NULL, "
+ "foo_id INTEGER, "
+ "foo_bar INTEGER, "
+ "PRIMARY KEY (id), "
+ "FOREIGN KEY(foo_id) REFERENCES foo (id), "
+ "FOREIGN KEY(foo_bar) REFERENCES foo (bar))"
+ )
+
+ def test_inline_literal(self):
+ context = op_fixture()
+ from sqlalchemy.sql import table, column
+ from sqlalchemy import String, Integer
+
+ account = table('account',
+ column('name', String),
+ column('id', Integer)
+ )
+ op.execute(
+ account.update().
+ where(account.c.name == op.inline_literal('account 1')).
+ values({'name': op.inline_literal('account 2')})
+ )
+ op.execute(
+ account.update().
+ where(account.c.id == op.inline_literal(1)).
+ values({'id': op.inline_literal(2)})
+ )
+ context.assert_(
+ "UPDATE account SET name='account 2' WHERE account.name = 'account 1'",
+ "UPDATE account SET id=2 WHERE account.id = 1"
+ )
+
+ def test_cant_op(self):
+ if hasattr(op, '_proxy'):
+ del op._proxy
+ assert_raises_message(
+ NameError,
+ "Can't invoke function 'inline_literal', as the "
+ "proxy object has not yet been established "
+ "for the Alembic 'Operations' class. "
+ "Try placing this code inside a callable.",
+ op.inline_literal, "asdf"
+ )
+
+ def test_naming_changes(self):
+ context = op_fixture()
+ op.alter_column("t", "c", name="x")
+ context.assert_("ALTER TABLE t RENAME c TO x")
+
+ context = op_fixture()
+ op.alter_column("t", "c", new_column_name="x")
+ context.assert_("ALTER TABLE t RENAME c TO x")
+
+ context = op_fixture('mssql')
+ op.drop_index('ik_test', tablename='t1')
+ context.assert_("DROP INDEX ik_test ON t1")
+
+ context = op_fixture('mysql')
+ op.drop_constraint("f1", "t1", type="foreignkey")
+ context.assert_("ALTER TABLE t1 DROP FOREIGN KEY f1")
+
+ context = op_fixture('mysql')
+ op.drop_constraint("f1", "t1", type_="foreignkey")
+ context.assert_("ALTER TABLE t1 DROP FOREIGN KEY f1")
+
+ assert_raises_message(
+ TypeError,
+ r"Unknown arguments: badarg\d, badarg\d",
+ op.alter_column, "t", "c", badarg1="x", badarg2="y"
+ )
diff --git a/tests/test_op_naming_convention.py b/tests/test_op_naming_convention.py
index 342609d..fd70faa 100644
--- a/tests/test_op_naming_convention.py
+++ b/tests/test_op_naming_convention.py
@@ -1,174 +1,157 @@
-from sqlalchemy import Integer, Column, ForeignKey, \
- Table, String, Boolean, MetaData, CheckConstraint
-from sqlalchemy.sql import column, func, text
-from sqlalchemy import event
+from sqlalchemy import Integer, Column, \
+ Table, Boolean, MetaData, CheckConstraint
+from sqlalchemy.sql import column, func
from alembic import op
-from . import op_fixture, assert_raises_message, requires_094
-
-
-@requires_094
-def test_add_check_constraint():
- context = op_fixture(naming_convention={
- "ck": "ck_%(table_name)s_%(constraint_name)s"
- })
- op.create_check_constraint(
- "foo",
- "user_table",
- func.len(column('name')) > 5
- )
- context.assert_(
- "ALTER TABLE user_table ADD CONSTRAINT ck_user_table_foo "
- "CHECK (len(name) > 5)"
- )
-
-
-@requires_094
-def test_add_check_constraint_name_is_none():
- context = op_fixture(naming_convention={
- "ck": "ck_%(table_name)s_foo"
- })
- op.create_check_constraint(
- None,
- "user_table",
- func.len(column('name')) > 5
- )
- context.assert_(
- "ALTER TABLE user_table ADD CONSTRAINT ck_user_table_foo "
- "CHECK (len(name) > 5)"
- )
-
-
-@requires_094
-def test_add_unique_constraint_name_is_none():
- context = op_fixture(naming_convention={
- "uq": "uq_%(table_name)s_foo"
- })
- op.create_unique_constraint(
- None,
- "user_table",
- 'x'
- )
- context.assert_(
- "ALTER TABLE user_table ADD CONSTRAINT uq_user_table_foo UNIQUE (x)"
- )
-
-
-@requires_094
-def test_add_index_name_is_none():
- context = op_fixture(naming_convention={
- "ix": "ix_%(table_name)s_foo"
- })
- op.create_index(
- None,
- "user_table",
- 'x'
- )
- context.assert_(
- "CREATE INDEX ix_user_table_foo ON user_table (x)"
- )
-
-
-@requires_094
-def test_add_check_constraint_already_named_from_schema():
- 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)
-
- context = op_fixture(
- naming_convention={"ck": "ck_%(table_name)s_%(constraint_name)s"})
-
- op.create_table(
- "some_table",
- Column('x', Integer, ck),
- )
- context.assert_(
- "CREATE TABLE some_table "
- "(x INTEGER CONSTRAINT ck_t_cc1 CHECK (im a constraint))"
- )
-
-
-@requires_094
-def test_add_check_constraint_inline_on_table():
- context = op_fixture(
- naming_convention={"ck": "ck_%(table_name)s_%(constraint_name)s"})
- op.create_table(
- "some_table",
- Column('x', Integer),
- CheckConstraint("im a constraint", name="cc1")
- )
- context.assert_(
- "CREATE TABLE some_table "
- "(x INTEGER, CONSTRAINT ck_some_table_cc1 CHECK (im a constraint))"
- )
-
-
-@requires_094
-def test_add_check_constraint_inline_on_table_w_f():
- context = op_fixture(
- naming_convention={"ck": "ck_%(table_name)s_%(constraint_name)s"})
- op.create_table(
- "some_table",
- Column('x', Integer),
- CheckConstraint("im a constraint", name=op.f("ck_some_table_cc1"))
- )
- context.assert_(
- "CREATE TABLE some_table "
- "(x INTEGER, CONSTRAINT ck_some_table_cc1 CHECK (im a constraint))"
- )
-
-
-@requires_094
-def test_add_check_constraint_inline_on_column():
- context = op_fixture(
- naming_convention={"ck": "ck_%(table_name)s_%(constraint_name)s"})
- op.create_table(
- "some_table",
- Column('x', Integer, CheckConstraint("im a constraint", name="cc1"))
- )
- context.assert_(
- "CREATE TABLE some_table "
- "(x INTEGER CONSTRAINT ck_some_table_cc1 CHECK (im a constraint))"
- )
-
-
-@requires_094
-def test_add_check_constraint_inline_on_column_w_f():
- context = op_fixture(
- naming_convention={"ck": "ck_%(table_name)s_%(constraint_name)s"})
- op.create_table(
- "some_table",
- Column(
- 'x', Integer,
- CheckConstraint("im a constraint", name=op.f("ck_q_cc1")))
- )
- context.assert_(
- "CREATE TABLE some_table "
- "(x INTEGER CONSTRAINT ck_q_cc1 CHECK (im a constraint))"
- )
-
-
-@requires_094
-def test_add_column_schema_type():
- context = op_fixture(naming_convention={
- "ck": "ck_%(table_name)s_%(constraint_name)s"
- })
- op.add_column('t1', Column('c1', Boolean(name='foo'), nullable=False))
- context.assert_(
- 'ALTER TABLE t1 ADD COLUMN c1 BOOLEAN NOT NULL',
- 'ALTER TABLE t1 ADD CONSTRAINT ck_t1_foo CHECK (c1 IN (0, 1))'
- )
-
-
-@requires_094
-def test_add_column_schema_type_w_f():
- context = op_fixture(naming_convention={
- "ck": "ck_%(table_name)s_%(constraint_name)s"
- })
- op.add_column(
- 't1', Column('c1', Boolean(name=op.f('foo')), nullable=False))
- context.assert_(
- 'ALTER TABLE t1 ADD COLUMN c1 BOOLEAN NOT NULL',
- 'ALTER TABLE t1 ADD CONSTRAINT foo CHECK (c1 IN (0, 1))'
- )
+
+from alembic.testing.fixtures import op_fixture
+from alembic.testing.fixtures import TestBase
+
+
+class AutoNamingConventionTest(TestBase):
+ __requires__ = ('sqlalchemy_094', )
+
+ def test_add_check_constraint(self):
+ context = op_fixture(naming_convention={
+ "ck": "ck_%(table_name)s_%(constraint_name)s"
+ })
+ op.create_check_constraint(
+ "foo",
+ "user_table",
+ func.len(column('name')) > 5
+ )
+ context.assert_(
+ "ALTER TABLE user_table ADD CONSTRAINT ck_user_table_foo "
+ "CHECK (len(name) > 5)"
+ )
+
+ def test_add_check_constraint_name_is_none(self):
+ context = op_fixture(naming_convention={
+ "ck": "ck_%(table_name)s_foo"
+ })
+ op.create_check_constraint(
+ None,
+ "user_table",
+ func.len(column('name')) > 5
+ )
+ context.assert_(
+ "ALTER TABLE user_table ADD CONSTRAINT ck_user_table_foo "
+ "CHECK (len(name) > 5)"
+ )
+
+ def test_add_unique_constraint_name_is_none(self):
+ context = op_fixture(naming_convention={
+ "uq": "uq_%(table_name)s_foo"
+ })
+ op.create_unique_constraint(
+ None,
+ "user_table",
+ 'x'
+ )
+ context.assert_(
+ "ALTER TABLE user_table ADD CONSTRAINT uq_user_table_foo UNIQUE (x)"
+ )
+
+ def test_add_index_name_is_none(self):
+ context = op_fixture(naming_convention={
+ "ix": "ix_%(table_name)s_foo"
+ })
+ op.create_index(
+ None,
+ "user_table",
+ 'x'
+ )
+ context.assert_(
+ "CREATE INDEX ix_user_table_foo ON user_table (x)"
+ )
+
+ def test_add_check_constraint_already_named_from_schema(self):
+ 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)
+
+ context = op_fixture(
+ naming_convention={"ck": "ck_%(table_name)s_%(constraint_name)s"})
+
+ op.create_table(
+ "some_table",
+ Column('x', Integer, ck),
+ )
+ context.assert_(
+ "CREATE TABLE some_table "
+ "(x INTEGER CONSTRAINT ck_t_cc1 CHECK (im a constraint))"
+ )
+
+ def test_add_check_constraint_inline_on_table(self):
+ context = op_fixture(
+ naming_convention={"ck": "ck_%(table_name)s_%(constraint_name)s"})
+ op.create_table(
+ "some_table",
+ Column('x', Integer),
+ CheckConstraint("im a constraint", name="cc1")
+ )
+ context.assert_(
+ "CREATE TABLE some_table "
+ "(x INTEGER, CONSTRAINT ck_some_table_cc1 CHECK (im a constraint))"
+ )
+
+ def test_add_check_constraint_inline_on_table_w_f(self):
+ context = op_fixture(
+ naming_convention={"ck": "ck_%(table_name)s_%(constraint_name)s"})
+ op.create_table(
+ "some_table",
+ Column('x', Integer),
+ CheckConstraint("im a constraint", name=op.f("ck_some_table_cc1"))
+ )
+ context.assert_(
+ "CREATE TABLE some_table "
+ "(x INTEGER, CONSTRAINT ck_some_table_cc1 CHECK (im a constraint))"
+ )
+
+ def test_add_check_constraint_inline_on_column(self):
+ context = op_fixture(
+ naming_convention={"ck": "ck_%(table_name)s_%(constraint_name)s"})
+ op.create_table(
+ "some_table",
+ Column('x', Integer, CheckConstraint("im a constraint", name="cc1"))
+ )
+ context.assert_(
+ "CREATE TABLE some_table "
+ "(x INTEGER CONSTRAINT ck_some_table_cc1 CHECK (im a constraint))"
+ )
+
+ def test_add_check_constraint_inline_on_column_w_f(self):
+ context = op_fixture(
+ naming_convention={"ck": "ck_%(table_name)s_%(constraint_name)s"})
+ op.create_table(
+ "some_table",
+ Column(
+ 'x', Integer,
+ CheckConstraint("im a constraint", name=op.f("ck_q_cc1")))
+ )
+ context.assert_(
+ "CREATE TABLE some_table "
+ "(x INTEGER CONSTRAINT ck_q_cc1 CHECK (im a constraint))"
+ )
+
+ def test_add_column_schema_type(self):
+ context = op_fixture(naming_convention={
+ "ck": "ck_%(table_name)s_%(constraint_name)s"
+ })
+ op.add_column('t1', Column('c1', Boolean(name='foo'), nullable=False))
+ context.assert_(
+ 'ALTER TABLE t1 ADD COLUMN c1 BOOLEAN NOT NULL',
+ 'ALTER TABLE t1 ADD CONSTRAINT ck_t1_foo CHECK (c1 IN (0, 1))'
+ )
+
+ def test_add_column_schema_type_w_f(self):
+ context = op_fixture(naming_convention={
+ "ck": "ck_%(table_name)s_%(constraint_name)s"
+ })
+ op.add_column(
+ 't1', Column('c1', Boolean(name=op.f('foo')), nullable=False))
+ context.assert_(
+ 'ALTER TABLE t1 ADD COLUMN c1 BOOLEAN NOT NULL',
+ 'ALTER TABLE t1 ADD CONSTRAINT foo CHECK (c1 IN (0, 1))'
+ )
diff --git a/tox.ini b/tox.ini
index d602bb0..0f007a3 100644
--- a/tox.ini
+++ b/tox.ini
@@ -30,6 +30,7 @@ commands = python -m flake8 {posargs}
[flake8]
show-source = True
-ignore = E711,E712,E721,F841,F811,F401
+ignore = E711,E712,E721
+# F841,F811,F401
exclude=.venv,.git,.tox,dist,doc,*egg,build