Move deactivation of notify trigger to tests (#3354)

* move deactivation of notify triggers to conftest
* deactivate triggers also before and after migration test

---------

Co-authored-by: Hannes Janott <hannes.janott@openslides.com>

Author: rrenkert

openslides_backend/services/postgresql/create_schema.py

@@ -1,7 +1,6 @@
 import os
-from typing import Any
 
-from psycopg import Connection, Cursor, rows, sql
+from psycopg import Connection, rows, sql
 
 from openslides_backend.migrations.exceptions import (
     MismatchingMigrationIndicesException,
@@ -60,7 +59,6 @@ def create_schema() -> None:
                     MigrationHelper.set_database_migration_info(
                         cursor, 100, MigrationState.FINALIZED
                     )
-                deactivate_notify_triggers(cursor)
                 return
             # We have a migration index if this is a legacy instance.
             # A migration index higher than or equal to MIN_NON_REL_MIGRATION is not
@@ -88,7 +86,6 @@ def create_schema() -> None:
                     type_ = "fresh"
                     db_migration_index = MigrationHelper.get_backend_migration_index()
                 print(f"Assuming {type_} database.")
-                deactivate_notify_triggers(cursor)
                 MigrationHelper.set_database_migration_info(
                     cursor,
                     db_migration_index,
@@ -100,28 +97,3 @@ def create_schema() -> None:
             except Exception as e:
                 print(f"On applying relational schema there was an error: {str(e)}\n")
                 return
-
-
-def deactivate_notify_triggers(cursor: Cursor[dict[str, Any]]) -> None:
-    if env.is_dev_mode():
-        # deactivate all notify triggers
-        for table in MigrationHelper.get_public_tables(cursor):
-            to_disable_triggers = cursor.execute(
-                sql.SQL(
-                    """SELECT
-                        tgname AS trigger_name,
-                        tgrelid::regclass AS table_name
-                    FROM
-                        pg_trigger
-                    WHERE
-                        tgrelid = {table_name}::regclass AND
-                        tgname LIKE 'tr_log_%' OR tgname LIKE 'notify_%';"""
-                ).format(table_name=table)
-            ).fetchall()
-            for trigger_dict in to_disable_triggers:
-                cursor.execute(
-                    sql.SQL("ALTER TABLE {table} DISABLE TRIGGER {trigger};").format(
-                        table=sql.Identifier(trigger_dict["table_name"]),
-                        trigger=sql.SQL(trigger_dict["trigger_name"]),
-                    )
-                )

tests/conftest.py

@@ -11,6 +11,7 @@
     get_new_os_conn,
 )
 from tests.conftest_helper import (
+    deactivate_notify_triggers,
     generate_remove_all_test_functions,
     generate_sql_for_test_initiation,
 )
@@ -52,6 +53,7 @@ def setup_pytest_session() -> Generator[None]:
     """
     with get_new_os_conn() as conn:
         with conn.cursor() as curs:
+            deactivate_notify_triggers(curs)
             tablenames = get_rel_db_table_names(curs)
             if tablenames:
                 curs.execute(

tests/conftest_helper.py

@@ -1,4 +1,7 @@
 from textwrap import dedent
+from typing import Any
+
+from psycopg import Cursor, sql
 
 from openslides_backend.migrations.migration_helper import (
     MigrationHelper,
@@ -9,6 +12,30 @@
 openslides_db = env.DATABASE_NAME
 
 
+def deactivate_notify_triggers(cursor: Cursor[dict[str, Any]]) -> None:
+    """Deactivates all notify triggers present in the database."""
+    for table in MigrationHelper.get_public_tables(cursor):
+        to_disable_triggers = cursor.execute(
+            sql.SQL(
+                """SELECT
+                    tgname AS trigger_name,
+                    tgrelid::regclass AS table_name
+                FROM
+                    pg_trigger
+                WHERE
+                    tgrelid = {table_name}::regclass AND
+                    tgname LIKE 'tr_log_%' OR tgname LIKE 'notify_%';"""
+            ).format(table_name=table)
+        ).fetchall()
+        for trigger_dict in to_disable_triggers:
+            cursor.execute(
+                sql.SQL("ALTER TABLE {table} DISABLE TRIGGER {trigger};").format(
+                    table=sql.Identifier(trigger_dict["table_name"]),
+                    trigger=sql.SQL(trigger_dict["trigger_name"]),
+                )
+            )
+
+
 def generate_trigger_sql_code(tablenames: tuple[str, ...]) -> str:
     sql: list[str] = []
     for table in tablenames:

tests/system/migrations/test_0100_init_reldb.py

@@ -35,7 +35,10 @@
 )
 from openslides_backend.shared.env import DEV_PASSWORD, Environment
 from tests.conftest import OLD_TABLES, get_rel_db_table_names
-from tests.conftest_helper import generate_sql_for_test_initiation
+from tests.conftest_helper import (
+    deactivate_notify_triggers,
+    generate_sql_for_test_initiation,
+)
 from tests.system.action.util import get_internal_auth_header
 from tests.system.util import create_action_test_application, get_route_path
 from tests.util import AuthData, Client, Response
@@ -107,6 +110,7 @@ def apply_test_relational_schema() -> None:
             with conn.cursor() as curs:
                 table_names = get_rel_db_table_names(curs)
                 curs.execute(generate_sql_for_test_initiation(tuple(table_names)))
+                deactivate_notify_triggers(curs)
 
     def tearDown(self) -> None:
         migration_module.Sql_helper.offset = 0