path: root/tests/test_oracle.py

from sqlalchemy import Column
from sqlalchemy import exc
from sqlalchemy import Integer

from alembic import command
from alembic import op
from alembic.testing import assert_raises_message
from alembic.testing import combinations
from alembic.testing import config
from alembic.testing.env import _no_sql_testing_config
from alembic.testing.env import clear_staging_env
from alembic.testing.env import staging_env
from alembic.testing.env import three_rev_fixture
from alembic.testing.fixtures import capture_context_buffer
from alembic.testing.fixtures import op_fixture
from alembic.testing.fixtures import TestBase
from alembic.util import sqla_compat


class FullEnvironmentTests(TestBase):
    @classmethod
    def setup_class(cls):
        staging_env()
        cls.cfg = cfg = _no_sql_testing_config("oracle")

        cls.a, cls.b, cls.c = three_rev_fixture(cfg)

    @classmethod
    def teardown_class(cls):
        clear_staging_env()

    def test_begin_comit(self):
        with capture_context_buffer(transactional_ddl=True) as buf:
            command.upgrade(self.cfg, self.a, sql=True)
        assert "SET TRANSACTION READ WRITE\n\n/" in buf.getvalue()
        assert "COMMIT\n\n/" in buf.getvalue()

    def test_batch_separator_default(self):
        with capture_context_buffer() as buf:
            command.upgrade(self.cfg, self.a, sql=True)
        assert "/" in buf.getvalue()
        assert ";" not in buf.getvalue()

    def test_batch_separator_custom(self):
        with capture_context_buffer(oracle_batch_separator="BYE") as buf:
            command.upgrade(self.cfg, self.a, sql=True)
        assert "BYE" in buf.getvalue()


