Wala shell

Advanced PHP File Manager

Server IP: 127.0.0.1 | PHP Version: 8.3.30 | Domains Hosted: 54

Name	Size	Last Modified	Actions
📁 ..	-	2026-03-21 15:45
📄 __init__.py 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	288 B	2026-01-20 12:58
📁 __pycache__	-	2026-03-21 15:45
📄 _autogen_fixtures.py from __future__ import annotations from typing import Any from typing import Dict from typing import Set from sqlalchemy import CHAR from sqlalchemy import CheckConstraint from sqlalchemy import Column from sqlalchemy import event from sqlalchemy import ForeignKey from sqlalchemy import Index from sqlalchemy import inspect from sqlalchemy import Integer from sqlalchemy import MetaData from sqlalchemy import Numeric from sqlalchemy import String from sqlalchemy import Table from sqlalchemy import Text from sqlalchemy import text from sqlalchemy import UniqueConstraint 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[Any] = set() @event.listens_for(Table, "after_parent_attach") def new_table(table, parent): names_in_this_test.add(table.name) def _default_include_object(obj, name, type_, reflected, compare_to): if type_ == "table": return name in names_in_this_test else: return True _default_object_filters: Any = _default_include_object _default_name_filters: Any = None class ModelOne: __requires__ = ("unique_constraint_reflection",) schema: Any = None @classmethod def _get_db_schema(cls): schema = cls.schema m = MetaData(schema=schema) Table( "user", m, Column("id", Integer, primary_key=True), Column("name", String(50)), Column("a1", Text), Column("pw", String(50)), Index("pw_idx", "pw"), ) Table( "address", m, Column("id", Integer, primary_key=True), Column("email_address", String(100), nullable=False), ) Table( "order", m, Column("order_id", Integer, primary_key=True), Column( "amount", Numeric(8, 2), nullable=False, server_default=text("0"), ), CheckConstraint("amount >= 0", name="ck_order_amount"), ) Table( "extra", m, Column("x", CHAR), Column("uid", Integer, ForeignKey("user.id")), ) return m @classmethod def _get_model_schema(cls): schema = cls.schema m = MetaData(schema=schema) Table( "user", m, Column("id", Integer, primary_key=True), Column("name", String(50), nullable=False), Column("a1", Text, server_default="x"), ) Table( "address", m, Column("id", Integer, primary_key=True), Column("email_address", String(100), nullable=False), Column("street", String(50)), UniqueConstraint("email_address", name="uq_email"), ) Table( "order", m, Column("order_id", Integer, primary_key=True), Column( "amount", Numeric(10, 2), nullable=True, server_default=text("0"), ), Column("user_id", Integer, ForeignKey("user.id")), CheckConstraint("amount > -1", name="ck_order_amount"), ) Table( "item", m, Column("id", Integer, primary_key=True), Column("description", String(100)), Column("order_id", Integer, ForeignKey("order.order_id")), CheckConstraint("len(description) > 5"), ) return m class _ComparesFKs: def _assert_fk_diff( self, diff, type_, source_table, source_columns, target_table, target_columns, name=None, conditional_name=None, source_schema=None, onupdate=None, ondelete=None, initially=None, deferrable=None, ): # the public API for ForeignKeyConstraint was not very rich # in 0.7, 0.8, so here we use the well-known but slightly # private API to get at its elements ( fk_source_schema, fk_source_table, fk_source_columns, fk_target_schema, fk_target_table, fk_target_columns, fk_onupdate, fk_ondelete, fk_deferrable, fk_initially, ) = _fk_spec(diff[1]) eq_(diff[0], type_) eq_(fk_source_table, source_table) eq_(fk_source_columns, source_columns) eq_(fk_target_table, target_table) eq_(fk_source_schema, source_schema) eq_(fk_onupdate, onupdate) eq_(fk_ondelete, ondelete) eq_(fk_initially, initially) eq_(fk_deferrable, deferrable) eq_([elem.column.name for elem in diff[1].elements], target_columns) if conditional_name is not None: if conditional_name == "servergenerated": fks = inspect(self.bind).get_foreign_keys(source_table) server_fk_name = fks[0]["name"] eq_(diff[1].name, server_fk_name) else: eq_(diff[1].name, conditional_name) else: eq_(diff[1].name, name) class AutogenTest(_ComparesFKs): def _flatten_diffs(self, diffs): for d in diffs: if isinstance(d, list): yield from self._flatten_diffs(d) else: yield d @classmethod def _get_bind(cls): return config.db configure_opts: Dict[Any, Any] = {} @classmethod def setup_class(cls): staging_env() cls.bind = cls._get_bind() cls.m1 = cls._get_db_schema() cls.m1.create_all(cls.bind) cls.m2 = cls._get_model_schema() @classmethod def teardown_class(cls): cls.m1.drop_all(cls.bind) clear_staging_env() def setUp(self): self.conn = conn = self.bind.connect() ctx_opts = { "compare_type": True, "compare_server_default": True, "target_metadata": self.m2, "upgrade_token": "upgrades", "downgrade_token": "downgrades", "alembic_module_prefix": "op.", "sqlalchemy_module_prefix": "sa.", "include_object": _default_object_filters, "include_name": _default_name_filters, } if self.configure_opts: ctx_opts.update(self.configure_opts) self.context = context = MigrationContext.configure( connection=conn, opts=ctx_opts ) self.autogen_context = api.AutogenContext(context, self.m2) def tearDown(self): self.conn.close() def _update_context( self, object_filters=None, name_filters=None, include_schemas=None ): if include_schemas is not None: self.autogen_context.opts["include_schemas"] = include_schemas if object_filters is not None: self.autogen_context._object_filters = [object_filters] if name_filters is not None: self.autogen_context._name_filters = [name_filters] return self.autogen_context class AutogenFixtureTest(_ComparesFKs): def _fixture( self, m1, m2, include_schemas=False, opts=None, object_filters=_default_object_filters, name_filters=_default_name_filters, return_ops=False, max_identifier_length=None, ): if max_identifier_length: dialect = self.bind.dialect existing_length = dialect.max_identifier_length dialect.max_identifier_length = ( dialect._user_defined_max_identifier_length ) = max_identifier_length try: self._alembic_metadata, model_metadata = m1, m2 for m in util.to_list(self._alembic_metadata): m.create_all(self.bind) with self.bind.connect() as conn: ctx_opts = { "compare_type": True, "compare_server_default": True, "target_metadata": model_metadata, "upgrade_token": "upgrades", "downgrade_token": "downgrades", "alembic_module_prefix": "op.", "sqlalchemy_module_prefix": "sa.", "include_object": object_filters, "include_name": name_filters, "include_schemas": include_schemas, } if opts: ctx_opts.update(opts) self.context = context = MigrationContext.configure( connection=conn, opts=ctx_opts ) autogen_context = api.AutogenContext(context, model_metadata) uo = ops.UpgradeOps(ops=[]) autogenerate._produce_net_changes(autogen_context, uo) if return_ops: return uo else: return uo.as_diffs() finally: if max_identifier_length: dialect = self.bind.dialect dialect.max_identifier_length = ( dialect._user_defined_max_identifier_length ) = existing_length def setUp(self): staging_env() self.bind = config.db def tearDown(self): if hasattr(self, "_alembic_metadata"): for m in util.to_list(self._alembic_metadata): m.drop_all(self.bind) clear_staging_env()	9.65 KB	2026-01-20 12:58
📄 test_autogen_comments.py from sqlalchemy import Column from sqlalchemy import Float from sqlalchemy import MetaData from sqlalchemy import String from sqlalchemy import Table from ._autogen_fixtures import AutogenFixtureTest from ...testing import eq_ from ...testing import mock from ...testing import TestBase class AutogenerateCommentsTest(AutogenFixtureTest, TestBase): __backend__ = True __requires__ = ("comments",) def test_existing_table_comment_no_change(self): m1 = MetaData() m2 = MetaData() Table( "some_table", m1, Column("test", String(10), primary_key=True), comment="this is some table", ) Table( "some_table", m2, Column("test", String(10), primary_key=True), comment="this is some table", ) diffs = self._fixture(m1, m2) eq_(diffs, []) def test_add_table_comment(self): m1 = MetaData() m2 = MetaData() Table("some_table", m1, Column("test", String(10), primary_key=True)) Table( "some_table", m2, Column("test", String(10), primary_key=True), comment="this is some table", ) diffs = self._fixture(m1, m2) eq_(diffs[0][0], "add_table_comment") eq_(diffs[0][1].comment, "this is some table") eq_(diffs[0][2], None) def test_remove_table_comment(self): m1 = MetaData() m2 = MetaData() Table( "some_table", m1, Column("test", String(10), primary_key=True), comment="this is some table", ) Table("some_table", m2, Column("test", String(10), primary_key=True)) diffs = self._fixture(m1, m2) eq_(diffs[0][0], "remove_table_comment") eq_(diffs[0][1].comment, None) def test_alter_table_comment(self): m1 = MetaData() m2 = MetaData() Table( "some_table", m1, Column("test", String(10), primary_key=True), comment="this is some table", ) Table( "some_table", m2, Column("test", String(10), primary_key=True), comment="this is also some table", ) diffs = self._fixture(m1, m2) eq_(diffs[0][0], "add_table_comment") eq_(diffs[0][1].comment, "this is also some table") eq_(diffs[0][2], "this is some table") def test_existing_column_comment_no_change(self): m1 = MetaData() m2 = MetaData() Table( "some_table", m1, Column("test", String(10), primary_key=True), Column("amount", Float, comment="the amount"), ) Table( "some_table", m2, Column("test", String(10), primary_key=True), Column("amount", Float, comment="the amount"), ) diffs = self._fixture(m1, m2) eq_(diffs, []) def test_add_column_comment(self): m1 = MetaData() m2 = MetaData() Table( "some_table", m1, Column("test", String(10), primary_key=True), Column("amount", Float), ) Table( "some_table", m2, Column("test", String(10), primary_key=True), Column("amount", Float, comment="the amount"), ) diffs = self._fixture(m1, m2) eq_( diffs, [ [ ( "modify_comment", None, "some_table", "amount", { "existing_nullable": True, "existing_type": mock.ANY, "existing_server_default": False, }, None, "the amount", ) ] ], ) def test_remove_column_comment(self): m1 = MetaData() m2 = MetaData() Table( "some_table", m1, Column("test", String(10), primary_key=True), Column("amount", Float, comment="the amount"), ) Table( "some_table", m2, Column("test", String(10), primary_key=True), Column("amount", Float), ) diffs = self._fixture(m1, m2) eq_( diffs, [ [ ( "modify_comment", None, "some_table", "amount", { "existing_nullable": True, "existing_type": mock.ANY, "existing_server_default": False, }, "the amount", None, ) ] ], ) def test_alter_column_comment(self): m1 = MetaData() m2 = MetaData() Table( "some_table", m1, Column("test", String(10), primary_key=True), Column("amount", Float, comment="the amount"), ) Table( "some_table", m2, Column("test", String(10), primary_key=True), Column("amount", Float, comment="the adjusted amount"), ) diffs = self._fixture(m1, m2) eq_( diffs, [ [ ( "modify_comment", None, "some_table", "amount", { "existing_nullable": True, "existing_type": mock.ANY, "existing_server_default": False, }, "the amount", "the adjusted amount", ) ] ], )	6.14 KB	2026-01-20 12:58
📄 test_autogen_computed.py import sqlalchemy as sa from sqlalchemy import Column from sqlalchemy import Integer from sqlalchemy import MetaData from sqlalchemy import Table 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): __requires__ = ("computed_columns",) __backend__ = True def test_add_computed_column(self): m1 = MetaData() m2 = MetaData() Table("user", m1, Column("id", Integer, primary_key=True)) Table( "user", m2, Column("id", Integer, primary_key=True), Column("foo", Integer, sa.Computed("5")), ) diffs = self._fixture(m1, m2) eq_(diffs[0][0], "add_column") eq_(diffs[0][2], "user") eq_(diffs[0][3].name, "foo") c = diffs[0][3].computed is_true(isinstance(c, sa.Computed)) is_(c.persisted, None) eq_(str(c.sqltext), "5") def test_remove_computed_column(self): m1 = MetaData() m2 = MetaData() Table( "user", m1, Column("id", Integer, primary_key=True), Column("foo", Integer, sa.Computed("5")), ) Table("user", m2, Column("id", Integer, primary_key=True)) diffs = self._fixture(m1, m2) eq_(diffs[0][0], "remove_column") eq_(diffs[0][2], "user") c = diffs[0][3] eq_(c.name, "foo") if config.requirements.computed_reflects_normally.enabled: is_true(isinstance(c.computed, sa.Computed)) else: is_(c.computed, None) if config.requirements.computed_reflects_as_server_default.enabled: is_true(isinstance(c.server_default, sa.DefaultClause)) eq_(str(c.server_default.arg.text), "5") elif config.requirements.computed_reflects_normally.enabled: is_true(isinstance(c.computed, sa.Computed)) else: is_(c.computed, None) @testing.combinations( lambda: (None, sa.Computed("bar5")), (lambda: (sa.Computed("bar5"), None)), lambda: ( sa.Computed("bar5"), sa.Computed("bar 42", persisted=True), ), lambda: (sa.Computed("bar5"), sa.Computed("bar 42")), ) @config.requirements.computed_reflects_normally def test_cant_change_computed_warning(self, test_case): arg_before, arg_after = testing.resolve_lambda(test_case, *locals()) m1 = MetaData() m2 = MetaData() arg_before = [] if arg_before is None else [arg_before] arg_after = [] if arg_after is None else [arg_after] Table( "user", m1, Column("id", Integer, primary_key=True), Column("bar", Integer), Column("foo", Integer, arg_before), ) Table( "user", m2, Column("id", Integer, primary_key=True), Column("bar", Integer), Column("foo", Integer, arg_after), ) with mock.patch("alembic.util.warn") as mock_warn: diffs = self._fixture(m1, m2) eq_( mock_warn.mock_calls, [mock.call("Computed default on user.foo cannot be modified")], ) eq_(list(diffs), []) @testing.combinations( lambda: (None, None), lambda: (sa.Computed("5"), sa.Computed("5")), lambda: (sa.Computed("bar5"), sa.Computed("bar5")), ( lambda: (sa.Computed("bar5"), None), config.requirements.computed_doesnt_reflect_as_server_default, ), ) def test_computed_unchanged(self, test_case): arg_before, arg_after = testing.resolve_lambda(test_case, *locals()) m1 = MetaData() m2 = MetaData() arg_before = [] if arg_before is None else [arg_before] arg_after = [] if arg_after is None else [arg_after] Table( "user", m1, Column("id", Integer, primary_key=True), Column("bar", Integer), Column("foo", Integer, arg_before), ) Table( "user", m2, Column("id", Integer, primary_key=True), Column("bar", Integer), Column("foo", Integer, *arg_after), ) with mock.patch("alembic.util.warn") as mock_warn: diffs = self._fixture(m1, m2) eq_(mock_warn.mock_calls, []) eq_(list(diffs), []) @config.requirements.computed_reflects_as_server_default def test_remove_computed_default_on_computed(self): """Asserts the current behavior which is that on PG and Oracle, the GENERATED ALWAYS AS is reflected as a server default which we can't tell is actually "computed", so these come out as a modification to the server default. """ m1 = MetaData() m2 = MetaData() Table( "user", m1, Column("id", Integer, primary_key=True), Column("bar", Integer), Column("foo", Integer, sa.Computed("bar + 42")), ) Table( "user", m2, Column("id", Integer, primary_key=True), Column("bar", Integer), Column("foo", Integer), ) diffs = self._fixture(m1, m2) eq_(diffs[0][0][0], "modify_default") eq_(diffs[0][0][2], "user") eq_(diffs[0][0][3], "foo") old = diffs[0][0][-2] new = diffs[0][0][-1] is_(new, None) is_true(isinstance(old, sa.DefaultClause)) if exclusions.against(config, "postgresql"): eq_(str(old.arg.text), "(bar + 42)") elif exclusions.against(config, "oracle"): eq_(str(old.arg.text), '"BAR"+42')	5.93 KB	2026-01-20 12:58
📄 test_autogen_diffs.py 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 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)	8.2 KB	2026-01-20 12:58
📄 test_autogen_fks.py from sqlalchemy import Column from sqlalchemy import ForeignKeyConstraint from sqlalchemy import Integer from sqlalchemy import MetaData from sqlalchemy import String from sqlalchemy import Table 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 class AutogenerateForeignKeysTest(AutogenFixtureTest, TestBase): __backend__ = True __requires__ = ("foreign_key_constraint_reflection",) def test_remove_fk(self): m1 = MetaData() m2 = MetaData() Table( "some_table", m1, Column("test", String(10), primary_key=True), ) Table( "user", m1, Column("id", Integer, primary_key=True), Column("name", String(50), nullable=False), Column("a1", String(10), server_default="x"), Column("test2", String(10)), ForeignKeyConstraint(["test2"], ["some_table.test"]), ) Table( "some_table", m2, Column("test", String(10), primary_key=True), ) Table( "user", m2, Column("id", Integer, primary_key=True), Column("name", String(50), nullable=False), Column("a1", String(10), server_default="x"), Column("test2", String(10)), ) diffs = self._fixture(m1, m2) self._assert_fk_diff( diffs[0], "remove_fk", "user", ["test2"], "some_table", ["test"], conditional_name="servergenerated", ) def test_add_fk(self): m1 = MetaData() m2 = MetaData() Table( "some_table", m1, Column("id", Integer, primary_key=True), Column("test", String(10)), ) Table( "user", m1, Column("id", Integer, primary_key=True), Column("name", String(50), nullable=False), Column("a1", String(10), server_default="x"), Column("test2", String(10)), ) Table( "some_table", m2, Column("id", Integer, primary_key=True), Column("test", String(10)), ) Table( "user", m2, Column("id", Integer, primary_key=True), Column("name", String(50), nullable=False), Column("a1", String(10), server_default="x"), Column("test2", String(10)), ForeignKeyConstraint(["test2"], ["some_table.test"]), ) diffs = self._fixture(m1, m2) self._assert_fk_diff( diffs[0], "add_fk", "user", ["test2"], "some_table", ["test"] ) def test_no_change(self): m1 = MetaData() m2 = MetaData() Table( "some_table", m1, Column("id", Integer, primary_key=True), Column("test", String(10)), ) Table( "user", m1, Column("id", Integer, primary_key=True), Column("name", String(50), nullable=False), Column("a1", String(10), server_default="x"), Column("test2", Integer), ForeignKeyConstraint(["test2"], ["some_table.id"]), ) Table( "some_table", m2, Column("id", Integer, primary_key=True), Column("test", String(10)), ) Table( "user", m2, Column("id", Integer, primary_key=True), Column("name", String(50), nullable=False), Column("a1", String(10), server_default="x"), Column("test2", Integer), ForeignKeyConstraint(["test2"], ["some_table.id"]), ) diffs = self._fixture(m1, m2) eq_(diffs, []) def test_no_change_composite_fk(self): m1 = MetaData() m2 = MetaData() Table( "some_table", m1, Column("id_1", String(10), primary_key=True), Column("id_2", String(10), primary_key=True), ) Table( "user", m1, Column("id", Integer, primary_key=True), Column("name", String(50), nullable=False), Column("a1", String(10), server_default="x"), Column("other_id_1", String(10)), Column("other_id_2", String(10)), ForeignKeyConstraint( ["other_id_1", "other_id_2"], ["some_table.id_1", "some_table.id_2"], ), ) Table( "some_table", m2, Column("id_1", String(10), primary_key=True), Column("id_2", String(10), primary_key=True), ) Table( "user", m2, Column("id", Integer, primary_key=True), Column("name", String(50), nullable=False), Column("a1", String(10), server_default="x"), Column("other_id_1", String(10)), Column("other_id_2", String(10)), ForeignKeyConstraint( ["other_id_1", "other_id_2"], ["some_table.id_1", "some_table.id_2"], ), ) diffs = self._fixture(m1, m2) eq_(diffs, []) def test_casing_convention_changed_so_put_drops_first(self): m1 = MetaData() m2 = MetaData() Table( "some_table", m1, Column("test", String(10), primary_key=True), ) Table( "user", m1, Column("id", Integer, primary_key=True), Column("name", String(50), nullable=False), Column("a1", String(10), server_default="x"), Column("test2", String(10)), ForeignKeyConstraint(["test2"], ["some_table.test"], name="MyFK"), ) Table( "some_table", m2, Column("test", String(10), primary_key=True), ) # foreign key autogen currently does not take "name" into account, # so change the def just for the purposes of testing the # add/drop order for now. Table( "user", m2, Column("id", Integer, primary_key=True), Column("name", String(50), nullable=False), Column("a1", String(10), server_default="x"), Column("test2", String(10)), ForeignKeyConstraint(["a1"], ["some_table.test"], name="myfk"), ) diffs = self._fixture(m1, m2) self._assert_fk_diff( diffs[0], "remove_fk", "user", ["test2"], "some_table", ["test"], name="MyFK" if config.requirements.fk_names.enabled else None, ) self._assert_fk_diff( diffs[1], "add_fk", "user", ["a1"], "some_table", ["test"], name="myfk", ) def test_add_composite_fk_with_name(self): m1 = MetaData() m2 = MetaData() Table( "some_table", m1, Column("id_1", String(10), primary_key=True), Column("id_2", String(10), primary_key=True), ) Table( "user", m1, Column("id", Integer, primary_key=True), Column("name", String(50), nullable=False), Column("a1", String(10), server_default="x"), Column("other_id_1", String(10)), Column("other_id_2", String(10)), ) Table( "some_table", m2, Column("id_1", String(10), primary_key=True), Column("id_2", String(10), primary_key=True), ) Table( "user", m2, Column("id", Integer, primary_key=True), Column("name", String(50), nullable=False), Column("a1", String(10), server_default="x"), Column("other_id_1", String(10)), Column("other_id_2", String(10)), ForeignKeyConstraint( ["other_id_1", "other_id_2"], ["some_table.id_1", "some_table.id_2"], name="fk_test_name", ), ) diffs = self._fixture(m1, m2) self._assert_fk_diff( diffs[0], "add_fk", "user", ["other_id_1", "other_id_2"], "some_table", ["id_1", "id_2"], name="fk_test_name", ) @config.requirements.no_name_normalize def test_remove_composite_fk(self): m1 = MetaData() m2 = MetaData() Table( "some_table", m1, Column("id_1", String(10), primary_key=True), Column("id_2", String(10), primary_key=True), ) Table( "user", m1, Column("id", Integer, primary_key=True), Column("name", String(50), nullable=False), Column("a1", String(10), server_default="x"), Column("other_id_1", String(10)), Column("other_id_2", String(10)), ForeignKeyConstraint( ["other_id_1", "other_id_2"], ["some_table.id_1", "some_table.id_2"], name="fk_test_name", ), ) Table( "some_table", m2, Column("id_1", String(10), primary_key=True), Column("id_2", String(10), primary_key=True), ) Table( "user", m2, Column("id", Integer, primary_key=True), Column("name", String(50), nullable=False), Column("a1", String(10), server_default="x"), Column("other_id_1", String(10)), Column("other_id_2", String(10)), ) diffs = self._fixture(m1, m2) self._assert_fk_diff( diffs[0], "remove_fk", "user", ["other_id_1", "other_id_2"], "some_table", ["id_1", "id_2"], conditional_name="fk_test_name", ) def test_add_fk_colkeys(self): m1 = MetaData() m2 = MetaData() Table( "some_table", m1, Column("id_1", String(10), primary_key=True), Column("id_2", String(10), primary_key=True), ) Table( "user", m1, Column("id", Integer, primary_key=True), Column("other_id_1", String(10)), Column("other_id_2", String(10)), ) Table( "some_table", m2, Column("id_1", String(10), key="tid1", primary_key=True), Column("id_2", String(10), key="tid2", primary_key=True), ) Table( "user", m2, Column("id", Integer, primary_key=True), Column("other_id_1", String(10), key="oid1"), Column("other_id_2", String(10), key="oid2"), ForeignKeyConstraint( ["oid1", "oid2"], ["some_table.tid1", "some_table.tid2"], name="fk_test_name", ), ) diffs = self._fixture(m1, m2) self._assert_fk_diff( diffs[0], "add_fk", "user", ["other_id_1", "other_id_2"], "some_table", ["id_1", "id_2"], name="fk_test_name", ) def test_no_change_colkeys(self): m1 = MetaData() m2 = MetaData() Table( "some_table", m1, Column("id_1", String(10), primary_key=True), Column("id_2", String(10), primary_key=True), ) Table( "user", m1, Column("id", Integer, primary_key=True), Column("other_id_1", String(10)), Column("other_id_2", String(10)), ForeignKeyConstraint( ["other_id_1", "other_id_2"], ["some_table.id_1", "some_table.id_2"], ), ) Table( "some_table", m2, Column("id_1", String(10), key="tid1", primary_key=True), Column("id_2", String(10), key="tid2", primary_key=True), ) Table( "user", m2, Column("id", Integer, primary_key=True), Column("other_id_1", String(10), key="oid1"), Column("other_id_2", String(10), key="oid2"), ForeignKeyConstraint( ["oid1", "oid2"], ["some_table.tid1", "some_table.tid2"] ), ) diffs = self._fixture(m1, m2) eq_(diffs, []) class IncludeHooksTest(AutogenFixtureTest, TestBase): __backend__ = True __requires__ = ("fk_names",) @combinations(("object",), ("name",)) @config.requirements.no_name_normalize def test_remove_connection_fk(self, hook_type): m1 = MetaData() m2 = MetaData() ref = Table( "ref", m1, Column("id", Integer, primary_key=True), ) t1 = Table( "t", m1, Column("x", Integer), Column("y", Integer), ) t1.append_constraint( ForeignKeyConstraint([t1.c.x], [ref.c.id], name="fk1") ) t1.append_constraint( ForeignKeyConstraint([t1.c.y], [ref.c.id], name="fk2") ) ref = Table( "ref", m2, Column("id", Integer, primary_key=True), ) Table( "t", m2, Column("x", Integer), Column("y", Integer), ) if hook_type == "object": def include_object(object_, name, type_, reflected, compare_to): return not ( isinstance(object_, ForeignKeyConstraint) and type_ == "foreign_key_constraint" and reflected and name == "fk1" ) diffs = self._fixture(m1, m2, object_filters=include_object) elif hook_type == "name": def include_name(name, type_, parent_names): if name == "fk1": if type_ == "index": # MariaDB thing return True eq_(type_, "foreign_key_constraint") eq_( parent_names, { "schema_name": None, "table_name": "t", "schema_qualified_table_name": "t", }, ) return False else: return True diffs = self._fixture(m1, m2, name_filters=include_name) self._assert_fk_diff( diffs[0], "remove_fk", "t", ["y"], "ref", ["id"], conditional_name="fk2", ) eq_(len(diffs), 1) def test_add_metadata_fk(self): m1 = MetaData() m2 = MetaData() Table( "ref", m1, Column("id", Integer, primary_key=True), ) Table( "t", m1, Column("x", Integer), Column("y", Integer), ) ref = Table( "ref", m2, Column("id", Integer, primary_key=True), ) t2 = Table( "t", m2, Column("x", Integer), Column("y", Integer), ) t2.append_constraint( ForeignKeyConstraint([t2.c.x], [ref.c.id], name="fk1") ) t2.append_constraint( ForeignKeyConstraint([t2.c.y], [ref.c.id], name="fk2") ) def include_object(object_, name, type_, reflected, compare_to): return not ( isinstance(object_, ForeignKeyConstraint) and type_ == "foreign_key_constraint" and not reflected and name == "fk1" ) diffs = self._fixture(m1, m2, object_filters=include_object) self._assert_fk_diff( diffs[0], "add_fk", "t", ["y"], "ref", ["id"], name="fk2" ) eq_(len(diffs), 1) @combinations(("object",), ("name",)) @config.requirements.no_name_normalize def test_change_fk(self, hook_type): m1 = MetaData() m2 = MetaData() r1a = Table( "ref_a", m1, Column("a", Integer, primary_key=True), ) Table( "ref_b", m1, Column("a", Integer, primary_key=True), Column("b", Integer, primary_key=True), ) t1 = Table( "t", m1, Column("x", Integer), Column("y", Integer), Column("z", Integer), ) t1.append_constraint( ForeignKeyConstraint([t1.c.x], [r1a.c.a], name="fk1") ) t1.append_constraint( ForeignKeyConstraint([t1.c.y], [r1a.c.a], name="fk2") ) Table( "ref_a", m2, Column("a", Integer, primary_key=True), ) r2b = Table( "ref_b", m2, Column("a", Integer, primary_key=True), Column("b", Integer, primary_key=True), ) t2 = Table( "t", m2, Column("x", Integer), Column("y", Integer), Column("z", Integer), ) t2.append_constraint( ForeignKeyConstraint( [t2.c.x, t2.c.z], [r2b.c.a, r2b.c.b], name="fk1" ) ) t2.append_constraint( ForeignKeyConstraint( [t2.c.y, t2.c.z], [r2b.c.a, r2b.c.b], name="fk2" ) ) if hook_type == "object": def include_object(object_, name, type_, reflected, compare_to): return not ( isinstance(object_, ForeignKeyConstraint) and type_ == "foreign_key_constraint" and name == "fk1" ) diffs = self._fixture(m1, m2, object_filters=include_object) elif hook_type == "name": def include_name(name, type_, parent_names): if type_ == "index": return True # MariaDB thing if name == "fk1": eq_(type_, "foreign_key_constraint") eq_( parent_names, { "schema_name": None, "table_name": "t", "schema_qualified_table_name": "t", }, ) return False else: return True diffs = self._fixture(m1, m2, name_filters=include_name) if hook_type == "object": self._assert_fk_diff( diffs[0], "remove_fk", "t", ["y"], "ref_a", ["a"], name="fk2" ) self._assert_fk_diff( diffs[1], "add_fk", "t", ["y", "z"], "ref_b", ["a", "b"], name="fk2", ) eq_(len(diffs), 2) elif hook_type == "name": eq_( {(d[0], d[1].name) for d in diffs}, {("add_fk", "fk2"), ("add_fk", "fk1"), ("remove_fk", "fk2")}, ) class AutogenerateFKOptionsTest(AutogenFixtureTest, TestBase): __backend__ = True def _fk_opts_fixture(self, old_opts, new_opts): m1 = MetaData() m2 = MetaData() Table( "some_table", m1, Column("id", Integer, primary_key=True), Column("test", String(10)), ) Table( "user", m1, Column("id", Integer, primary_key=True), Column("name", String(50), nullable=False), Column("tid", Integer), ForeignKeyConstraint(["tid"], ["some_table.id"], old_opts), ) Table( "some_table", m2, Column("id", Integer, primary_key=True), Column("test", String(10)), ) Table( "user", m2, Column("id", Integer, primary_key=True), Column("name", String(50), nullable=False), Column("tid", Integer), ForeignKeyConstraint(["tid"], ["some_table.id"], new_opts), ) return self._fixture(m1, m2) @config.requirements.fk_ondelete_is_reflected def test_add_ondelete(self): diffs = self._fk_opts_fixture({}, {"ondelete": "cascade"}) self._assert_fk_diff( diffs[0], "remove_fk", "user", ["tid"], "some_table", ["id"], ondelete=None, conditional_name="servergenerated", ) self._assert_fk_diff( diffs[1], "add_fk", "user", ["tid"], "some_table", ["id"], ondelete="cascade", ) @config.requirements.fk_ondelete_is_reflected def test_remove_ondelete(self): diffs = self._fk_opts_fixture({"ondelete": "CASCADE"}, {}) self._assert_fk_diff( diffs[0], "remove_fk", "user", ["tid"], "some_table", ["id"], ondelete="CASCADE", conditional_name="servergenerated", ) self._assert_fk_diff( diffs[1], "add_fk", "user", ["tid"], "some_table", ["id"], ondelete=None, ) def test_nochange_ondelete(self): """test case sensitivity""" diffs = self._fk_opts_fixture( {"ondelete": "caSCAde"}, {"ondelete": "CasCade"} ) eq_(diffs, []) @config.requirements.fk_onupdate_is_reflected def test_add_onupdate(self): diffs = self._fk_opts_fixture({}, {"onupdate": "cascade"}) self._assert_fk_diff( diffs[0], "remove_fk", "user", ["tid"], "some_table", ["id"], onupdate=None, conditional_name="servergenerated", ) self._assert_fk_diff( diffs[1], "add_fk", "user", ["tid"], "some_table", ["id"], onupdate="cascade", ) @config.requirements.fk_onupdate_is_reflected def test_remove_onupdate(self): diffs = self._fk_opts_fixture({"onupdate": "CASCADE"}, {}) self._assert_fk_diff( diffs[0], "remove_fk", "user", ["tid"], "some_table", ["id"], onupdate="CASCADE", conditional_name="servergenerated", ) self._assert_fk_diff( diffs[1], "add_fk", "user", ["tid"], "some_table", ["id"], onupdate=None, ) @config.requirements.fk_onupdate def test_nochange_onupdate(self): """test case sensitivity""" diffs = self._fk_opts_fixture( {"onupdate": "caSCAde"}, {"onupdate": "CasCade"} ) eq_(diffs, []) @config.requirements.fk_ondelete_restrict def test_nochange_ondelete_restrict(self): """test the RESTRICT option which MySQL doesn't report on""" diffs = self._fk_opts_fixture( {"ondelete": "restrict"}, {"ondelete": "restrict"} ) eq_(diffs, []) @config.requirements.fk_onupdate_restrict def test_nochange_onupdate_restrict(self): """test the RESTRICT option which MySQL doesn't report on""" diffs = self._fk_opts_fixture( {"onupdate": "restrict"}, {"onupdate": "restrict"} ) eq_(diffs, []) @config.requirements.fk_ondelete_noaction def test_nochange_ondelete_noaction(self): """test the NO ACTION option which generally comes back as None""" diffs = self._fk_opts_fixture( {"ondelete": "no action"}, {"ondelete": "no action"} ) eq_(diffs, []) @config.requirements.fk_onupdate def test_nochange_onupdate_noaction(self): """test the NO ACTION option which generally comes back as None""" diffs = self._fk_opts_fixture( {"onupdate": "no action"}, {"onupdate": "no action"} ) eq_(diffs, []) @config.requirements.fk_ondelete_restrict def test_change_ondelete_from_restrict(self): """test the RESTRICT option which MySQL doesn't report on""" # note that this is impossible to detect if we change # from RESTRICT to NO ACTION on MySQL. diffs = self._fk_opts_fixture( {"ondelete": "restrict"}, {"ondelete": "cascade"} ) self._assert_fk_diff( diffs[0], "remove_fk", "user", ["tid"], "some_table", ["id"], onupdate=None, ondelete=mock.ANY, # MySQL reports None, PG reports RESTRICT conditional_name="servergenerated", ) self._assert_fk_diff( diffs[1], "add_fk", "user", ["tid"], "some_table", ["id"], onupdate=None, ondelete="cascade", ) @config.requirements.fk_ondelete_restrict def test_change_onupdate_from_restrict(self): """test the RESTRICT option which MySQL doesn't report on""" # note that this is impossible to detect if we change # from RESTRICT to NO ACTION on MySQL. diffs = self._fk_opts_fixture( {"onupdate": "restrict"}, {"onupdate": "cascade"} ) self._assert_fk_diff( diffs[0], "remove_fk", "user", ["tid"], "some_table", ["id"], onupdate=mock.ANY, # MySQL reports None, PG reports RESTRICT ondelete=None, conditional_name="servergenerated", ) self._assert_fk_diff( diffs[1], "add_fk", "user", ["tid"], "some_table", ["id"], onupdate="cascade", ondelete=None, ) @config.requirements.fk_ondelete_is_reflected @config.requirements.fk_onupdate_is_reflected def test_ondelete_onupdate_combo(self): diffs = self._fk_opts_fixture( {"onupdate": "CASCADE", "ondelete": "SET NULL"}, {"onupdate": "RESTRICT", "ondelete": "RESTRICT"}, ) self._assert_fk_diff( diffs[0], "remove_fk", "user", ["tid"], "some_table", ["id"], onupdate="CASCADE", ondelete="SET NULL", conditional_name="servergenerated", ) self._assert_fk_diff( diffs[1], "add_fk", "user", ["tid"], "some_table", ["id"], onupdate="RESTRICT", ondelete="RESTRICT", ) @config.requirements.fk_initially def test_add_initially_deferred(self): diffs = self._fk_opts_fixture({}, {"initially": "deferred"}) self._assert_fk_diff( diffs[0], "remove_fk", "user", ["tid"], "some_table", ["id"], initially=None, conditional_name="servergenerated", ) self._assert_fk_diff( diffs[1], "add_fk", "user", ["tid"], "some_table", ["id"], initially="deferred", ) @config.requirements.fk_initially def test_remove_initially_deferred(self): diffs = self._fk_opts_fixture({"initially": "deferred"}, {}) self._assert_fk_diff( diffs[0], "remove_fk", "user", ["tid"], "some_table", ["id"], initially="DEFERRED", deferrable=True, conditional_name="servergenerated", ) self._assert_fk_diff( diffs[1], "add_fk", "user", ["tid"], "some_table", ["id"], initially=None, ) @config.requirements.fk_deferrable @config.requirements.fk_initially def test_add_initially_immediate_plus_deferrable(self): diffs = self._fk_opts_fixture( {}, {"initially": "immediate", "deferrable": True} ) self._assert_fk_diff( diffs[0], "remove_fk", "user", ["tid"], "some_table", ["id"], initially=None, conditional_name="servergenerated", ) self._assert_fk_diff( diffs[1], "add_fk", "user", ["tid"], "some_table", ["id"], initially="immediate", deferrable=True, ) @config.requirements.fk_deferrable @config.requirements.fk_initially def test_remove_initially_immediate_plus_deferrable(self): diffs = self._fk_opts_fixture( {"initially": "immediate", "deferrable": True}, {} ) self._assert_fk_diff( diffs[0], "remove_fk", "user", ["tid"], "some_table", ["id"], initially=None, # immediate is the default deferrable=True, conditional_name="servergenerated", ) self._assert_fk_diff( diffs[1], "add_fk", "user", ["tid"], "some_table", ["id"], initially=None, deferrable=None, ) @config.requirements.fk_initially @config.requirements.fk_deferrable def test_add_initially_deferrable_nochange_one(self): diffs = self._fk_opts_fixture( {"deferrable": True, "initially": "immediate"}, {"deferrable": True, "initially": "immediate"}, ) eq_(diffs, []) @config.requirements.fk_initially @config.requirements.fk_deferrable def test_add_initially_deferrable_nochange_two(self): diffs = self._fk_opts_fixture( {"deferrable": True, "initially": "deferred"}, {"deferrable": True, "initially": "deferred"}, ) eq_(diffs, []) @config.requirements.fk_initially @config.requirements.fk_deferrable def test_add_initially_deferrable_nochange_three(self): diffs = self._fk_opts_fixture( {"deferrable": None, "initially": "deferred"}, {"deferrable": None, "initially": "deferred"}, ) eq_(diffs, []) @config.requirements.fk_deferrable def test_add_deferrable(self): diffs = self._fk_opts_fixture({}, {"deferrable": True}) self._assert_fk_diff( diffs[0], "remove_fk", "user", ["tid"], "some_table", ["id"], deferrable=None, conditional_name="servergenerated", ) self._assert_fk_diff( diffs[1], "add_fk", "user", ["tid"], "some_table", ["id"], deferrable=True, ) @config.requirements.fk_deferrable_is_reflected def test_remove_deferrable(self): diffs = self._fk_opts_fixture({"deferrable": True}, {}) self._assert_fk_diff( diffs[0], "remove_fk", "user", ["tid"], "some_table", ["id"], deferrable=True, conditional_name="servergenerated", ) self._assert_fk_diff( diffs[1], "add_fk", "user", ["tid"], "some_table", ["id"], deferrable=None, )	32.16 KB	2026-01-20 12:58
📄 test_autogen_identity.py import sqlalchemy as sa from sqlalchemy import Column from sqlalchemy import Integer from sqlalchemy import MetaData from sqlalchemy import Table 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): __requires__ = ("identity_columns",) __backend__ = True def test_add_identity_column(self): m1 = MetaData() m2 = MetaData() Table("user", m1, Column("other", sa.Text)) Table( "user", m2, Column("other", sa.Text), Column( "id", Integer, sa.Identity(start=5, increment=7), primary_key=True, ), ) diffs = self._fixture(m1, m2) eq_(diffs[0][0], "add_column") eq_(diffs[0][2], "user") eq_(diffs[0][3].name, "id") i = diffs[0][3].identity is_true(isinstance(i, sa.Identity)) eq_(i.start, 5) eq_(i.increment, 7) def test_remove_identity_column(self): m1 = MetaData() m2 = MetaData() Table( "user", m1, Column( "id", Integer, sa.Identity(start=2, increment=3), primary_key=True, ), ) Table("user", m2) diffs = self._fixture(m1, m2) eq_(diffs[0][0], "remove_column") eq_(diffs[0][2], "user") c = diffs[0][3] eq_(c.name, "id") is_true(isinstance(c.identity, sa.Identity)) eq_(c.identity.start, 2) eq_(c.identity.increment, 3) def test_no_change_identity_column(self): m1 = MetaData() m2 = MetaData() for m in (m1, m2): Table( "user", m, Column("id", Integer, sa.Identity(start=2)), ) diffs = self._fixture(m1, m2) eq_(diffs, []) @testing.combinations( (None, dict(start=2)), (dict(start=2), None), (dict(start=2), dict(start=2, increment=7)), (dict(always=False), dict(always=True)), ( dict(start=1, minvalue=0, maxvalue=100, cycle=True), dict(start=1, minvalue=0, maxvalue=100, cycle=False), ), ( dict(start=10, increment=3, maxvalue=9999), dict(start=10, increment=1, maxvalue=3333), ), ) @config.requirements.identity_columns_alter def test_change_identity(self, before, after): arg_before = (sa.Identity(before),) if before else () arg_after = (sa.Identity(after),) if after else () m1 = MetaData() m2 = MetaData() Table( "user", m1, Column("id", Integer, arg_before), Column("other", sa.Text), ) Table( "user", m2, Column("id", Integer, arg_after), Column("other", sa.Text), ) diffs = self._fixture(m1, m2) eq_(len(diffs[0]), 1) diffs = diffs[0][0] eq_(diffs[0], "modify_default") eq_(diffs[2], "user") eq_(diffs[3], "id") old = diffs[5] new = diffs[6] def check(kw, idt): if kw: is_true(isinstance(idt, sa.Identity)) for k, v in kw.items(): eq_(getattr(idt, k), v) else: is_true(idt in (None, False)) check(before, old) check(after, new) def test_add_identity_to_column(self): m1 = MetaData() m2 = MetaData() Table( "user", m1, Column("id", Integer), Column("other", sa.Text), ) Table( "user", m2, Column("id", Integer, sa.Identity(start=2, maxvalue=1000)), Column("other", sa.Text), ) diffs = self._fixture(m1, m2) eq_(len(diffs[0]), 1) diffs = diffs[0][0] eq_(diffs[0], "modify_default") eq_(diffs[2], "user") eq_(diffs[3], "id") eq_(diffs[5], None) added = diffs[6] is_true(isinstance(added, sa.Identity)) eq_(added.start, 2) eq_(added.maxvalue, 1000) def test_remove_identity_from_column(self): m1 = MetaData() m2 = MetaData() Table( "user", m1, Column("id", Integer, sa.Identity(start=2, maxvalue=1000)), Column("other", sa.Text), ) Table( "user", m2, Column("id", Integer), Column("other", sa.Text), ) diffs = self._fixture(m1, m2) eq_(len(diffs[0]), 1) diffs = diffs[0][0] eq_(diffs[0], "modify_default") eq_(diffs[2], "user") eq_(diffs[3], "id") eq_(diffs[6], None) removed = diffs[5] is_true(isinstance(removed, sa.Identity)) def test_identity_on_null(self): m1 = MetaData() m2 = MetaData() Table( "user", m1, Column("id", Integer, sa.Identity(start=2, on_null=True)), Column("other", sa.Text), ) Table( "user", m2, Column("id", Integer, sa.Identity(start=2, on_null=False)), Column("other", sa.Text), ) diffs = self._fixture(m1, m2) if not config.requirements.supports_identity_on_null.enabled: eq_(diffs, []) else: eq_(len(diffs[0]), 1) diffs = diffs[0][0] eq_(diffs[0], "modify_default") eq_(diffs[2], "user") eq_(diffs[3], "id") old = diffs[5] new = diffs[6] is_true(isinstance(old, sa.Identity)) is_true(isinstance(new, sa.Identity))	5.95 KB	2026-01-20 12:58
📄 test_environment.py import io from ...migration import MigrationContext from ...testing import assert_raises from ...testing import config from ...testing import eq_ from ...testing import is_ from ...testing import is_false from ...testing import is_not_ from ...testing import is_true from ...testing import ne_ from ...testing.fixtures import TestBase 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 ) = io.StringIO() else: self.context = MigrationContext.configure( connection=conn, opts=opts ) return self.context def teardown_method(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(): # in 1.x, self.conn is separate due to the # execution_options call. however for future they are the # same connection and there is a "transaction" block # despite autocommit if self.is_sqlalchemy_future: is_(context.connection, self.conn) else: is_not_(context.connection, self.conn) is_false(self.conn.in_transaction()) eq_( context.connection._execution_options[ "isolation_level" ], "AUTOCOMMIT", ) ne_( context.connection._execution_options.get( "isolation_level", None ), "AUTOCOMMIT", ) 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_true(context.connection.in_transaction()) # in 1.x, self.conn is separate due to the execution_options # call. however for future they are the same connection and there # is a "transaction" block despite autocommit if self.is_sqlalchemy_future: is_(context.connection, self.conn) else: is_not_(context.connection, self.conn) is_false(self.conn.in_transaction()) eq_( context.connection._execution_options["isolation_level"], "AUTOCOMMIT", ) ne_( context.connection._execution_options.get("isolation_level", None), "AUTOCOMMIT", ) 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 = io.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())	11.6 KB	2026-01-20 12:58
📄 test_op.py """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})	1.31 KB	2026-01-20 12:58

Terminal