From f73456034f51de5a1a7a9c4d6b23f459049776d0 Mon Sep 17 00:00:00 2001 From: Adrien Berchet Date: Tue, 31 May 2022 08:31:25 -0400 Subject: Trigger events before and after drop table statements The ``op.drop_table()`` operation directive will now trigger the ``before_drop()`` and ``after_drop()`` DDL event hooks at the table level, which is similar to how the ``before_create()`` and ``after_create()`` hooks are triggered by the ``op.create_table()`` directive. Note that as ``op.drop_table()`` accepts only a table name and optional schema name, the ``Table`` object received by the event will not have any information within it other than the table name and schema name. Fixes: #1037 Closes: #1036 Pull-request: https://github.com/sqlalchemy/alembic/pull/1036 Pull-request-sha: ea44e7f3398aa5001e86e6a7e30f61f6a86d9674 Change-Id: I20a1702e17ed88054206d964152ce05b81d0f89e --- tests/test_op.py | 53 +++++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 53 insertions(+) (limited to 'tests') diff --git a/tests/test_op.py b/tests/test_op.py index 677b7bc..5682abb 100644 --- a/tests/test_op.py +++ b/tests/test_op.py @@ -3,6 +3,7 @@ from sqlalchemy import Boolean from sqlalchemy import CheckConstraint from sqlalchemy import Column +from sqlalchemy import event from sqlalchemy import exc from sqlalchemy import ForeignKey from sqlalchemy import Index @@ -1087,6 +1088,58 @@ class OpTest(TestBase): context.assert_("COMMENT ON TABLE some_table IS NULL") + def test_create_table_event(self): + context = op_fixture() + + events_triggered = [] + + TestTable = Table( + "tb_test", MetaData(), Column("c1", Integer, nullable=False) + ) + + @event.listens_for(Table, "before_create") + def record_before_event(table, conn, **kwargs): + events_triggered.append(("before_create", table.name)) + + @event.listens_for(Table, "after_create") + def record_after_event(table, conn, **kwargs): + events_triggered.append(("after_create", table.name)) + + op.create_table(TestTable) + op.drop_table(TestTable) + context.assert_("CREATE TABLE tb_test ()", "DROP TABLE tb_test") + + assert events_triggered == [ + ("before_create", "tb_test"), + ("after_create", "tb_test"), + ] + + def test_drop_table_event(self): + context = op_fixture() + + events_triggered = [] + + TestTable = Table( + "tb_test", MetaData(), Column("c1", Integer, nullable=False) + ) + + @event.listens_for(Table, "before_drop") + def record_before_event(table, conn, **kwargs): + events_triggered.append(("before_drop", table.name)) + + @event.listens_for(Table, "after_drop") + def record_after_event(table, conn, **kwargs): + events_triggered.append(("after_drop", table.name)) + + op.create_table(TestTable) + op.drop_table(TestTable) + context.assert_("CREATE TABLE tb_test ()", "DROP TABLE tb_test") + + assert events_triggered == [ + ("before_drop", "tb_test"), + ("after_drop", "tb_test"), + ] + class SQLModeOpTest(TestBase): def test_auto_literals(self): -- cgit v1.2.1