class OpTest(TestBase):
    def test_add_column(self):
        context = op_fixture("oracle")
        op.add_column("t1", Column("c1", Integer, nullable=False))
        context.assert_("ALTER TABLE t1 ADD c1 INTEGER NOT NULL")

    def test_add_column_with_default(self):
        context = op_fixture("oracle")
        op.add_column(
            "t1", Column("c1", Integer, nullable=False, server_default="12")
        )
        context.assert_("ALTER TABLE t1 ADD c1 INTEGER DEFAULT '12' NOT NULL")

    @config.requirements.comments
    def test_add_column_with_comment(self):
        context = op_fixture("oracle")
        op.add_column(
            "t1", Column("c1", Integer, nullable=False, comment="c1 comment")
        )
        context.assert_(
            "ALTER TABLE t1 ADD c1 INTEGER NOT NULL",
            "COMMENT ON COLUMN t1.c1 IS 'c1 comment'",
        )

    @config.requirements.computed_columns
    def test_add_column_computed(self):
        context = op_fixture("oracle")
        op.add_column(
            "t1",
            Column("some_column", Integer, sqla_compat.Computed("foo * 5")),
        )
        context.assert_(
            "ALTER TABLE t1 ADD some_column "
            "INTEGER GENERATED ALWAYS AS (foo * 5)"
        )

    @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
    def test_alter_column_computed_not_supported(self, sd, esd):
        op_fixture("oracle")
        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(),
        )

    def test_alter_table_rename_oracle(self):
        context = op_fixture("oracle")
        op.rename_table("s", "t")
        context.assert_("ALTER TABLE s RENAME TO t")

    def test_alter_table_rename_schema_oracle(self):
        context = op_fixture("oracle")
        op.rename_table("s", "t", schema="myowner")
        context.assert_("ALTER TABLE myowner.s RENAME TO t")

    def test_alter_column_rename_oracle(self):
        context = op_fixture("oracle")
        op.alter_column("t", "c", new_column_name="x")
        context.assert_("ALTER TABLE t RENAME COLUMN c TO x")

    def test_alter_column_new_type(self):
        context = op_fixture("oracle")
        op.alter_column("t", "c", type_=Integer)
        context.assert_("ALTER TABLE t MODIFY c INTEGER")

    def test_alter_column_add_comment(self):
        context = op_fixture("oracle")
        op.alter_column("t", "c", type_=Integer, comment="c comment")
        context.assert_(
            "ALTER TABLE t MODIFY c INTEGER",
            "COMMENT ON COLUMN t.c IS 'c comment'",
        )

    def test_alter_column_add_comment_quotes(self):
        context = op_fixture("oracle")
        op.alter_column("t", "c", type_=Integer, comment="c 'comment'")
        context.assert_(
            "ALTER TABLE t MODIFY c INTEGER",
            "COMMENT ON COLUMN t.c IS 'c ''comment'''",
        )

    def test_alter_column_drop_comment(self):
        context = op_fixture("oracle")
        op.alter_column("t", "c", type_=Integer, comment=None)
        context.assert_(
            "ALTER TABLE t MODIFY c INTEGER", "COMMENT ON COLUMN t.c IS ''"
        )

    def test_create_table_comment(self):
        # this is handled by SQLAlchemy's compilers
        context = op_fixture("oracle")
        op.create_table_comment("t2", comment="t2 table", schema="foo")
        context.assert_("COMMENT ON TABLE foo.t2 IS 't2 table'")

    def test_drop_table_comment(self):
        # this is handled by SQLAlchemy's compilers
        context = op_fixture("oracle")
        op.drop_table_comment("t2", existing_comment="t2 table", schema="foo")
        context.assert_("COMMENT ON TABLE foo.t2 IS ''")

    def test_drop_index(self):
        context = op_fixture("oracle")
        op.drop_index("my_idx", table_name="my_table")
        context.assert_contains("DROP INDEX my_idx")

    def test_drop_column_w_default(self):
        context = op_fixture("oracle")
        op.drop_column("t1", "c1")
        context.assert_("ALTER TABLE t1 DROP COLUMN c1")

    def test_drop_column_w_check(self):
        context = op_fixture("oracle")
        op.drop_column("t1", "c1")
        context.assert_("ALTER TABLE t1 DROP COLUMN c1")

    def test_alter_column_nullable_w_existing_type(self):
        context = op_fixture("oracle")
        op.alter_column("t", "c", nullable=True, existing_type=Integer)
        context.assert_("ALTER TABLE t MODIFY c NULL")

    def test_alter_column_not_nullable_w_existing_type(self):
        context = op_fixture("oracle")
        op.alter_column("t", "c", nullable=False, existing_type=Integer)
        context.assert_("ALTER TABLE t MODIFY c NOT NULL")

    def test_alter_column_nullable_w_new_type(self):
        context = op_fixture("oracle")
        op.alter_column("t", "c", nullable=True, type_=Integer)
        context.assert_(
            "ALTER TABLE t MODIFY c NULL", "ALTER TABLE t MODIFY c INTEGER"
        )

    def test_alter_column_not_nullable_w_new_type(self):
        context = op_fixture("oracle")
        op.alter_column("t", "c", nullable=False, type_=Integer)
        context.assert_(
            "ALTER TABLE t MODIFY c NOT NULL", "ALTER TABLE t MODIFY c INTEGER"
        )

    def test_alter_add_server_default(self):
        context = op_fixture("oracle")
        op.alter_column("t", "c", server_default="5")
        context.assert_("ALTER TABLE t MODIFY c DEFAULT '5'")

    def test_alter_replace_server_default(self):
        context = op_fixture("oracle")
        op.alter_column(
            "t", "c", server_default="5", existing_server_default="6"
        )
        context.assert_("ALTER TABLE t MODIFY c DEFAULT '5'")

    def test_alter_remove_server_default(self):
        context = op_fixture("oracle")
        op.alter_column("t", "c", server_default=None)
        context.assert_("ALTER TABLE t MODIFY c DEFAULT NULL")

    def test_alter_do_everything(self):
        context = op_fixture("oracle")
        op.alter_column(
            "t",
            "c",
            new_column_name="c2",
            nullable=True,
            type_=Integer,
            server_default="5",
        )
        context.assert_(
            "ALTER TABLE t MODIFY c NULL",
            "ALTER TABLE t MODIFY c DEFAULT '5'",
            "ALTER TABLE t MODIFY c INTEGER",
            "ALTER TABLE t RENAME COLUMN c TO c2",
        )

    @config.requirements.comments
    def test_create_table_with_column_comments(self):
        context = op_fixture("oracle")
        op.create_table(
            "t2", Column("c1", Integer, primary_key=True), comment="t2 comment"
        )
        context.assert_(
            "CREATE TABLE t2 (c1 INTEGER NOT NULL, PRIMARY KEY (c1))",
            "COMMENT ON TABLE t2 IS 't2 comment'",
        )

    # TODO: when we add schema support
    # def test_alter_column_rename_oracle_schema(self):
    #    context = op_fixture('oracle')
    #    op.alter_column("t", "c", name="x", schema="y")
    #    context.assert_(
    #        'ALTER TABLE y.t RENAME COLUMN c TO c2'
    #    )

    def _identity_qualification(self, kw):
        always = kw.get("always", False)
        if always is None:
            return ""
        qualification = "ALWAYS" if always else "BY DEFAULT"
        if kw.get("on_null", False):
            qualification += " ON NULL"
        return qualification

    @config.requirements.identity_columns
    @combinations(
        ({}, None),
        (dict(always=True), None),
        (dict(always=None, order=True), "ORDER"),
        (
            dict(start=3, increment=33, maxvalue=99, cycle=True),
            "INCREMENT BY 33 START WITH 3 MAXVALUE 99 CYCLE",
        ),
        (dict(on_null=True, start=42), "START WITH 42"),
    )
    def test_add_column_identity(self, kw, text):
        context = op_fixture("oracle")
        op.add_column(
            "t1",
            Column("some_column", Integer, sqla_compat.Identity(**kw)),
        )
        qualification = self._identity_qualification(kw)
        options = " (%s)" % text if text else ""
        context.assert_(
            "ALTER TABLE t1 ADD some_column "
            "INTEGER GENERATED %s AS IDENTITY%s" % (qualification, options)
        )

    @config.requirements.identity_columns
    @combinations(
        ({}, None),
        (dict(always=True), None),
        (dict(always=None, cycle=True), "CYCLE"),
        (
            dict(start=3, increment=33, maxvalue=99, cycle=True),
            "INCREMENT BY 33 START WITH 3 MAXVALUE 99 CYCLE",
        ),
        (dict(on_null=True, start=42), "START WITH 42"),
    )
    def test_add_identity_to_column(self, kw, text):
        context = op_fixture("oracle")
        op.alter_column(
            "t1",
            "some_column",
            server_default=sqla_compat.Identity(**kw),
            existing_server_default=None,
        )
        qualification = self._identity_qualification(kw)
        options = " (%s)" % text if text else ""
        context.assert_(
            "ALTER TABLE t1 MODIFY some_column "
            "GENERATED %s AS IDENTITY%s" % (qualification, options)
        )

    @config.requirements.identity_columns
    def test_remove_identity_from_column(self):
        context = op_fixture("oracle")
        op.alter_column(
            "t1",
            "some_column",
            server_default=None,
            existing_server_default=sqla_compat.Identity(),
        )
        context.assert_("ALTER TABLE t1 MODIFY some_column DROP IDENTITY")

    @config.requirements.identity_columns
    @combinations(
        ({}, dict(always=True), None),
        (
            dict(always=True),
            dict(always=False, start=3),
            "START WITH 3",
        ),
        (
            dict(always=True, start=3, increment=2, minvalue=-3, maxvalue=99),
            dict(
                always=True,
                start=3,
                increment=1,
                minvalue=-3,
                maxvalue=99,
                cycle=True,
            ),
            "INCREMENT BY 1 START WITH 3 MINVALUE -3 MAXVALUE 99 CYCLE",
        ),
        (
            dict(
                always=False,
                start=3,
                maxvalue=9999,
                minvalue=0,
            ),
            dict(always=False, start=3, order=True, on_null=False, cache=2),
            "START WITH 3 CACHE 2 ORDER",
        ),
        (
            dict(always=False),
            dict(always=None, minvalue=0),
            "MINVALUE 0",
        ),
    )
    def test_change_identity_in_column(self, existing, updated, text):
        context = op_fixture("oracle")
        op.alter_column(
            "t1",
            "some_column",
            server_default=sqla_compat.Identity(**updated),
            existing_server_default=sqla_compat.Identity(**existing),
        )

        qualification = self._identity_qualification(updated)
        options = " (%s)" % text if text else ""
        context.assert_(
            "ALTER TABLE t1 MODIFY some_column "
            "GENERATED %s AS IDENTITY%s" % (qualification, options)
        )