from sqlalchemy.testing.requirements import Requirements
from alembic import util
-from alembic.testing import exclusions
from alembic.util import sqla_compat
+from ..testing import exclusions
class SuiteRequirements(Requirements):
exclusions.BooleanPredicate(sqla_compat.has_computed)
)
+ @property
+ def computed_reflects_normally(self):
+ return exclusions.only_if(
+ exclusions.BooleanPredicate(sqla_compat.has_computed_reflection)
+ )
+
+ @property
+ def computed_reflects_as_server_default(self):
+ return exclusions.closed()
+
+ @property
+ def computed_doesnt_reflect_as_server_default(self):
+ return exclusions.closed()
+
+ @property
+ def autoincrement_on_composite_pk(self):
+ return exclusions.closed()
+
+ @property
+ def fk_ondelete_is_reflected(self):
+ return exclusions.closed()
+
+ @property
+ def fk_onupdate_is_reflected(self):
+ return exclusions.closed()
+
+ @property
+ def fk_onupdate(self):
+ return exclusions.open()
+
+ @property
+ def fk_ondelete_restrict(self):
+ return exclusions.open()
+
+ @property
+ def fk_onupdate_restrict(self):
+ return exclusions.open()
+
+ @property
+ def fk_ondelete_noaction(self):
+ return exclusions.open()
+
+ @property
+ def fk_initially(self):
+ return exclusions.closed()
+
+ @property
+ def fk_deferrable(self):
+ return exclusions.closed()
+
+ @property
+ def fk_deferrable_is_reflected(self):
+ return exclusions.closed()
+
+ @property
+ def fk_names(self):
+ return exclusions.open()
+
+ @property
+ def integer_subtype_comparisons(self):
+ return exclusions.open()
+
+ @property
+ def no_name_normalize(self):
+ return exclusions.skip_if(
+ lambda config: config.db.dialect.requires_name_normalize
+ )
+
@property
def identity_columns(self):
return exclusions.closed()
--- /dev/null
+from .test_autogen_comments import * # noqa
+from .test_autogen_computed import * # noqa
+from .test_autogen_diffs import * # noqa
+from .test_autogen_fks import * # noqa
+from .test_autogen_identity import * # noqa
+from .test_environment import * # noqa
+from .test_op import * # noqa
from sqlalchemy import text
from sqlalchemy import UniqueConstraint
-from alembic import autogenerate
-from alembic import util
-from alembic.autogenerate import api
-from alembic.ddl.base import _fk_spec
-from alembic.migration import MigrationContext
-from alembic.operations import ops
-from alembic.testing import config
-from alembic.testing import eq_
-from alembic.testing.env import clear_staging_env
-from alembic.testing.env import staging_env
+from ... import autogenerate
+from ... import util
+from ...autogenerate import api
+from ...ddl.base import _fk_spec
+from ...migration import MigrationContext
+from ...operations import ops
+from ...testing import config
+from ...testing import eq_
+from ...testing.env import clear_staging_env
+from ...testing.env import staging_env
names_in_this_test = set()
-import sys
-
from sqlalchemy import Column
from sqlalchemy import Float
from sqlalchemy import MetaData
from sqlalchemy import String
from sqlalchemy import Table
-from alembic.testing import eq_
-from alembic.testing import mock
-from alembic.testing import TestBase
from ._autogen_fixtures import AutogenFixtureTest
+from ...testing import eq_
+from ...testing import mock
+from ...testing import TestBase
+from ...util import compat
-py3k = sys.version_info.major >= 3
+py3k = compat.py3k
class AutogenerateCommentsTest(AutogenFixtureTest, TestBase):
from sqlalchemy import MetaData
from sqlalchemy import Table
-from alembic import testing
-from alembic.testing import config
-from alembic.testing import eq_
-from alembic.testing import exclusions
-from alembic.testing import is_
-from alembic.testing import is_true
-from alembic.testing import mock
-from alembic.testing import TestBase
from ._autogen_fixtures import AutogenFixtureTest
+from ... import testing
+from ...testing import config
+from ...testing import eq_
+from ...testing import exclusions
+from ...testing import is_
+from ...testing import is_true
+from ...testing import mock
+from ...testing import TestBase
class AutogenerateComputedTest(AutogenFixtureTest, TestBase):
--- /dev/null
+from sqlalchemy import BigInteger
+from sqlalchemy import Column
+from sqlalchemy import Integer
+from sqlalchemy import MetaData
+from sqlalchemy import Table
+from sqlalchemy.testing import in_
+
+from ._autogen_fixtures import AutogenFixtureTest
+from ... import testing
+from ...testing import config
+from ...testing import eq_
+from ...testing import is_
+from ...testing import TestBase
+from ...util import compat
+
+py3k = compat.py3k
+
+
+class AlterColumnTest(AutogenFixtureTest, TestBase):
+ __backend__ = True
+
+ @testing.combinations((True,), (False,))
+ @config.requirements.comments
+ def test_all_existings_filled(self, pk):
+ m1 = MetaData()
+ m2 = MetaData()
+
+ Table("a", m1, Column("x", Integer, primary_key=pk))
+ Table("a", m2, Column("x", Integer, comment="x", primary_key=pk))
+
+ alter_col = self._assert_alter_col(m1, m2, pk)
+ eq_(alter_col.modify_comment, "x")
+
+ @testing.combinations((True,), (False,))
+ @config.requirements.comments
+ def test_all_existings_filled_in_notnull(self, pk):
+ m1 = MetaData()
+ m2 = MetaData()
+
+ Table("a", m1, Column("x", Integer, nullable=False, primary_key=pk))
+ Table(
+ "a",
+ m2,
+ Column("x", Integer, nullable=False, comment="x", primary_key=pk),
+ )
+
+ self._assert_alter_col(m1, m2, pk, nullable=False)
+
+ @testing.combinations((True,), (False,))
+ @config.requirements.comments
+ def test_all_existings_filled_in_comment(self, pk):
+ m1 = MetaData()
+ m2 = MetaData()
+
+ Table("a", m1, Column("x", Integer, comment="old", primary_key=pk))
+ Table("a", m2, Column("x", Integer, comment="new", primary_key=pk))
+
+ alter_col = self._assert_alter_col(m1, m2, pk)
+ eq_(alter_col.existing_comment, "old")
+
+ @testing.combinations((True,), (False,))
+ @config.requirements.comments
+ def test_all_existings_filled_in_server_default(self, pk):
+ m1 = MetaData()
+ m2 = MetaData()
+
+ Table(
+ "a", m1, Column("x", Integer, server_default="5", primary_key=pk)
+ )
+ Table(
+ "a",
+ m2,
+ Column(
+ "x", Integer, server_default="5", comment="new", primary_key=pk
+ ),
+ )
+
+ alter_col = self._assert_alter_col(m1, m2, pk)
+ in_("5", alter_col.existing_server_default.arg.text)
+
+ def _assert_alter_col(self, m1, m2, pk, nullable=None):
+ ops = self._fixture(m1, m2, return_ops=True)
+ modify_table = ops.ops[-1]
+ alter_col = modify_table.ops[0]
+
+ if nullable is None:
+ eq_(alter_col.existing_nullable, not pk)
+ else:
+ eq_(alter_col.existing_nullable, nullable)
+ assert alter_col.existing_type._compare_type_affinity(Integer())
+ return alter_col
+
+
+class AutoincrementTest(AutogenFixtureTest, TestBase):
+ __backend__ = True
+ __requires__ = ("integer_subtype_comparisons",)
+
+ def test_alter_column_autoincrement_none(self):
+ m1 = MetaData()
+ m2 = MetaData()
+
+ Table("a", m1, Column("x", Integer, nullable=False))
+ Table("a", m2, Column("x", Integer, nullable=True))
+
+ ops = self._fixture(m1, m2, return_ops=True)
+ assert "autoincrement" not in ops.ops[0].ops[0].kw
+
+ def test_alter_column_autoincrement_pk_false(self):
+ m1 = MetaData()
+ m2 = MetaData()
+
+ Table(
+ "a",
+ m1,
+ Column("x", Integer, primary_key=True, autoincrement=False),
+ )
+ Table(
+ "a",
+ m2,
+ Column("x", BigInteger, primary_key=True, autoincrement=False),
+ )
+
+ ops = self._fixture(m1, m2, return_ops=True)
+ is_(ops.ops[0].ops[0].kw["autoincrement"], False)
+
+ def test_alter_column_autoincrement_pk_implicit_true(self):
+ m1 = MetaData()
+ m2 = MetaData()
+
+ Table("a", m1, Column("x", Integer, primary_key=True))
+ Table("a", m2, Column("x", BigInteger, primary_key=True))
+
+ ops = self._fixture(m1, m2, return_ops=True)
+ is_(ops.ops[0].ops[0].kw["autoincrement"], True)
+
+ def test_alter_column_autoincrement_pk_explicit_true(self):
+ m1 = MetaData()
+ m2 = MetaData()
+
+ Table(
+ "a", m1, Column("x", Integer, primary_key=True, autoincrement=True)
+ )
+ Table(
+ "a",
+ m2,
+ Column("x", BigInteger, primary_key=True, autoincrement=True),
+ )
+
+ ops = self._fixture(m1, m2, return_ops=True)
+ is_(ops.ops[0].ops[0].kw["autoincrement"], True)
+
+ def test_alter_column_autoincrement_nonpk_false(self):
+ m1 = MetaData()
+ m2 = MetaData()
+
+ Table(
+ "a",
+ m1,
+ Column("id", Integer, primary_key=True),
+ Column("x", Integer, autoincrement=False),
+ )
+ Table(
+ "a",
+ m2,
+ Column("id", Integer, primary_key=True),
+ Column("x", BigInteger, autoincrement=False),
+ )
+
+ ops = self._fixture(m1, m2, return_ops=True)
+ is_(ops.ops[0].ops[0].kw["autoincrement"], False)
+
+ def test_alter_column_autoincrement_nonpk_implicit_false(self):
+ m1 = MetaData()
+ m2 = MetaData()
+
+ Table(
+ "a",
+ m1,
+ Column("id", Integer, primary_key=True),
+ Column("x", Integer),
+ )
+ Table(
+ "a",
+ m2,
+ Column("id", Integer, primary_key=True),
+ Column("x", BigInteger),
+ )
+
+ ops = self._fixture(m1, m2, return_ops=True)
+ assert "autoincrement" not in ops.ops[0].ops[0].kw
+
+ def test_alter_column_autoincrement_nonpk_explicit_true(self):
+ m1 = MetaData()
+ m2 = MetaData()
+
+ Table(
+ "a",
+ m1,
+ Column("id", Integer, primary_key=True, autoincrement=False),
+ Column("x", Integer, autoincrement=True),
+ )
+ Table(
+ "a",
+ m2,
+ Column("id", Integer, primary_key=True, autoincrement=False),
+ Column("x", BigInteger, autoincrement=True),
+ )
+
+ ops = self._fixture(m1, m2, return_ops=True)
+ is_(ops.ops[0].ops[0].kw["autoincrement"], True)
+
+ def test_alter_column_autoincrement_compositepk_false(self):
+ m1 = MetaData()
+ m2 = MetaData()
+
+ Table(
+ "a",
+ m1,
+ Column("id", Integer, primary_key=True),
+ Column("x", Integer, primary_key=True, autoincrement=False),
+ )
+ Table(
+ "a",
+ m2,
+ Column("id", Integer, primary_key=True),
+ Column("x", BigInteger, primary_key=True, autoincrement=False),
+ )
+
+ ops = self._fixture(m1, m2, return_ops=True)
+ is_(ops.ops[0].ops[0].kw["autoincrement"], False)
+
+ def test_alter_column_autoincrement_compositepk_implicit_false(self):
+ m1 = MetaData()
+ m2 = MetaData()
+
+ Table(
+ "a",
+ m1,
+ Column("id", Integer, primary_key=True),
+ Column("x", Integer, primary_key=True),
+ )
+ Table(
+ "a",
+ m2,
+ Column("id", Integer, primary_key=True),
+ Column("x", BigInteger, primary_key=True),
+ )
+
+ ops = self._fixture(m1, m2, return_ops=True)
+ assert "autoincrement" not in ops.ops[0].ops[0].kw
+
+ @config.requirements.autoincrement_on_composite_pk
+ def test_alter_column_autoincrement_compositepk_explicit_true(self):
+ m1 = MetaData()
+ m2 = MetaData()
+
+ Table(
+ "a",
+ m1,
+ Column("id", Integer, primary_key=True, autoincrement=False),
+ Column("x", Integer, primary_key=True, autoincrement=True),
+ # on SQLA 1.0 and earlier, this being present
+ # trips the "add KEY for the primary key" so that the
+ # AUTO_INCREMENT keyword is accepted by MySQL. SQLA 1.1 and
+ # greater the columns are just reorganized.
+ mysql_engine="InnoDB",
+ )
+ Table(
+ "a",
+ m2,
+ Column("id", Integer, primary_key=True, autoincrement=False),
+ Column("x", BigInteger, primary_key=True, autoincrement=True),
+ )
+
+ ops = self._fixture(m1, m2, return_ops=True)
+ is_(ops.ops[0].ops[0].kw["autoincrement"], True)
-import sys
-
from sqlalchemy import Column
from sqlalchemy import ForeignKeyConstraint
from sqlalchemy import Integer
from sqlalchemy import String
from sqlalchemy import Table
-from alembic.testing import combinations
-from alembic.testing import config
-from alembic.testing import eq_
-from alembic.testing import mock
-from alembic.testing import TestBase
from ._autogen_fixtures import AutogenFixtureTest
+from ...testing import combinations
+from ...testing import config
+from ...testing import eq_
+from ...testing import mock
+from ...testing import TestBase
+from ...util import compat
-py3k = sys.version_info.major >= 3
+py3k = compat.py3k
class AutogenerateForeignKeysTest(AutogenFixtureTest, TestBase):
"some_table",
m1,
Column("test", String(10), primary_key=True),
- mysql_engine="InnoDB",
)
Table(
Column("a1", String(10), server_default="x"),
Column("test2", String(10)),
ForeignKeyConstraint(["test2"], ["some_table.test"]),
- mysql_engine="InnoDB",
)
Table(
"some_table",
m2,
Column("test", String(10), primary_key=True),
- mysql_engine="InnoDB",
)
Table(
Column("name", String(50), nullable=False),
Column("a1", String(10), server_default="x"),
Column("test2", String(10)),
- mysql_engine="InnoDB",
)
diffs = self._fixture(m1, m2)
m1,
Column("id", Integer, primary_key=True),
Column("test", String(10)),
- mysql_engine="InnoDB",
)
Table(
Column("name", String(50), nullable=False),
Column("a1", String(10), server_default="x"),
Column("test2", String(10)),
- mysql_engine="InnoDB",
)
Table(
m2,
Column("id", Integer, primary_key=True),
Column("test", String(10)),
- mysql_engine="InnoDB",
)
Table(
Column("a1", String(10), server_default="x"),
Column("test2", String(10)),
ForeignKeyConstraint(["test2"], ["some_table.test"]),
- mysql_engine="InnoDB",
)
diffs = self._fixture(m1, m2)
m1,
Column("id", Integer, primary_key=True),
Column("test", String(10)),
- mysql_engine="InnoDB",
)
Table(
Column("a1", String(10), server_default="x"),
Column("test2", Integer),
ForeignKeyConstraint(["test2"], ["some_table.id"]),
- mysql_engine="InnoDB",
)
Table(
m2,
Column("id", Integer, primary_key=True),
Column("test", String(10)),
- mysql_engine="InnoDB",
)
Table(
Column("a1", String(10), server_default="x"),
Column("test2", Integer),
ForeignKeyConstraint(["test2"], ["some_table.id"]),
- mysql_engine="InnoDB",
)
diffs = self._fixture(m1, m2)
m1,
Column("id_1", String(10), primary_key=True),
Column("id_2", String(10), primary_key=True),
- mysql_engine="InnoDB",
)
Table(
["other_id_1", "other_id_2"],
["some_table.id_1", "some_table.id_2"],
),
- mysql_engine="InnoDB",
)
Table(
m2,
Column("id_1", String(10), primary_key=True),
Column("id_2", String(10), primary_key=True),
- mysql_engine="InnoDB",
)
Table(
["other_id_1", "other_id_2"],
["some_table.id_1", "some_table.id_2"],
),
- mysql_engine="InnoDB",
)
diffs = self._fixture(m1, m2)
"some_table",
m1,
Column("test", String(10), primary_key=True),
- mysql_engine="InnoDB",
)
Table(
Column("a1", String(10), server_default="x"),
Column("test2", String(10)),
ForeignKeyConstraint(["test2"], ["some_table.test"], name="MyFK"),
- mysql_engine="InnoDB",
)
Table(
"some_table",
m2,
Column("test", String(10), primary_key=True),
- mysql_engine="InnoDB",
)
# foreign key autogen currently does not take "name" into account,
Column("a1", String(10), server_default="x"),
Column("test2", String(10)),
ForeignKeyConstraint(["a1"], ["some_table.test"], name="myfk"),
- mysql_engine="InnoDB",
)
diffs = self._fixture(m1, m2)
m1,
Column("id_1", String(10), primary_key=True),
Column("id_2", String(10), primary_key=True),
- mysql_engine="InnoDB",
)
Table(
Column("a1", String(10), server_default="x"),
Column("other_id_1", String(10)),
Column("other_id_2", String(10)),
- mysql_engine="InnoDB",
)
Table(
m2,
Column("id_1", String(10), primary_key=True),
Column("id_2", String(10), primary_key=True),
- mysql_engine="InnoDB",
)
Table(
["some_table.id_1", "some_table.id_2"],
name="fk_test_name",
),
- mysql_engine="InnoDB",
)
diffs = self._fixture(m1, m2)
m1,
Column("id_1", String(10), primary_key=True),
Column("id_2", String(10), primary_key=True),
- mysql_engine="InnoDB",
)
Table(
["some_table.id_1", "some_table.id_2"],
name="fk_test_name",
),
- mysql_engine="InnoDB",
)
Table(
m2,
Column("id_1", String(10), primary_key=True),
Column("id_2", String(10), primary_key=True),
- mysql_engine="InnoDB",
)
Table(
Column("a1", String(10), server_default="x"),
Column("other_id_1", String(10)),
Column("other_id_2", String(10)),
- mysql_engine="InnoDB",
)
diffs = self._fixture(m1, m2)
m1,
Column("id_1", String(10), primary_key=True),
Column("id_2", String(10), primary_key=True),
- mysql_engine="InnoDB",
)
Table(
Column("id", Integer, primary_key=True),
Column("other_id_1", String(10)),
Column("other_id_2", String(10)),
- mysql_engine="InnoDB",
)
Table(
m2,
Column("id_1", String(10), key="tid1", primary_key=True),
Column("id_2", String(10), key="tid2", primary_key=True),
- mysql_engine="InnoDB",
)
Table(
["some_table.tid1", "some_table.tid2"],
name="fk_test_name",
),
- mysql_engine="InnoDB",
)
diffs = self._fixture(m1, m2)
m1,
Column("id_1", String(10), primary_key=True),
Column("id_2", String(10), primary_key=True),
- mysql_engine="InnoDB",
)
Table(
["other_id_1", "other_id_2"],
["some_table.id_1", "some_table.id_2"],
),
- mysql_engine="InnoDB",
)
Table(
m2,
Column("id_1", String(10), key="tid1", primary_key=True),
Column("id_2", String(10), key="tid2", primary_key=True),
- mysql_engine="InnoDB",
)
Table(
ForeignKeyConstraint(
["oid1", "oid2"], ["some_table.tid1", "some_table.tid2"]
),
- mysql_engine="InnoDB",
)
diffs = self._fixture(m1, m2)
"ref",
m1,
Column("id", Integer, primary_key=True),
- mysql_engine="InnoDB",
)
t1 = Table(
"t",
m1,
Column("x", Integer),
Column("y", Integer),
- mysql_engine="InnoDB",
)
t1.append_constraint(
ForeignKeyConstraint([t1.c.x], [ref.c.id], name="fk1")
"ref",
m2,
Column("id", Integer, primary_key=True),
- mysql_engine="InnoDB",
)
Table(
"t",
m2,
Column("x", Integer),
Column("y", Integer),
- mysql_engine="InnoDB",
)
if hook_type == "object":
"ref",
m1,
Column("id", Integer, primary_key=True),
- mysql_engine="InnoDB",
)
Table(
"t",
m1,
Column("x", Integer),
Column("y", Integer),
- mysql_engine="InnoDB",
)
ref = Table(
"ref",
m2,
Column("id", Integer, primary_key=True),
- mysql_engine="InnoDB",
)
t2 = Table(
"t",
m2,
Column("x", Integer),
Column("y", Integer),
- mysql_engine="InnoDB",
)
t2.append_constraint(
ForeignKeyConstraint([t2.c.x], [ref.c.id], name="fk1")
"ref_a",
m1,
Column("a", Integer, primary_key=True),
- mysql_engine="InnoDB",
)
Table(
"ref_b",
m1,
Column("a", Integer, primary_key=True),
Column("b", Integer, primary_key=True),
- mysql_engine="InnoDB",
)
t1 = Table(
"t",
Column("x", Integer),
Column("y", Integer),
Column("z", Integer),
- mysql_engine="InnoDB",
)
t1.append_constraint(
ForeignKeyConstraint([t1.c.x], [r1a.c.a], name="fk1")
"ref_a",
m2,
Column("a", Integer, primary_key=True),
- mysql_engine="InnoDB",
)
r2b = Table(
"ref_b",
m2,
Column("a", Integer, primary_key=True),
Column("b", Integer, primary_key=True),
- mysql_engine="InnoDB",
)
t2 = Table(
"t",
Column("x", Integer),
Column("y", Integer),
Column("z", Integer),
- mysql_engine="InnoDB",
)
t2.append_constraint(
ForeignKeyConstraint(
m1,
Column("id", Integer, primary_key=True),
Column("test", String(10)),
- mysql_engine="InnoDB",
)
Table(
Column("name", String(50), nullable=False),
Column("tid", Integer),
ForeignKeyConstraint(["tid"], ["some_table.id"], **old_opts),
- mysql_engine="InnoDB",
)
Table(
m2,
Column("id", Integer, primary_key=True),
Column("test", String(10)),
- mysql_engine="InnoDB",
)
Table(
Column("name", String(50), nullable=False),
Column("tid", Integer),
ForeignKeyConstraint(["tid"], ["some_table.id"], **new_opts),
- mysql_engine="InnoDB",
)
return self._fixture(m1, m2)
from sqlalchemy import MetaData
from sqlalchemy import Table
-from alembic import testing
-from alembic.testing import config
-from alembic.testing import eq_
-from alembic.testing import is_true
-from alembic.testing import TestBase
from ._autogen_fixtures import AutogenFixtureTest
+from ... import testing
+from ...testing import config
+from ...testing import eq_
+from ...testing import is_true
+from ...testing import TestBase
class AutogenerateIdentityTest(AutogenFixtureTest, TestBase):
--- /dev/null
+#!coding: utf-8
+from ...migration import MigrationContext
+from ...testing import assert_raises
+from ...testing import config
+from ...testing import eq_
+from ...testing import is_false
+from ...testing import is_true
+from ...testing.fixtures import TestBase
+from ...util import compat
+
+
+class MigrationTransactionTest(TestBase):
+ __backend__ = True
+
+ conn = None
+
+ def _fixture(self, opts):
+ self.conn = conn = config.db.connect()
+
+ if opts.get("as_sql", False):
+ self.context = MigrationContext.configure(
+ dialect=conn.dialect, opts=opts
+ )
+ self.context.output_buffer = (
+ self.context.impl.output_buffer
+ ) = compat.StringIO()
+ else:
+ self.context = MigrationContext.configure(
+ connection=conn, opts=opts
+ )
+ return self.context
+
+ def teardown(self):
+ if self.conn:
+ self.conn.close()
+
+ def test_proxy_transaction_rollback(self):
+ context = self._fixture(
+ {"transaction_per_migration": True, "transactional_ddl": True}
+ )
+
+ is_false(self.conn.in_transaction())
+ proxy = context.begin_transaction(_per_migration=True)
+ is_true(self.conn.in_transaction())
+ proxy.rollback()
+ is_false(self.conn.in_transaction())
+
+ def test_proxy_transaction_commit(self):
+ context = self._fixture(
+ {"transaction_per_migration": True, "transactional_ddl": True}
+ )
+ proxy = context.begin_transaction(_per_migration=True)
+ is_true(self.conn.in_transaction())
+ proxy.commit()
+ is_false(self.conn.in_transaction())
+
+ def test_proxy_transaction_contextmanager_commit(self):
+ context = self._fixture(
+ {"transaction_per_migration": True, "transactional_ddl": True}
+ )
+ proxy = context.begin_transaction(_per_migration=True)
+ is_true(self.conn.in_transaction())
+ with proxy:
+ pass
+ is_false(self.conn.in_transaction())
+
+ def test_proxy_transaction_contextmanager_rollback(self):
+ context = self._fixture(
+ {"transaction_per_migration": True, "transactional_ddl": True}
+ )
+ proxy = context.begin_transaction(_per_migration=True)
+ is_true(self.conn.in_transaction())
+
+ def go():
+ with proxy:
+ raise Exception("hi")
+
+ assert_raises(Exception, go)
+ is_false(self.conn.in_transaction())
+
+ def test_proxy_transaction_contextmanager_explicit_rollback(self):
+ context = self._fixture(
+ {"transaction_per_migration": True, "transactional_ddl": True}
+ )
+ proxy = context.begin_transaction(_per_migration=True)
+ is_true(self.conn.in_transaction())
+
+ with proxy:
+ is_true(self.conn.in_transaction())
+ proxy.rollback()
+ is_false(self.conn.in_transaction())
+
+ is_false(self.conn.in_transaction())
+
+ def test_proxy_transaction_contextmanager_explicit_commit(self):
+ context = self._fixture(
+ {"transaction_per_migration": True, "transactional_ddl": True}
+ )
+ proxy = context.begin_transaction(_per_migration=True)
+ is_true(self.conn.in_transaction())
+
+ with proxy:
+ is_true(self.conn.in_transaction())
+ proxy.commit()
+ is_false(self.conn.in_transaction())
+
+ is_false(self.conn.in_transaction())
+
+ def test_transaction_per_migration_transactional_ddl(self):
+ context = self._fixture(
+ {"transaction_per_migration": True, "transactional_ddl": True}
+ )
+
+ is_false(self.conn.in_transaction())
+
+ with context.begin_transaction():
+ is_false(self.conn.in_transaction())
+ with context.begin_transaction(_per_migration=True):
+ is_true(self.conn.in_transaction())
+
+ is_false(self.conn.in_transaction())
+ is_false(self.conn.in_transaction())
+
+ def test_transaction_per_migration_non_transactional_ddl(self):
+ context = self._fixture(
+ {"transaction_per_migration": True, "transactional_ddl": False}
+ )
+
+ is_false(self.conn.in_transaction())
+
+ with context.begin_transaction():
+ is_false(self.conn.in_transaction())
+ with context.begin_transaction(_per_migration=True):
+ is_true(self.conn.in_transaction())
+
+ is_false(self.conn.in_transaction())
+ is_false(self.conn.in_transaction())
+
+ def test_transaction_per_all_transactional_ddl(self):
+ context = self._fixture({"transactional_ddl": True})
+
+ is_false(self.conn.in_transaction())
+
+ with context.begin_transaction():
+ is_true(self.conn.in_transaction())
+ with context.begin_transaction(_per_migration=True):
+ is_true(self.conn.in_transaction())
+
+ is_true(self.conn.in_transaction())
+ is_false(self.conn.in_transaction())
+
+ def test_transaction_per_all_non_transactional_ddl(self):
+ context = self._fixture({"transactional_ddl": False})
+
+ is_false(self.conn.in_transaction())
+
+ with context.begin_transaction():
+ is_false(self.conn.in_transaction())
+ with context.begin_transaction(_per_migration=True):
+ is_true(self.conn.in_transaction())
+
+ is_false(self.conn.in_transaction())
+ is_false(self.conn.in_transaction())
+
+ def test_transaction_per_all_sqlmode(self):
+ context = self._fixture({"as_sql": True})
+
+ context.execute("step 1")
+ with context.begin_transaction():
+ context.execute("step 2")
+ with context.begin_transaction(_per_migration=True):
+ context.execute("step 3")
+
+ context.execute("step 4")
+ context.execute("step 5")
+
+ if context.impl.transactional_ddl:
+ self._assert_impl_steps(
+ "step 1",
+ "BEGIN",
+ "step 2",
+ "step 3",
+ "step 4",
+ "COMMIT",
+ "step 5",
+ )
+ else:
+ self._assert_impl_steps(
+ "step 1", "step 2", "step 3", "step 4", "step 5"
+ )
+
+ def test_transaction_per_migration_sqlmode(self):
+ context = self._fixture(
+ {"as_sql": True, "transaction_per_migration": True}
+ )
+
+ context.execute("step 1")
+ with context.begin_transaction():
+ context.execute("step 2")
+ with context.begin_transaction(_per_migration=True):
+ context.execute("step 3")
+
+ context.execute("step 4")
+ context.execute("step 5")
+
+ if context.impl.transactional_ddl:
+ self._assert_impl_steps(
+ "step 1",
+ "step 2",
+ "BEGIN",
+ "step 3",
+ "COMMIT",
+ "step 4",
+ "step 5",
+ )
+ else:
+ self._assert_impl_steps(
+ "step 1", "step 2", "step 3", "step 4", "step 5"
+ )
+
+ @config.requirements.autocommit_isolation
+ def test_autocommit_block(self):
+ context = self._fixture({"transaction_per_migration": True})
+
+ is_false(self.conn.in_transaction())
+
+ with context.begin_transaction():
+ is_false(self.conn.in_transaction())
+ with context.begin_transaction(_per_migration=True):
+ is_true(self.conn.in_transaction())
+
+ with context.autocommit_block():
+ is_false(self.conn.in_transaction())
+
+ is_true(self.conn.in_transaction())
+
+ is_false(self.conn.in_transaction())
+ is_false(self.conn.in_transaction())
+
+ @config.requirements.autocommit_isolation
+ def test_autocommit_block_no_transaction(self):
+ context = self._fixture({"transaction_per_migration": True})
+
+ is_false(self.conn.in_transaction())
+
+ with context.autocommit_block():
+ is_false(self.conn.in_transaction())
+ is_false(self.conn.in_transaction())
+
+ def test_autocommit_block_transactional_ddl_sqlmode(self):
+ context = self._fixture(
+ {
+ "transaction_per_migration": True,
+ "transactional_ddl": True,
+ "as_sql": True,
+ }
+ )
+
+ with context.begin_transaction():
+ context.execute("step 1")
+ with context.begin_transaction(_per_migration=True):
+ context.execute("step 2")
+
+ with context.autocommit_block():
+ context.execute("step 3")
+
+ context.execute("step 4")
+
+ context.execute("step 5")
+
+ self._assert_impl_steps(
+ "step 1",
+ "BEGIN",
+ "step 2",
+ "COMMIT",
+ "step 3",
+ "BEGIN",
+ "step 4",
+ "COMMIT",
+ "step 5",
+ )
+
+ def test_autocommit_block_nontransactional_ddl_sqlmode(self):
+ context = self._fixture(
+ {
+ "transaction_per_migration": True,
+ "transactional_ddl": False,
+ "as_sql": True,
+ }
+ )
+
+ with context.begin_transaction():
+ context.execute("step 1")
+ with context.begin_transaction(_per_migration=True):
+ context.execute("step 2")
+
+ with context.autocommit_block():
+ context.execute("step 3")
+
+ context.execute("step 4")
+
+ context.execute("step 5")
+
+ self._assert_impl_steps(
+ "step 1", "step 2", "step 3", "step 4", "step 5"
+ )
+
+ def _assert_impl_steps(self, *steps):
+ to_check = self.context.output_buffer.getvalue()
+
+ self.context.impl.output_buffer = buf = compat.StringIO()
+ for step in steps:
+ if step == "BEGIN":
+ self.context.impl.emit_begin()
+ elif step == "COMMIT":
+ self.context.impl.emit_commit()
+ else:
+ self.context.impl._exec(step)
+
+ eq_(to_check, buf.getvalue())
--- /dev/null
+"""Test against the builders in the op.* module."""
+
+from sqlalchemy import Column
+from sqlalchemy import event
+from sqlalchemy import Integer
+from sqlalchemy import String
+from sqlalchemy import Table
+from sqlalchemy.sql import text
+
+from ...testing.fixtures import AlterColRoundTripFixture
+from ...testing.fixtures import TestBase
+
+
+@event.listens_for(Table, "after_parent_attach")
+def _add_cols(table, metadata):
+ if table.name == "tbl_with_auto_appended_column":
+ table.append_column(Column("bat", Integer))
+
+
+class BackendAlterColumnTest(AlterColRoundTripFixture, TestBase):
+ __backend__ = True
+
+ def test_rename_column(self):
+ self._run_alter_col({}, {"name": "newname"})
+
+ def test_modify_type_int_str(self):
+ self._run_alter_col({"type": Integer()}, {"type": String(50)})
+
+ def test_add_server_default_int(self):
+ self._run_alter_col({"type": Integer}, {"server_default": text("5")})
+
+ def test_modify_server_default_int(self):
+ self._run_alter_col(
+ {"type": Integer, "server_default": text("2")},
+ {"server_default": text("5")},
+ )
+
+ def test_modify_nullable_to_non(self):
+ self._run_alter_col({}, {"nullable": False})
+
+ def test_modify_non_nullable_to_nullable(self):
+ self._run_alter_col({"nullable": False}, {"nullable": True})
"""foreign key constraints always have names in the DB"""
return exclusions.fails_on("sqlite")
- @property
- def no_name_normalize(self):
- return exclusions.skip_if(
- lambda config: config.db.dialect.requires_name_normalize
- )
-
@property
def reflects_fk_options(self):
return exclusions.open()
["postgresql < 12", "sqlite < 3.31", "mysql < 5.7"]
)
- @property
- def computed_reflects_normally(self):
- return exclusions.only_if(
- exclusions.BooleanPredicate(sqla_compat.has_computed_reflection)
- )
-
@property
def computed_reflects_as_server_default(self):
# note that this rule will go away when SQLAlchemy correctly
from alembic.migration import MigrationContext
from alembic.testing import eq_
from alembic.testing import TestBase
-from ._autogen_fixtures import _default_include_object
-from ._autogen_fixtures import AutogenTest
-from ._autogen_fixtures import ModelOne
+from alembic.testing.suite._autogen_fixtures import _default_include_object
+from alembic.testing.suite._autogen_fixtures import AutogenTest
+from alembic.testing.suite._autogen_fixtures import ModelOne
class AutogenerateDiffTest(ModelOne, AutogenTest, TestBase):
from sqlalchemy import VARCHAR
from sqlalchemy.dialects import mysql
from sqlalchemy.dialects import sqlite
-from sqlalchemy.testing import in_
from sqlalchemy.types import NULLTYPE
from sqlalchemy.types import VARBINARY
from alembic.testing import TestBase
from alembic.testing.env import clear_staging_env
from alembic.testing.env import staging_env
+from alembic.testing.suite._autogen_fixtures import _default_name_filters
+from alembic.testing.suite._autogen_fixtures import _default_object_filters
+from alembic.testing.suite._autogen_fixtures import AutogenFixtureTest
+from alembic.testing.suite._autogen_fixtures import AutogenTest
from alembic.util import CommandError
-from ._autogen_fixtures import _default_name_filters
-from ._autogen_fixtures import _default_object_filters
-from ._autogen_fixtures import AutogenFixtureTest
-from ._autogen_fixtures import AutogenTest
+
+# TODO: we should make an adaptation of CompareMetadataToInspectorTest that is
+# more well suited towards generic backends (2021-06-10)
py3k = sys.version_info >= (3,)
eq_(diffs, [])
-class AlterColumnTest(AutogenFixtureTest, TestBase):
- __backend__ = True
-
- @testing.combinations((True,), (False,))
- @config.requirements.comments
- def test_all_existings_filled(self, pk):
- m1 = MetaData()
- m2 = MetaData()
-
- Table("a", m1, Column("x", Integer, primary_key=pk))
- Table("a", m2, Column("x", Integer, comment="x", primary_key=pk))
-
- alter_col = self._assert_alter_col(m1, m2, pk)
- eq_(alter_col.modify_comment, "x")
-
- @testing.combinations((True,), (False,))
- @config.requirements.comments
- def test_all_existings_filled_in_notnull(self, pk):
- m1 = MetaData()
- m2 = MetaData()
-
- Table("a", m1, Column("x", Integer, nullable=False, primary_key=pk))
- Table(
- "a",
- m2,
- Column("x", Integer, nullable=False, comment="x", primary_key=pk),
- )
-
- self._assert_alter_col(m1, m2, pk, nullable=False)
-
- @testing.combinations((True,), (False,))
- @config.requirements.comments
- def test_all_existings_filled_in_comment(self, pk):
- m1 = MetaData()
- m2 = MetaData()
-
- Table("a", m1, Column("x", Integer, comment="old", primary_key=pk))
- Table("a", m2, Column("x", Integer, comment="new", primary_key=pk))
-
- alter_col = self._assert_alter_col(m1, m2, pk)
- eq_(alter_col.existing_comment, "old")
-
- @testing.combinations((True,), (False,))
- @config.requirements.comments
- def test_all_existings_filled_in_server_default(self, pk):
- m1 = MetaData()
- m2 = MetaData()
-
- Table(
- "a", m1, Column("x", Integer, server_default="5", primary_key=pk)
- )
- Table(
- "a",
- m2,
- Column(
- "x", Integer, server_default="5", comment="new", primary_key=pk
- ),
- )
-
- alter_col = self._assert_alter_col(m1, m2, pk)
- in_("5", alter_col.existing_server_default.arg.text)
-
- def _assert_alter_col(self, m1, m2, pk, nullable=None):
- ops = self._fixture(m1, m2, return_ops=True)
- modify_table = ops.ops[-1]
- alter_col = modify_table.ops[0]
-
- if nullable is None:
- eq_(alter_col.existing_nullable, not pk)
- else:
- eq_(alter_col.existing_nullable, nullable)
- assert alter_col.existing_type._compare_type_affinity(Integer())
- return alter_col
-
-
class AutogenKeyTest(AutogenTest, TestBase):
__only_on__ = "sqlite"
[m1a, m1b],
[m2a, m2b],
)
-
-
-class AutoincrementTest(AutogenFixtureTest, TestBase):
- __backend__ = True
- __requires__ = ("integer_subtype_comparisons",)
-
- def test_alter_column_autoincrement_none(self):
- m1 = MetaData()
- m2 = MetaData()
-
- Table("a", m1, Column("x", Integer, nullable=False))
- Table("a", m2, Column("x", Integer, nullable=True))
-
- ops = self._fixture(m1, m2, return_ops=True)
- assert "autoincrement" not in ops.ops[0].ops[0].kw
-
- def test_alter_column_autoincrement_pk_false(self):
- m1 = MetaData()
- m2 = MetaData()
-
- Table(
- "a",
- m1,
- Column("x", Integer, primary_key=True, autoincrement=False),
- )
- Table(
- "a",
- m2,
- Column("x", BigInteger, primary_key=True, autoincrement=False),
- )
-
- ops = self._fixture(m1, m2, return_ops=True)
- is_(ops.ops[0].ops[0].kw["autoincrement"], False)
-
- def test_alter_column_autoincrement_pk_implicit_true(self):
- m1 = MetaData()
- m2 = MetaData()
-
- Table("a", m1, Column("x", Integer, primary_key=True))
- Table("a", m2, Column("x", BigInteger, primary_key=True))
-
- ops = self._fixture(m1, m2, return_ops=True)
- is_(ops.ops[0].ops[0].kw["autoincrement"], True)
-
- def test_alter_column_autoincrement_pk_explicit_true(self):
- m1 = MetaData()
- m2 = MetaData()
-
- Table(
- "a", m1, Column("x", Integer, primary_key=True, autoincrement=True)
- )
- Table(
- "a",
- m2,
- Column("x", BigInteger, primary_key=True, autoincrement=True),
- )
-
- ops = self._fixture(m1, m2, return_ops=True)
- is_(ops.ops[0].ops[0].kw["autoincrement"], True)
-
- def test_alter_column_autoincrement_nonpk_false(self):
- m1 = MetaData()
- m2 = MetaData()
-
- Table(
- "a",
- m1,
- Column("id", Integer, primary_key=True),
- Column("x", Integer, autoincrement=False),
- )
- Table(
- "a",
- m2,
- Column("id", Integer, primary_key=True),
- Column("x", BigInteger, autoincrement=False),
- )
-
- ops = self._fixture(m1, m2, return_ops=True)
- is_(ops.ops[0].ops[0].kw["autoincrement"], False)
-
- def test_alter_column_autoincrement_nonpk_implicit_false(self):
- m1 = MetaData()
- m2 = MetaData()
-
- Table(
- "a",
- m1,
- Column("id", Integer, primary_key=True),
- Column("x", Integer),
- )
- Table(
- "a",
- m2,
- Column("id", Integer, primary_key=True),
- Column("x", BigInteger),
- )
-
- ops = self._fixture(m1, m2, return_ops=True)
- assert "autoincrement" not in ops.ops[0].ops[0].kw
-
- def test_alter_column_autoincrement_nonpk_explicit_true(self):
- m1 = MetaData()
- m2 = MetaData()
-
- Table(
- "a",
- m1,
- Column("id", Integer, primary_key=True, autoincrement=False),
- Column("x", Integer, autoincrement=True),
- )
- Table(
- "a",
- m2,
- Column("id", Integer, primary_key=True, autoincrement=False),
- Column("x", BigInteger, autoincrement=True),
- )
-
- ops = self._fixture(m1, m2, return_ops=True)
- is_(ops.ops[0].ops[0].kw["autoincrement"], True)
-
- def test_alter_column_autoincrement_compositepk_false(self):
- m1 = MetaData()
- m2 = MetaData()
-
- Table(
- "a",
- m1,
- Column("id", Integer, primary_key=True),
- Column("x", Integer, primary_key=True, autoincrement=False),
- )
- Table(
- "a",
- m2,
- Column("id", Integer, primary_key=True),
- Column("x", BigInteger, primary_key=True, autoincrement=False),
- )
-
- ops = self._fixture(m1, m2, return_ops=True)
- is_(ops.ops[0].ops[0].kw["autoincrement"], False)
-
- def test_alter_column_autoincrement_compositepk_implicit_false(self):
- m1 = MetaData()
- m2 = MetaData()
-
- Table(
- "a",
- m1,
- Column("id", Integer, primary_key=True),
- Column("x", Integer, primary_key=True),
- )
- Table(
- "a",
- m2,
- Column("id", Integer, primary_key=True),
- Column("x", BigInteger, primary_key=True),
- )
-
- ops = self._fixture(m1, m2, return_ops=True)
- assert "autoincrement" not in ops.ops[0].ops[0].kw
-
- @config.requirements.autoincrement_on_composite_pk
- def test_alter_column_autoincrement_compositepk_explicit_true(self):
- m1 = MetaData()
- m2 = MetaData()
-
- Table(
- "a",
- m1,
- Column("id", Integer, primary_key=True, autoincrement=False),
- Column("x", Integer, primary_key=True, autoincrement=True),
- # on SQLA 1.0 and earlier, this being present
- # trips the "add KEY for the primary key" so that the
- # AUTO_INCREMENT keyword is accepted by MySQL. SQLA 1.1 and
- # greater the columns are just reorganized.
- mysql_engine="InnoDB",
- )
- Table(
- "a",
- m2,
- Column("id", Integer, primary_key=True, autoincrement=False),
- Column("x", BigInteger, primary_key=True, autoincrement=True),
- )
-
- ops = self._fixture(m1, m2, return_ops=True)
- is_(ops.ops[0].ops[0].kw["autoincrement"], True)
from alembic.testing import TestBase
from alembic.testing import util
from alembic.testing.env import staging_env
+from alembic.testing.suite._autogen_fixtures import AutogenFixtureTest
from alembic.util import sqla_compat
-from ._autogen_fixtures import AutogenFixtureTest
+
+# TODO: create new suites that are taking tests from this suite, with a
+# separate class for AutogenIndexes, AutogenUniqueConstraint, and a
+# subset of the tests here. @zzzeek can work on this at a later point.
+# (2021-06-10)
py3k = sys.version_info >= (3,)
from alembic.environment import EnvironmentContext
from alembic.migration import MigrationContext
from alembic.script import ScriptDirectory
-from alembic.testing import assert_raises
from alembic.testing import config
from alembic.testing import eq_
from alembic.testing import is_
-from alembic.testing import is_false
-from alembic.testing import is_true
from alembic.testing import mock
from alembic.testing.assertions import expect_raises_message
from alembic.testing.env import _get_staging_directory
from alembic.testing.env import write_script
from alembic.testing.fixtures import capture_context_buffer
from alembic.testing.fixtures import TestBase
-from alembic.util import compat
class EnvironmentTest(TestBase):
env.configure(url="sqlite://", fn=assert_)
with env:
script.run_env()
-
-
-class MigrationTransactionTest(TestBase):
- __backend__ = True
-
- conn = None
-
- def _fixture(self, opts):
- self.conn = conn = config.db.connect()
-
- if opts.get("as_sql", False):
- self.context = MigrationContext.configure(
- dialect=conn.dialect, opts=opts
- )
- self.context.output_buffer = (
- self.context.impl.output_buffer
- ) = compat.StringIO()
- else:
- self.context = MigrationContext.configure(
- connection=conn, opts=opts
- )
- return self.context
-
- def teardown(self):
- if self.conn:
- self.conn.close()
-
- def test_proxy_transaction_rollback(self):
- context = self._fixture(
- {"transaction_per_migration": True, "transactional_ddl": True}
- )
-
- is_false(self.conn.in_transaction())
- proxy = context.begin_transaction(_per_migration=True)
- is_true(self.conn.in_transaction())
- proxy.rollback()
- is_false(self.conn.in_transaction())
-
- def test_proxy_transaction_commit(self):
- context = self._fixture(
- {"transaction_per_migration": True, "transactional_ddl": True}
- )
- proxy = context.begin_transaction(_per_migration=True)
- is_true(self.conn.in_transaction())
- proxy.commit()
- is_false(self.conn.in_transaction())
-
- def test_proxy_transaction_contextmanager_commit(self):
- context = self._fixture(
- {"transaction_per_migration": True, "transactional_ddl": True}
- )
- proxy = context.begin_transaction(_per_migration=True)
- is_true(self.conn.in_transaction())
- with proxy:
- pass
- is_false(self.conn.in_transaction())
-
- def test_proxy_transaction_contextmanager_rollback(self):
- context = self._fixture(
- {"transaction_per_migration": True, "transactional_ddl": True}
- )
- proxy = context.begin_transaction(_per_migration=True)
- is_true(self.conn.in_transaction())
-
- def go():
- with proxy:
- raise Exception("hi")
-
- assert_raises(Exception, go)
- is_false(self.conn.in_transaction())
-
- def test_proxy_transaction_contextmanager_explicit_rollback(self):
- context = self._fixture(
- {"transaction_per_migration": True, "transactional_ddl": True}
- )
- proxy = context.begin_transaction(_per_migration=True)
- is_true(self.conn.in_transaction())
-
- with proxy:
- is_true(self.conn.in_transaction())
- proxy.rollback()
- is_false(self.conn.in_transaction())
-
- is_false(self.conn.in_transaction())
-
- def test_proxy_transaction_contextmanager_explicit_commit(self):
- context = self._fixture(
- {"transaction_per_migration": True, "transactional_ddl": True}
- )
- proxy = context.begin_transaction(_per_migration=True)
- is_true(self.conn.in_transaction())
-
- with proxy:
- is_true(self.conn.in_transaction())
- proxy.commit()
- is_false(self.conn.in_transaction())
-
- is_false(self.conn.in_transaction())
-
- def test_transaction_per_migration_transactional_ddl(self):
- context = self._fixture(
- {"transaction_per_migration": True, "transactional_ddl": True}
- )
-
- is_false(self.conn.in_transaction())
-
- with context.begin_transaction():
- is_false(self.conn.in_transaction())
- with context.begin_transaction(_per_migration=True):
- is_true(self.conn.in_transaction())
-
- is_false(self.conn.in_transaction())
- is_false(self.conn.in_transaction())
-
- def test_transaction_per_migration_non_transactional_ddl(self):
- context = self._fixture(
- {"transaction_per_migration": True, "transactional_ddl": False}
- )
-
- is_false(self.conn.in_transaction())
-
- with context.begin_transaction():
- is_false(self.conn.in_transaction())
- with context.begin_transaction(_per_migration=True):
- is_true(self.conn.in_transaction())
-
- is_false(self.conn.in_transaction())
- is_false(self.conn.in_transaction())
-
- def test_transaction_per_all_transactional_ddl(self):
- context = self._fixture({"transactional_ddl": True})
-
- is_false(self.conn.in_transaction())
-
- with context.begin_transaction():
- is_true(self.conn.in_transaction())
- with context.begin_transaction(_per_migration=True):
- is_true(self.conn.in_transaction())
-
- is_true(self.conn.in_transaction())
- is_false(self.conn.in_transaction())
-
- def test_transaction_per_all_non_transactional_ddl(self):
- context = self._fixture({"transactional_ddl": False})
-
- is_false(self.conn.in_transaction())
-
- with context.begin_transaction():
- is_false(self.conn.in_transaction())
- with context.begin_transaction(_per_migration=True):
- is_true(self.conn.in_transaction())
-
- is_false(self.conn.in_transaction())
- is_false(self.conn.in_transaction())
-
- def test_transaction_per_all_sqlmode(self):
- context = self._fixture({"as_sql": True})
-
- context.execute("step 1")
- with context.begin_transaction():
- context.execute("step 2")
- with context.begin_transaction(_per_migration=True):
- context.execute("step 3")
-
- context.execute("step 4")
- context.execute("step 5")
-
- if context.impl.transactional_ddl:
- self._assert_impl_steps(
- "step 1",
- "BEGIN",
- "step 2",
- "step 3",
- "step 4",
- "COMMIT",
- "step 5",
- )
- else:
- self._assert_impl_steps(
- "step 1", "step 2", "step 3", "step 4", "step 5"
- )
-
- def test_transaction_per_migration_sqlmode(self):
- context = self._fixture(
- {"as_sql": True, "transaction_per_migration": True}
- )
-
- context.execute("step 1")
- with context.begin_transaction():
- context.execute("step 2")
- with context.begin_transaction(_per_migration=True):
- context.execute("step 3")
-
- context.execute("step 4")
- context.execute("step 5")
-
- if context.impl.transactional_ddl:
- self._assert_impl_steps(
- "step 1",
- "step 2",
- "BEGIN",
- "step 3",
- "COMMIT",
- "step 4",
- "step 5",
- )
- else:
- self._assert_impl_steps(
- "step 1", "step 2", "step 3", "step 4", "step 5"
- )
-
- @config.requirements.autocommit_isolation
- def test_autocommit_block(self):
- context = self._fixture({"transaction_per_migration": True})
-
- is_false(self.conn.in_transaction())
-
- with context.begin_transaction():
- is_false(self.conn.in_transaction())
- with context.begin_transaction(_per_migration=True):
- is_true(self.conn.in_transaction())
-
- with context.autocommit_block():
- is_false(self.conn.in_transaction())
-
- is_true(self.conn.in_transaction())
-
- is_false(self.conn.in_transaction())
- is_false(self.conn.in_transaction())
-
- @config.requirements.autocommit_isolation
- def test_autocommit_block_no_transaction(self):
- context = self._fixture({"transaction_per_migration": True})
-
- is_false(self.conn.in_transaction())
-
- with context.autocommit_block():
- is_false(self.conn.in_transaction())
- is_false(self.conn.in_transaction())
-
- def test_autocommit_block_transactional_ddl_sqlmode(self):
- context = self._fixture(
- {
- "transaction_per_migration": True,
- "transactional_ddl": True,
- "as_sql": True,
- }
- )
-
- with context.begin_transaction():
- context.execute("step 1")
- with context.begin_transaction(_per_migration=True):
- context.execute("step 2")
-
- with context.autocommit_block():
- context.execute("step 3")
-
- context.execute("step 4")
-
- context.execute("step 5")
-
- self._assert_impl_steps(
- "step 1",
- "BEGIN",
- "step 2",
- "COMMIT",
- "step 3",
- "BEGIN",
- "step 4",
- "COMMIT",
- "step 5",
- )
-
- def test_autocommit_block_nontransactional_ddl_sqlmode(self):
- context = self._fixture(
- {
- "transaction_per_migration": True,
- "transactional_ddl": False,
- "as_sql": True,
- }
- )
-
- with context.begin_transaction():
- context.execute("step 1")
- with context.begin_transaction(_per_migration=True):
- context.execute("step 2")
-
- with context.autocommit_block():
- context.execute("step 3")
-
- context.execute("step 4")
-
- context.execute("step 5")
-
- self._assert_impl_steps(
- "step 1", "step 2", "step 3", "step 4", "step 5"
- )
-
- def _assert_impl_steps(self, *steps):
- to_check = self.context.output_buffer.getvalue()
-
- self.context.impl.output_buffer = buf = compat.StringIO()
- for step in steps:
- if step == "BEGIN":
- self.context.impl.emit_begin()
- elif step == "COMMIT":
- self.context.impl.emit_commit()
- else:
- self.context.impl._exec(step)
-
- eq_(to_check, buf.getvalue())
from sqlalchemy import Boolean
from sqlalchemy import CheckConstraint
from sqlalchemy import Column
-from sqlalchemy import event
from sqlalchemy import exc
from sqlalchemy import ForeignKey
from sqlalchemy import Index
from alembic.testing import eq_
from alembic.testing import is_not_
from alembic.testing import mock
-from alembic.testing.fixtures import AlterColRoundTripFixture
from alembic.testing.fixtures import op_fixture
from alembic.testing.fixtures import TestBase
from alembic.util import sqla_compat
-@event.listens_for(Table, "after_parent_attach")
-def _add_cols(table, metadata):
- if table.name == "tbl_with_auto_appended_column":
- table.append_column(Column("bat", Integer))
-
-
class OpTest(TestBase):
def test_rename_table(self):
context = op_fixture()
assert_raises_message(
ValueError, "constraint cannot be produced", op.to_constraint
)
-
-
-class BackendAlterColumnTest(AlterColRoundTripFixture, TestBase):
- __backend__ = True
-
- def test_rename_column(self):
- self._run_alter_col({}, {"name": "newname"})
-
- def test_modify_type_int_str(self):
- self._run_alter_col({"type": Integer()}, {"type": String(50)})
-
- def test_add_server_default_int(self):
- self._run_alter_col({"type": Integer}, {"server_default": text("5")})
-
- def test_modify_server_default_int(self):
- self._run_alter_col(
- {"type": Integer, "server_default": text("2")},
- {"server_default": text("5")},
- )
-
- def test_modify_nullable_to_non(self):
- self._run_alter_col({}, {"nullable": False})
-
- def test_modify_non_nullable_to_nullable(self):
- self._run_alter_col({"nullable": False}, {"nullable": True})
--- /dev/null
+from alembic.testing.suite import * # noqa
projects / thirdparty / sqlalchemy / alembic.git / commitdiff
