summaryrefslogtreecommitdiff
path: root/tests
diff options
context:
space:
mode:
authorAdrien Berchet <adrien.berchet@gmail.com>2022-05-31 08:31:25 -0400
committerMike Bayer <mike_mp@zzzcomputing.com>2022-05-31 16:26:37 -0400
commitf73456034f51de5a1a7a9c4d6b23f459049776d0 (patch)
treea17b866e7ed193f5c287bd8adc51894b15ae324e /tests
parent188dd8b2f098e804c8e64956e9c6490d41f1d7ce (diff)
downloadalembic-f73456034f51de5a1a7a9c4d6b23f459049776d0.tar.gz
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
Diffstat (limited to 'tests')
-rw-r--r--tests/test_op.py53
1 files changed, 53 insertions, 0 deletions
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):