"""Test against the builders in the op.* module.""" from sqlalchemy import Boolean from sqlalchemy import CheckConstraint from sqlalchemy import Column from sqlalchemy import exc from sqlalchemy import ForeignKey from sqlalchemy import Index from sqlalchemy import Integer from sqlalchemy import MetaData from sqlalchemy import String from sqlalchemy import Table from sqlalchemy import UniqueConstraint from sqlalchemy.sql import column from sqlalchemy.sql import func from sqlalchemy.sql import text from sqlalchemy.sql.schema import quoted_name from alembic import op from alembic.operations import ops from alembic.operations import schemaobj from alembic.testing import assert_raises_message from alembic.testing import combinations from alembic.testing import config from alembic.testing import eq_ from alembic.testing import is_not_ from alembic.testing import mock from alembic.testing.fixtures import op_fixture from alembic.testing.fixtures import TestBase from alembic.util import sqla_compat class OpTest(TestBase): def test_rename_table(self): context = op_fixture() op.rename_table("t1", "t2") context.assert_("ALTER TABLE t1 RENAME TO t2") def test_rename_table_schema(self): context = op_fixture() op.rename_table("t1", "t2", schema="foo") context.assert_("ALTER TABLE foo.t1 RENAME TO foo.t2") def test_create_index_arbitrary_expr(self): context = op_fixture() op.create_index("name", "tname", [func.foo(column("x"))]) context.assert_("CREATE INDEX name ON tname (foo(x))") def test_add_column_schema_hard_quoting(self): context = op_fixture("postgresql") op.add_column( "somename", Column("colname", String), schema=quoted_name("some.schema", quote=True), ) context.assert_( 'ALTER TABLE "some.schema".somename ADD COLUMN colname VARCHAR' ) def test_rename_table_schema_hard_quoting(self): context = op_fixture("postgresql") op.rename_table( "t1", "t2", schema=quoted_name("some.schema", quote=True) ) context.assert_('ALTER TABLE "some.schema".t1 RENAME TO t2') def test_add_constraint_schema_hard_quoting(self): context = op_fixture("postgresql") op.create_check_constraint( "ck_user_name_len", "user_table", func.len(column("name")) > 5, schema=quoted_name("some.schema", quote=True), ) context.assert_( 'ALTER TABLE "some.schema".user_table ADD ' "CONSTRAINT ck_user_name_len CHECK (len(name) > 5)" ) def test_create_index_quoting(self): context = op_fixture("postgresql") op.create_index("geocoded", "locations", ["IShouldBeQuoted"]) context.assert_( 'CREATE INDEX geocoded ON locations ("IShouldBeQuoted")' ) def test_create_index_expressions(self): context = op_fixture() op.create_index("geocoded", "locations", [text("lower(coordinates)")]) context.assert_( "CREATE INDEX geocoded ON locations (lower(coordinates))" ) def test_add_column(self): context = op_fixture() op.add_column("t1", Column("c1", Integer, nullable=False)) context.assert_("ALTER TABLE t1 ADD COLUMN c1 INTEGER NOT NULL") def test_add_column_already_attached(self): context = op_fixture() c1 = Column("c1", Integer, nullable=False) Table("t", MetaData(), c1) op.add_column("t1", c1) context.assert_("ALTER TABLE t1 ADD COLUMN c1 INTEGER NOT NULL") def test_add_column_w_check(self): context = op_fixture() op.add_column( "t1", Column("c1", Integer, CheckConstraint("c1 > 5"), nullable=False), ) context.assert_( "ALTER TABLE t1 ADD COLUMN c1 INTEGER NOT NULL CHECK (c1 > 5)" ) def test_add_column_schema(self): context = op_fixture() op.add_column( "t1", Column("c1", Integer, nullable=False), schema="foo" ) context.assert_("ALTER TABLE foo.t1 ADD COLUMN c1 INTEGER NOT NULL") def test_add_column_with_default(self): context = op_fixture() op.add_column( "t1", Column("c1", Integer, nullable=False, server_default="12") ) context.assert_( "ALTER TABLE t1 ADD COLUMN c1 INTEGER DEFAULT '12' NOT NULL" ) def test_add_column_with_index(self): context = op_fixture() op.add_column("t1", Column("c1", Integer, nullable=False, index=True)) context.assert_( "ALTER TABLE t1 ADD COLUMN c1 INTEGER NOT NULL", "CREATE INDEX ix_t1_c1 ON t1 (c1)", ) def test_add_column_schema_with_default(self): context = op_fixture() op.add_column( "t1", Column("c1", Integer, nullable=False, server_default="12"), schema="foo", ) context.assert_( "ALTER TABLE foo.t1 ADD COLUMN c1 INTEGER DEFAULT '12' NOT NULL" ) def test_add_column_fk(self): context = op_fixture() op.add_column( "t1", Column("c1", Integer, ForeignKey("c2.id"), nullable=False) ) context.assert_( "ALTER TABLE t1 ADD COLUMN c1 INTEGER NOT NULL", "ALTER TABLE t1 ADD FOREIGN KEY(c1) REFERENCES c2 (id)", ) def test_add_column_schema_fk(self): context = op_fixture() op.add_column( "t1", Column("c1", Integer, ForeignKey("c2.id"), nullable=False), schema="foo", ) context.assert_( "ALTER TABLE foo.t1 ADD COLUMN c1 INTEGER NOT NULL", "ALTER TABLE foo.t1 ADD FOREIGN KEY(c1) REFERENCES c2 (id)", ) def test_add_column_schema_type(self): """Test that a schema type generates its constraints....""" context = op_fixture() op.add_column( "t1", Column("c1", Boolean(create_constraint=True), nullable=False) ) context.assert_( "ALTER TABLE t1 ADD COLUMN c1 BOOLEAN NOT NULL", "ALTER TABLE t1 ADD CHECK (c1 IN (0, 1))", ) def test_add_column_schema_schema_type(self): """Test that a schema type generates its constraints....""" context = op_fixture() op.add_column( "t1", Column("c1", Boolean(create_constraint=True), nullable=False), schema="foo", ) context.assert_( "ALTER TABLE foo.t1 ADD COLUMN c1 BOOLEAN NOT NULL", "ALTER TABLE foo.t1 ADD CHECK (c1 IN (0, 1))", ) def test_add_column_schema_type_checks_rule(self): """Test that a schema type doesn't generate a constraint based on check rule.""" context = op_fixture("postgresql") op.add_column( "t1", Column("c1", Boolean(create_constraint=True), nullable=False) ) context.assert_("ALTER TABLE t1 ADD COLUMN c1 BOOLEAN NOT NULL") def test_add_column_fk_self_referential(self): context = op_fixture() op.add_column( "t1", Column("c1", Integer, ForeignKey("t1.c2"), nullable=False) ) context.assert_( "ALTER TABLE t1 ADD COLUMN c1 INTEGER NOT NULL", "ALTER TABLE t1 ADD FOREIGN KEY(c1) REFERENCES t1 (c2)", ) def test_add_column_schema_fk_self_referential(self): context = op_fixture() op.add_column( "t1", Column("c1", Integer, ForeignKey("foo.t1.c2"), nullable=False), schema="foo", ) context.assert_( "ALTER TABLE foo.t1 ADD COLUMN c1 INTEGER NOT NULL", "ALTER TABLE foo.t1 ADD FOREIGN KEY(c1) REFERENCES foo.t1 (c2)", ) def test_add_column_fk_schema(self): context = op_fixture() op.add_column( "t1", Column("c1", Integer, ForeignKey("remote.t2.c2"), nullable=False), ) context.assert_( "ALTER TABLE t1 ADD COLUMN c1 INTEGER NOT NULL", "ALTER TABLE t1 ADD FOREIGN KEY(c1) REFERENCES remote.t2 (c2)", ) def test_add_column_schema_fk_schema(self): context = op_fixture() op.add_column( "t1", Column("c1", Integer, ForeignKey("remote.t2.c2"), nullable=False), schema="foo", ) context.assert_( "ALTER TABLE foo.t1 ADD COLUMN c1 INTEGER NOT NULL", "ALTER TABLE foo.t1 ADD FOREIGN KEY(c1) REFERENCES remote.t2 (c2)", ) def test_drop_column(self): context = op_fixture() op.drop_column("t1", "c1") context.assert_("ALTER TABLE t1 DROP COLUMN c1") def test_drop_column_schema(self): context = op_fixture() op.drop_column("t1", "c1", schema="foo") context.assert_("ALTER TABLE foo.t1 DROP COLUMN c1") def test_alter_column_nullable(self): context = op_fixture() op.alter_column("t", "c", nullable=True) context.assert_( # TODO: not sure if this is PG only or standard # SQL "ALTER TABLE t ALTER COLUMN c DROP NOT NULL" ) def test_alter_column_schema_nullable(self): context = op_fixture() op.alter_column("t", "c", nullable=True, schema="foo") context.assert_( # TODO: not sure if this is PG only or standard # SQL "ALTER TABLE foo.t ALTER COLUMN c DROP NOT NULL" ) def test_alter_column_not_nullable(self): context = op_fixture() op.alter_column("t", "c", nullable=False) context.assert_( # TODO: not sure if this is PG only or standard # SQL "ALTER TABLE t ALTER COLUMN c SET NOT NULL" ) def test_alter_column_schema_not_nullable(self): context = op_fixture() op.alter_column("t", "c", nullable=False, schema="foo") context.assert_( # TODO: not sure if this is PG only or standard # SQL "ALTER TABLE foo.t ALTER COLUMN c SET NOT NULL" ) def test_alter_column_rename(self): context = op_fixture() op.alter_column("t", "c", new_column_name="x") context.assert_("ALTER TABLE t RENAME c TO x") def test_alter_column_schema_rename(self): context = op_fixture() op.alter_column("t", "c", new_column_name="x", schema="foo") context.assert_("ALTER TABLE foo.t RENAME c TO x") def test_alter_column_type(self): context = op_fixture() op.alter_column("t", "c", type_=String(50)) context.assert_("ALTER TABLE t ALTER COLUMN c TYPE VARCHAR(50)") def test_alter_column_schema_type(self): context = op_fixture() op.alter_column("t", "c", type_=String(50), schema="foo") context.assert_("ALTER TABLE foo.t ALTER COLUMN c TYPE VARCHAR(50)") def test_alter_column_set_default(self): context = op_fixture() op.alter_column("t", "c", server_default="q") context.assert_("ALTER TABLE t ALTER COLUMN c SET DEFAULT 'q'") def test_alter_column_schema_set_default(self): context = op_fixture() op.alter_column("t", "c", server_default="q", schema="foo") context.assert_("ALTER TABLE foo.t ALTER COLUMN c SET DEFAULT 'q'") def test_alter_column_set_compiled_default(self): context = op_fixture() op.alter_column( "t", "c", server_default=func.utc_thing(func.current_timestamp()) ) context.assert_( "ALTER TABLE t ALTER COLUMN c " "SET DEFAULT utc_thing(CURRENT_TIMESTAMP)" ) def test_alter_column_schema_set_compiled_default(self): context = op_fixture() op.alter_column( "t", "c", server_default=func.utc_thing(func.current_timestamp()), schema="foo", ) context.assert_( "ALTER TABLE foo.t ALTER COLUMN c " "SET DEFAULT utc_thing(CURRENT_TIMESTAMP)" ) def test_alter_column_drop_default(self): context = op_fixture() op.alter_column("t", "c", server_default=None) context.assert_("ALTER TABLE t ALTER COLUMN c DROP DEFAULT") def test_alter_column_schema_drop_default(self): context = op_fixture() op.alter_column("t", "c", server_default=None, schema="foo") context.assert_("ALTER TABLE foo.t ALTER COLUMN c DROP DEFAULT") @combinations( (lambda: sqla_compat.Computed("foo * 5"), lambda: None), (lambda: None, lambda: sqla_compat.Computed("foo * 5")), ( lambda: sqla_compat.Computed("foo * 42"), lambda: sqla_compat.Computed("foo * 5"), ), ) @config.requirements.computed_columns_api def test_alter_column_computed_not_supported(self, sd, esd): op_fixture() assert_raises_message( exc.CompileError, 'Adding or removing a "computed" construct, e.g. ' "GENERATED ALWAYS AS, to or from an existing column is not " "supported.", op.alter_column, "t1", "c1", server_default=sd(), existing_server_default=esd(), ) @combinations( (lambda: sqla_compat.Identity(), lambda: None), (lambda: None, lambda: sqla_compat.Identity()), ( lambda: sqla_compat.Identity(), lambda: sqla_compat.Identity(), ), ) @config.requirements.identity_columns_api def test_alter_column_identity_not_supported(self, sd, esd): op_fixture() assert_raises_message( exc.CompileError, 'Adding, removing or modifying an "identity" construct, ' "e.g. GENERATED AS IDENTITY, to or from an existing " "column is not supported in this dialect.", op.alter_column, "t1", "c1", server_default=sd(), existing_server_default=esd(), ) def test_alter_column_schema_type_unnamed(self): context = op_fixture("mssql", native_boolean=False) op.alter_column("t", "c", type_=Boolean(create_constraint=True)) context.assert_( "ALTER TABLE t ALTER COLUMN c BIT", "ALTER TABLE t ADD CHECK (c IN (0, 1))", ) def test_alter_column_schema_schema_type_unnamed(self): context = op_fixture("mssql", native_boolean=False) op.alter_column( "t", "c", type_=Boolean(create_constraint=True), schema="foo" ) context.assert_( "ALTER TABLE foo.t ALTER COLUMN c BIT", "ALTER TABLE foo.t ADD CHECK (c IN (0, 1))", ) def test_alter_column_schema_type_named(self): context = op_fixture("mssql", native_boolean=False) op.alter_column( "t", "c", type_=Boolean(name="xyz", create_constraint=True) ) context.assert_( "ALTER TABLE t ALTER COLUMN c BIT", "ALTER TABLE t ADD CONSTRAINT xyz CHECK (c IN (0, 1))", ) def test_alter_column_schema_schema_type_named(self): context = op_fixture("mssql", native_boolean=False) op.alter_column( "t", "c", type_=Boolean(name="xyz", create_constraint=True), schema="foo", ) context.assert_( "ALTER TABLE foo.t ALTER COLUMN c BIT", "ALTER TABLE foo.t ADD CONSTRAINT xyz CHECK (c IN (0, 1))", ) @combinations((True,), (False,), argnames="pass_existing_type") @combinations((True,), (False,), argnames="change_nullability") def test_generic_alter_column_type_and_nullability( self, pass_existing_type, change_nullability ): # this test is also on the mssql dialect in test_mssql context = op_fixture() args = dict(type_=Integer) if pass_existing_type: args["existing_type"] = String(15) if change_nullability: args["nullable"] = False op.alter_column("t", "c", **args) if change_nullability: context.assert_( "ALTER TABLE t ALTER COLUMN c SET NOT NULL", "ALTER TABLE t ALTER COLUMN c TYPE INTEGER", ) else: context.assert_("ALTER TABLE t ALTER COLUMN c TYPE INTEGER") def test_alter_column_schema_type_existing_type(self): context = op_fixture("mssql", native_boolean=False) op.alter_column( "t", "c", type_=String(10), existing_type=Boolean(name="xyz", create_constraint=True), ) context.assert_( "ALTER TABLE t DROP CONSTRAINT xyz", "ALTER TABLE t ALTER COLUMN c VARCHAR(10)", ) def test_alter_column_schema_schema_type_existing_type(self): context = op_fixture("mssql", native_boolean=False) op.alter_column( "t", "c", type_=String(10), existing_type=Boolean(name="xyz", create_constraint=True), schema="foo", ) context.assert_( "ALTER TABLE foo.t DROP CONSTRAINT xyz", "ALTER TABLE foo.t ALTER COLUMN c VARCHAR(10)", ) def test_alter_column_schema_type_existing_type_no_const(self): context = op_fixture("postgresql") op.alter_column("t", "c", type_=String(10), existing_type=Boolean()) context.assert_("ALTER TABLE t ALTER COLUMN c TYPE VARCHAR(10)") def test_alter_column_schema_schema_type_existing_type_no_const(self): context = op_fixture("postgresql") op.alter_column( "t", "c", type_=String(10), existing_type=Boolean(), schema="foo" ) context.assert_("ALTER TABLE foo.t ALTER COLUMN c TYPE VARCHAR(10)") def test_alter_column_schema_type_existing_type_no_new_type(self): context = op_fixture("postgresql") op.alter_column("t", "c", nullable=False, existing_type=Boolean()) context.assert_("ALTER TABLE t ALTER COLUMN c SET NOT NULL") def test_alter_column_schema_schema_type_existing_type_no_new_type(self): context = op_fixture("postgresql") op.alter_column( "t", "c", nullable=False, existing_type=Boolean(), schema="foo" ) context.assert_("ALTER TABLE foo.t ALTER COLUMN c SET NOT NULL") def test_add_foreign_key(self): context = op_fixture() op.create_foreign_key( "fk_test", "t1", "t2", ["foo", "bar"], ["bat", "hoho"] ) context.assert_( "ALTER TABLE t1 ADD CONSTRAINT fk_test FOREIGN KEY(foo, bar) " "REFERENCES t2 (bat, hoho)" ) def test_add_foreign_key_schema(self): context = op_fixture() op.create_foreign_key( "fk_test", "t1", "t2", ["foo", "bar"], ["bat", "hoho"], source_schema="foo2", referent_schema="bar2", ) context.assert_( "ALTER TABLE foo2.t1 ADD CONSTRAINT fk_test FOREIGN KEY(foo, bar) " "REFERENCES bar2.t2 (bat, hoho)" ) def test_add_foreign_key_schema_same_tablename(self): context = op_fixture() op.create_foreign_key( "fk_test", "t1", "t1", ["foo", "bar"], ["bat", "hoho"], source_schema="foo2", referent_schema="bar2", ) context.assert_( "ALTER TABLE foo2.t1 ADD CONSTRAINT fk_test FOREIGN KEY(foo, bar) " "REFERENCES bar2.t1 (bat, hoho)" ) def test_add_foreign_key_onupdate(self): context = op_fixture() op.create_foreign_key( "fk_test", "t1", "t2", ["foo", "bar"], ["bat", "hoho"], onupdate="CASCADE", ) context.assert_( "ALTER TABLE t1 ADD CONSTRAINT fk_test FOREIGN KEY(foo, bar) " "REFERENCES t2 (bat, hoho) ON UPDATE CASCADE" ) def test_add_foreign_key_ondelete(self): context = op_fixture() op.create_foreign_key( "fk_test", "t1", "t2", ["foo", "bar"], ["bat", "hoho"], ondelete="CASCADE", ) context.assert_( "ALTER TABLE t1 ADD CONSTRAINT fk_test FOREIGN KEY(foo, bar) " "REFERENCES t2 (bat, hoho) ON DELETE CASCADE" ) def test_add_foreign_key_deferrable(self): context = op_fixture() op.create_foreign_key( "fk_test", "t1", "t2", ["foo", "bar"], ["bat", "hoho"], deferrable=True, ) context.assert_( "ALTER TABLE t1 ADD CONSTRAINT fk_test FOREIGN KEY(foo, bar) " "REFERENCES t2 (bat, hoho) DEFERRABLE" ) def test_add_foreign_key_initially(self): context = op_fixture() op.create_foreign_key( "fk_test", "t1", "t2", ["foo", "bar"], ["bat", "hoho"], initially="deferred", ) context.assert_( "ALTER TABLE t1 ADD CONSTRAINT fk_test FOREIGN KEY(foo, bar) " "REFERENCES t2 (bat, hoho) INITIALLY deferred" ) @config.requirements.foreign_key_match def test_add_foreign_key_match(self): context = op_fixture() op.create_foreign_key( "fk_test", "t1", "t2", ["foo", "bar"], ["bat", "hoho"], match="SIMPLE", ) context.assert_( "ALTER TABLE t1 ADD CONSTRAINT fk_test FOREIGN KEY(foo, bar) " "REFERENCES t2 (bat, hoho) MATCH SIMPLE" ) def test_add_foreign_key_dialect_kw(self): op_fixture() with mock.patch("sqlalchemy.schema.ForeignKeyConstraint") as fkc: op.create_foreign_key( "fk_test", "t1", "t2", ["foo", "bar"], ["bat", "hoho"], foobar_arg="xyz", ) if config.requirements.foreign_key_match.enabled: eq_( fkc.mock_calls[0], mock.call( ["foo", "bar"], ["t2.bat", "t2.hoho"], onupdate=None, ondelete=None, name="fk_test", foobar_arg="xyz", deferrable=None, initially=None, match=None, ), ) else: eq_( fkc.mock_calls[0], mock.call( ["foo", "bar"], ["t2.bat", "t2.hoho"], onupdate=None, ondelete=None, name="fk_test", foobar_arg="xyz", deferrable=None, initially=None, ), ) def test_add_foreign_key_self_referential(self): context = op_fixture() op.create_foreign_key("fk_test", "t1", "t1", ["foo"], ["bar"]) context.assert_( "ALTER TABLE t1 ADD CONSTRAINT fk_test " "FOREIGN KEY(foo) REFERENCES t1 (bar)" ) def test_add_primary_key_constraint(self): context = op_fixture() op.create_primary_key("pk_test", "t1", ["foo", "bar"]) context.assert_( "ALTER TABLE t1 ADD CONSTRAINT pk_test PRIMARY KEY (foo, bar)" ) def test_add_primary_key_constraint_schema(self): context = op_fixture() op.create_primary_key("pk_test", "t1", ["foo"], schema="bar") context.assert_( "ALTER TABLE bar.t1 ADD CONSTRAINT pk_test PRIMARY KEY (foo)" ) def test_add_check_constraint(self): context = op_fixture() op.create_check_constraint( "ck_user_name_len", "user_table", func.len(column("name")) > 5 ) context.assert_( "ALTER TABLE user_table ADD CONSTRAINT ck_user_name_len " "CHECK (len(name) > 5)" ) def test_add_check_constraint_schema(self): context = op_fixture() op.create_check_constraint( "ck_user_name_len", "user_table", func.len(column("name")) > 5, schema="foo", ) context.assert_( "ALTER TABLE foo.user_table ADD CONSTRAINT ck_user_name_len " "CHECK (len(name) > 5)" ) def test_add_unique_constraint(self): context = op_fixture() op.create_unique_constraint("uk_test", "t1", ["foo", "bar"]) context.assert_( "ALTER TABLE t1 ADD CONSTRAINT uk_test UNIQUE (foo, bar)" ) def test_add_unique_constraint_schema(self): context = op_fixture() op.create_unique_constraint( "uk_test", "t1", ["foo", "bar"], schema="foo" ) context.assert_( "ALTER TABLE foo.t1 ADD CONSTRAINT uk_test UNIQUE (foo, bar)" ) def test_drop_constraint(self): context = op_fixture() op.drop_constraint("foo_bar_bat", "t1") context.assert_("ALTER TABLE t1 DROP CONSTRAINT foo_bar_bat") def test_drop_constraint_schema(self): context = op_fixture() op.drop_constraint("foo_bar_bat", "t1", schema="foo") context.assert_("ALTER TABLE foo.t1 DROP CONSTRAINT foo_bar_bat") def test_create_index(self): context = op_fixture() op.create_index("ik_test", "t1", ["foo", "bar"]) context.assert_("CREATE INDEX ik_test ON t1 (foo, bar)") def test_create_unique_index(self): context = op_fixture() op.create_index("ik_test", "t1", ["foo", "bar"], unique=True) context.assert_("CREATE UNIQUE INDEX ik_test ON t1 (foo, bar)") def test_create_index_quote_flag(self): context = op_fixture() op.create_index("ik_test", "t1", ["foo", "bar"], quote=True) context.assert_('CREATE INDEX "ik_test" ON t1 (foo, bar)') def test_create_index_table_col_event(self): context = op_fixture() op.create_index( "ik_test", "tbl_with_auto_appended_column", ["foo", "bar"] ) context.assert_( "CREATE INDEX ik_test ON tbl_with_auto_appended_column (foo, bar)" ) def test_add_unique_constraint_col_event(self): context = op_fixture() op.create_unique_constraint( "ik_test", "tbl_with_auto_appended_column", ["foo", "bar"] ) context.assert_( "ALTER TABLE tbl_with_auto_appended_column " "ADD CONSTRAINT ik_test UNIQUE (foo, bar)" ) def test_create_index_schema(self): context = op_fixture() op.create_index("ik_test", "t1", ["foo", "bar"], schema="foo") context.assert_("CREATE INDEX ik_test ON foo.t1 (foo, bar)") def test_drop_index(self): context = op_fixture() op.drop_index("ik_test") context.assert_("DROP INDEX ik_test") def test_drop_index_schema(self): context = op_fixture() op.drop_index("ik_test", schema="foo") context.assert_("DROP INDEX foo.ik_test") def test_drop_table(self): context = op_fixture() op.drop_table("tb_test") context.assert_("DROP TABLE tb_test") def test_drop_table_schema(self): context = op_fixture() op.drop_table("tb_test", schema="foo") context.assert_("DROP TABLE foo.tb_test") def test_create_table_selfref(self): context = op_fixture() op.create_table( "some_table", Column("id", Integer, primary_key=True), Column("st_id", Integer, ForeignKey("some_table.id")), ) context.assert_( "CREATE TABLE some_table (" "id INTEGER NOT NULL, " "st_id INTEGER, " "PRIMARY KEY (id), " "FOREIGN KEY(st_id) REFERENCES some_table (id))" ) def test_create_table_check_constraint(self): context = op_fixture() t1 = op.create_table( "some_table", Column("id", Integer, primary_key=True), Column("foo_id", Integer), CheckConstraint("foo_id>5", name="ck_1"), ) context.assert_( "CREATE TABLE some_table (" "id INTEGER NOT NULL, " "foo_id INTEGER, " "PRIMARY KEY (id), " "CONSTRAINT ck_1 CHECK (foo_id>5))" ) ck = [c for c in t1.constraints if isinstance(c, CheckConstraint)] eq_(ck[0].name, "ck_1") def test_create_table_unique_constraint(self): context = op_fixture() t1 = op.create_table( "some_table", Column("id", Integer, primary_key=True), Column("foo_id", Integer), UniqueConstraint("foo_id", name="uq_1"), ) context.assert_( "CREATE TABLE some_table (" "id INTEGER NOT NULL, " "foo_id INTEGER, " "PRIMARY KEY (id), " "CONSTRAINT uq_1 UNIQUE (foo_id))" ) uq = [c for c in t1.constraints if isinstance(c, UniqueConstraint)] eq_(uq[0].name, "uq_1") def test_create_table_unique_flag(self): context = op_fixture() t1 = op.create_table( "some_table", Column("id", Integer, primary_key=True), Column("foo_id", Integer, unique=True), ) context.assert_( "CREATE TABLE some_table (id INTEGER NOT NULL, foo_id INTEGER, " "PRIMARY KEY (id), UNIQUE (foo_id))" ) uq = [c for c in t1.constraints if isinstance(c, UniqueConstraint)] assert uq def test_create_table_index_flag(self): context = op_fixture() t1 = op.create_table( "some_table", Column("id", Integer, primary_key=True), Column("foo_id", Integer, index=True), ) context.assert_( "CREATE TABLE some_table (id INTEGER NOT NULL, foo_id INTEGER, " "PRIMARY KEY (id))", "CREATE INDEX ix_some_table_foo_id ON some_table (foo_id)", ) assert t1.indexes def test_create_table_index(self): context = op_fixture() t1 = op.create_table( "some_table", Column("id", Integer, primary_key=True), Column("foo_id", Integer), Index("ix_1", "foo_id"), ) context.assert_( "CREATE TABLE some_table (" "id INTEGER NOT NULL, " "foo_id INTEGER, " "PRIMARY KEY (id))", "CREATE INDEX ix_1 ON some_table (foo_id)", ) ix = list(t1.indexes) eq_(ix[0].name, "ix_1") def test_create_table_fk_and_schema(self): context = op_fixture() t1 = op.create_table( "some_table", Column("id", Integer, primary_key=True), Column("foo_id", Integer, ForeignKey("foo.id")), schema="schema", ) context.assert_( "CREATE TABLE schema.some_table (" "id INTEGER NOT NULL, " "foo_id INTEGER, " "PRIMARY KEY (id), " "FOREIGN KEY(foo_id) REFERENCES foo (id))" ) eq_(t1.c.id.name, "id") eq_(t1.schema, "schema") def test_create_table_no_pk(self): context = op_fixture() t1 = op.create_table( "some_table", Column("x", Integer), Column("y", Integer), Column("z", Integer), ) context.assert_( "CREATE TABLE some_table (x INTEGER, y INTEGER, z INTEGER)" ) assert not t1.primary_key def test_create_table_two_fk(self): context = op_fixture() op.create_table( "some_table", Column("id", Integer, primary_key=True), Column("foo_id", Integer, ForeignKey("foo.id")), Column("foo_bar", Integer, ForeignKey("foo.bar")), ) context.assert_( "CREATE TABLE some_table (" "id INTEGER NOT NULL, " "foo_id INTEGER, " "foo_bar INTEGER, " "PRIMARY KEY (id), " "FOREIGN KEY(foo_id) REFERENCES foo (id), " "FOREIGN KEY(foo_bar) REFERENCES foo (bar))" ) def test_inline_literal(self): context = op_fixture() from sqlalchemy.sql import table, column from sqlalchemy import String, Integer account = table( "account", column("name", String), column("id", Integer) ) op.execute( account.update() .where(account.c.name == op.inline_literal("account 1")) .values({"name": op.inline_literal("account 2")}) ) op.execute( account.update() .where(account.c.id == op.inline_literal(1)) .values({"id": op.inline_literal(2)}) ) context.assert_( "UPDATE account SET name='account 2' " "WHERE account.name = 'account 1'", "UPDATE account SET id=2 WHERE account.id = 1", ) def test_cant_op(self): if hasattr(op, "_proxy"): del op._proxy assert_raises_message( NameError, "Can't invoke function 'inline_literal', as the " "proxy object has not yet been established " "for the Alembic 'Operations' class. " "Try placing this code inside a callable.", op.inline_literal, "asdf", ) def test_naming_changes(self): context = op_fixture() op.alter_column("t", "c", new_column_name="x") context.assert_("ALTER TABLE t RENAME c TO x") context = op_fixture("mysql") op.drop_constraint("f1", "t1", type_="foreignkey") context.assert_("ALTER TABLE t1 DROP FOREIGN KEY f1") def test_naming_changes_drop_idx(self): context = op_fixture("mssql") op.drop_index("ik_test", table_name="t1") context.assert_("DROP INDEX ik_test ON t1") @config.requirements.comments def test_create_table_comment_op(self): context = op_fixture() op.create_table_comment("some_table", "table comment") context.assert_("COMMENT ON TABLE some_table IS 'table comment'") @config.requirements.comments def test_drop_table_comment_op(self): context = op_fixture() op.drop_table_comment("some_table") context.assert_("COMMENT ON TABLE some_table IS NULL") class SQLModeOpTest(TestBase): def test_auto_literals(self): context = op_fixture(as_sql=True, literal_binds=True) from sqlalchemy.sql import table, column from sqlalchemy import String, Integer account = table( "account", column("name", String), column("id", Integer) ) op.execute( account.update() .where(account.c.name == op.inline_literal("account 1")) .values({"name": op.inline_literal("account 2")}) ) op.execute(text("update table set foo=:bar").bindparams(bar="bat")) context.assert_( "UPDATE account SET name='account 2' " "WHERE account.name = 'account 1'", "update table set foo='bat'", ) def test_create_table_literal_binds(self): context = op_fixture(as_sql=True, literal_binds=True) op.create_table( "some_table", Column("id", Integer, primary_key=True), Column("st_id", Integer, ForeignKey("some_table.id")), ) context.assert_( "CREATE TABLE some_table (id INTEGER NOT NULL, st_id INTEGER, " "PRIMARY KEY (id), FOREIGN KEY(st_id) REFERENCES some_table (id))" ) class CustomOpTest(TestBase): def test_custom_op(self): from alembic.operations import Operations, MigrateOperation @Operations.register_operation("create_sequence") class CreateSequenceOp(MigrateOperation): """Create a SEQUENCE.""" def __init__(self, sequence_name, **kw): self.sequence_name = sequence_name self.kw = kw @classmethod def create_sequence(cls, operations, sequence_name, **kw): """Issue a "CREATE SEQUENCE" instruction.""" op = CreateSequenceOp(sequence_name, **kw) return operations.invoke(op) @Operations.implementation_for(CreateSequenceOp) def create_sequence(operations, operation): operations.execute("CREATE SEQUENCE %s" % operation.sequence_name) context = op_fixture() op.create_sequence("foob") context.assert_("CREATE SEQUENCE foob") class ObjectFromToTest(TestBase): """Test operation round trips for to_obj() / from_obj(). Previously, these needed to preserve the "original" item to this, but this makes them harder to work with. As of #803 the constructs try to behave more intelligently about the state they were given, so that they can both "reverse" themselves but also take into accout their current state. """ def test_drop_index(self): schema_obj = schemaobj.SchemaObjects() idx = schema_obj.index("x", "y", ["z"]) op = ops.DropIndexOp.from_index(idx) is_not_(op.to_index(), idx) def test_drop_index_add_kw(self): schema_obj = schemaobj.SchemaObjects() idx = schema_obj.index("x", "y", ["z"]) op = ops.DropIndexOp.from_index(idx) op.kw["postgresql_concurrently"] = True eq_(op.to_index().dialect_kwargs["postgresql_concurrently"], True) eq_( op.reverse().to_index().dialect_kwargs["postgresql_concurrently"], True, ) def test_create_index(self): schema_obj = schemaobj.SchemaObjects() idx = schema_obj.index("x", "y", ["z"]) op = ops.CreateIndexOp.from_index(idx) is_not_(op.to_index(), idx) def test_create_index_add_kw(self): schema_obj = schemaobj.SchemaObjects() idx = schema_obj.index("x", "y", ["z"]) op = ops.CreateIndexOp.from_index(idx) op.kw["postgresql_concurrently"] = True eq_(op.to_index().dialect_kwargs["postgresql_concurrently"], True) eq_( op.reverse().to_index().dialect_kwargs["postgresql_concurrently"], True, ) def test_drop_table(self): schema_obj = schemaobj.SchemaObjects() table = schema_obj.table( "x", Column("q", Integer), info={"custom": "value"}, prefixes=["FOREIGN"], postgresql_partition_by="x", comment="some comment", ) op = ops.DropTableOp.from_table(table) is_not_(op.to_table(), table) eq_(op.to_table().comment, table.comment) eq_(op.to_table().info, table.info) eq_(op.to_table()._prefixes, table._prefixes) def test_drop_table_add_kw(self): schema_obj = schemaobj.SchemaObjects() table = schema_obj.table("x", Column("q", Integer)) op = ops.DropTableOp.from_table(table) op.table_kw["postgresql_partition_by"] = "x" eq_(op.to_table().dialect_kwargs["postgresql_partition_by"], "x") eq_( op.reverse().to_table().dialect_kwargs["postgresql_partition_by"], "x", ) def test_create_table(self): schema_obj = schemaobj.SchemaObjects() table = schema_obj.table( "x", Column("q", Integer), postgresql_partition_by="x", prefixes=["FOREIGN"], info={"custom": "value"}, comment="some comment", ) op = ops.CreateTableOp.from_table(table) is_not_(op.to_table(), table) eq_(op.to_table().comment, table.comment) eq_(op.to_table().info, table.info) eq_(op.to_table()._prefixes, table._prefixes) def test_create_table_add_kw(self): schema_obj = schemaobj.SchemaObjects() table = schema_obj.table("x", Column("q", Integer)) op = ops.CreateTableOp.from_table(table) op.kw["postgresql_partition_by"] = "x" eq_(op.to_table().dialect_kwargs["postgresql_partition_by"], "x") eq_( op.reverse().to_table().dialect_kwargs["postgresql_partition_by"], "x", ) def test_create_unique_constraint(self): schema_obj = schemaobj.SchemaObjects() const = schema_obj.unique_constraint("x", "foobar", ["a"]) op = ops.AddConstraintOp.from_constraint(const) is_not_(op.to_constraint(), const) def test_create_unique_constraint_add_kw(self): schema_obj = schemaobj.SchemaObjects() const = schema_obj.unique_constraint("x", "foobar", ["a"]) op = ops.AddConstraintOp.from_constraint(const) is_not_(op.to_constraint(), const) op.kw["sqlite_on_conflict"] = "IGNORE" eq_(op.to_constraint().dialect_kwargs["sqlite_on_conflict"], "IGNORE") eq_( op.reverse().to_constraint().dialect_kwargs["sqlite_on_conflict"], "IGNORE", ) def test_drop_unique_constraint(self): schema_obj = schemaobj.SchemaObjects() const = schema_obj.unique_constraint("x", "foobar", ["a"]) op = ops.DropConstraintOp.from_constraint(const) is_not_(op.to_constraint(), const) def test_drop_unique_constraint_change_name(self): schema_obj = schemaobj.SchemaObjects() const = schema_obj.unique_constraint("x", "foobar", ["a"]) op = ops.DropConstraintOp.from_constraint(const) op.constraint_name = "my_name" eq_(op.to_constraint().name, "my_name") eq_(op.reverse().to_constraint().name, "my_name") def test_drop_constraint_not_available(self): op = ops.DropConstraintOp("x", "y", type_="unique") assert_raises_message( ValueError, "constraint cannot be produced", op.to_constraint )