feat: Update database permissions in async mode (#32231)

2026-04-18 23:55:00 +00:00 · 2025-02-28 21:25:47 -03:00
parent 84b52b2323
commit d79f7b28c2
22 changed files with 1715 additions and 425 deletions
--- a/tests/unit_tests/commands/databases/conftest.py
+++ b/tests/unit_tests/commands/databases/conftest.py
@@ -0,0 +1,90 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+from unittest.mock import MagicMock
+
+import pytest
+from pytest_mock import MockerFixture
+
+from superset.db_engine_specs.base import BaseEngineSpec
+from superset.exceptions import OAuth2RedirectError
+from superset.utils import json
+
+oauth2_client_info = {
+    "id": "client_id",
+    "secret": "client_secret",
+    "scope": "scope-a",
+    "redirect_uri": "redirect_uri",
+    "authorization_request_uri": "auth_uri",
+    "token_request_uri": "token_uri",
+    "request_content_type": "json",
+}
+
+
+@pytest.fixture
+def database_with_catalog(mocker: MockerFixture) -> MagicMock:
+    """
+    Mock a database with catalogs and schemas.
+    """
+    database = mocker.MagicMock()
+    database.database_name = "my_db"
+    database.db_engine_spec.__name__ = "test_engine"
+    database.db_engine_spec.supports_catalog = True
+    database.get_all_catalog_names.return_value = ["catalog1", "catalog2"]
+    database.get_all_schema_names.side_effect = [
+        ["schema1", "schema2"],
+        ["schema3", "schema4"],
+    ]
+    database.get_default_catalog.return_value = "catalog2"
+
+    return database
+
+
+@pytest.fixture
+def database_without_catalog(mocker: MockerFixture) -> MagicMock:
+    """
+    Mock a database without catalogs.
+    """
+    database = mocker.MagicMock()
+    database.database_name = "my_db"
+    database.db_engine_spec.__name__ = "test_engine"
+    database.db_engine_spec.supports_catalog = False
+    database.get_all_schema_names.return_value = ["schema1", "schema2"]
+
+    return database
+
+
+@pytest.fixture
+def database_needs_oauth2(mocker: MockerFixture) -> MagicMock:
+    """
+    Mock a database without catalogs that needs OAuth2.
+    """
+    database = mocker.MagicMock()
+    database.database_name = "my_db"
+    database.db_engine_spec.__name__ = "test_engine"
+    database.db_engine_spec.supports_catalog = False
+    database.get_all_schema_names.side_effect = OAuth2RedirectError(
+        "url",
+        "tab_id",
+        "redirect_uri",
+    )
+    database.encrypted_extra = json.dumps({"oauth2_client_info": oauth2_client_info})
+    database.db_engine_spec.unmask_encrypted_extra = (
+        BaseEngineSpec.unmask_encrypted_extra
+    )
+
+    return database
--- a/tests/unit_tests/commands/databases/sync_permissions_test.py
+++ b/tests/unit_tests/commands/databases/sync_permissions_test.py
@@ -0,0 +1,389 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations
+
+from unittest.mock import MagicMock
+
+import pytest
+from pytest_mock import MockerFixture
+
+from superset import db
+from superset.commands.database.exceptions import (
+    DatabaseConnectionFailedError,
+    DatabaseNotFoundError,
+    UserNotFoundInSessionError,
+)
+from superset.commands.database.sync_permissions import SyncPermissionsCommand
+from superset.db_engine_specs.base import GenericDBException
+from superset.exceptions import OAuth2RedirectError
+from superset.extensions import security_manager
+from tests.conftest import with_config
+
+
+@with_config({"SYNC_DB_PERMISSIONS_IN_ASYNC_MODE": False})
+def test_sync_permissions_command_sync_mode(
+    mocker: MockerFixture,
+    database_with_catalog: MagicMock,
+):
+    """
+    Test ``SyncPermissionsCommand`` in sync mode.
+    """
+    mock_ssh = mocker.MagicMock()
+    user_mock = mocker.patch(
+        "superset.commands.database.sync_permissions.security_manager.get_user_by_username"
+    )
+    mocker.patch("superset.commands.database.sync_permissions.ping", return_value=True)
+    find_pvm_mock = mocker.patch(
+        "superset.commands.database.sync_permissions.security_manager.find_permission_view_menu"
+    )
+    find_pvm_mock.side_effect = [mocker.MagicMock(), None]
+    add_pvm_mock = mocker.patch("superset.commands.database.sync_permissions.add_pvm")
+
+    cmmd = SyncPermissionsCommand(
+        1, "admin", db_connection=database_with_catalog, ssh_tunnel=mock_ssh
+    )
+    mock_refresh_schemas = mocker.patch.object(cmmd, "_refresh_schemas")
+    mock_rename_db_perm = mocker.patch.object(cmmd, "_rename_database_in_permissions")
+
+    cmmd.run()
+
+    assert cmmd.db_connection == database_with_catalog
+    assert cmmd.old_db_connection_name == "my_db"
+    assert cmmd.db_connection_ssh_tunnel == mock_ssh
+    user_mock.assert_called_once_with("admin")
+    add_pvm_mock.assert_has_calls(
+        [
+            mocker.call(
+                db.session, security_manager, "catalog_access", "[my_db].[catalog2]"
+            ),
+            mocker.call(
+                db.session,
+                security_manager,
+                "schema_access",
+                "[my_db].[catalog2].[schema3]",
+            ),
+            mocker.call(
+                db.session,
+                security_manager,
+                "schema_access",
+                "[my_db].[catalog2].[schema4]",
+            ),
+        ]
+    )
+    mock_refresh_schemas.assert_called_once_with("catalog1", ["schema1", "schema2"])
+    mock_rename_db_perm.assert_not_called()
+
+
+@with_config({"SYNC_DB_PERMISSIONS_IN_ASYNC_MODE": True})
+def test_sync_permissions_command_async_mode(
+    mocker: MockerFixture, database_with_catalog: MagicMock
+) -> None:
+    """
+    Test ``SyncPermissionsCommand`` in async mode.
+    """
+    mock_database_dao = mocker.patch(
+        "superset.commands.database.sync_permissions.DatabaseDAO"
+    )
+    mock_database_dao.find_by_id.return_value = database_with_catalog
+    mocker.patch(
+        "superset.commands.database.sync_permissions.security_manager.get_user_by_username"
+    )
+    async_task_mock = mocker.patch(
+        "superset.commands.database.sync_permissions.sync_database_permissions_task"
+    )
+    mocker.patch("superset.commands.database.sync_permissions.ping", return_value=True)
+
+    cmmd = SyncPermissionsCommand(1, "admin")
+    cmmd.run()
+    async_task_mock.delay.assert_called_once_with(1, "admin", "my_db")
+
+
+@with_config({"SYNC_DB_PERMISSIONS_IN_ASYNC_MODE": False})
+def test_sync_permissions_command_passing_all_values(
+    mocker: MockerFixture, database_with_catalog: MagicMock
+):
+    """
+    Test ``SyncPermissionsCommand`` when providing all arguments to the constructor.
+    """
+    mock_ssh = mocker.MagicMock()
+    mock_database_dao = mocker.patch(
+        "superset.commands.database.sync_permissions.DatabaseDAO"
+    )
+    mocker.patch(
+        "superset.commands.database.sync_permissions.security_manager.get_user_by_username"
+    )
+    mocker.patch("superset.commands.database.sync_permissions.ping", return_value=True)
+
+    cmmd = SyncPermissionsCommand(
+        1,
+        "admin",
+        old_db_connection_name="old name",
+        db_connection=database_with_catalog,
+        ssh_tunnel=mock_ssh,
+    )
+    mocker.patch.object(cmmd, "sync_database_permissions")
+    cmmd.run()
+
+    assert cmmd.db_connection == database_with_catalog
+    assert cmmd.old_db_connection_name == "old name"
+    assert cmmd.db_connection_ssh_tunnel == mock_ssh
+    mock_database_dao.find_by_id.assert_not_called()
+    mock_database_dao.get_ssh_tunnel.assert_not_called()
+
+
+@with_config({"SYNC_DB_PERMISSIONS_IN_ASYNC_MODE": False})
+def test_sync_permissions_command_raise(mocker: MockerFixture):
+    """
+    Test ``SyncPermissionsCommand`` when an exception is raised.
+    """
+    mock_database_dao = mocker.patch(
+        "superset.commands.database.sync_permissions.DatabaseDAO"
+    )
+    mock_database_dao.find_by_id.return_value = mocker.MagicMock()
+    mock_database_dao.get_ssh_tunnel.return_value = mocker.MagicMock()
+    mock_user = mocker.patch(
+        "superset.commands.database.sync_permissions.security_manager.get_user_by_username"
+    )
+
+    # Connection issues
+    mock_ping = mocker.patch(
+        "superset.commands.database.sync_permissions.ping", return_value=False
+    )
+    with pytest.raises(DatabaseConnectionFailedError):
+        SyncPermissionsCommand(1, "admin").run()
+    mock_ping.reset_mock()
+    mock_ping.side_effect = Exception
+    with pytest.raises(DatabaseConnectionFailedError):
+        SyncPermissionsCommand(1, "admin").run()
+
+    # User not found in session
+    mock_user.reset_mock()
+    mock_user.return_value = None
+    with pytest.raises(UserNotFoundInSessionError):
+        SyncPermissionsCommand(1, "admin").run()
+    mock_user.reset_mock()
+    mock_user.return_value = mocker.MagicMock()
+
+    # DB connection not found
+    mock_database_dao.reset_mock()
+    mock_database_dao.find_by_id.return_value = None
+    with pytest.raises(DatabaseNotFoundError):
+        SyncPermissionsCommand(1, "admin").run()
+
+
+@with_config({"SYNC_DB_PERMISSIONS_IN_ASYNC_MODE": False})
+def test_sync_permissions_command_new_db_name(
+    mocker: MockerFixture, database_with_catalog: MagicMock
+):
+    """
+    Test ``SyncPermissionsCommand`` when the database name changed.
+    """
+    mocker.patch(
+        "superset.commands.database.sync_permissions.security_manager.get_user_by_username"
+    )
+    cmmd = SyncPermissionsCommand(
+        1,
+        "admin",
+        old_db_connection_name="Old Name",
+        db_connection=database_with_catalog,
+    )
+    cmmd.run()
+
+    assert cmmd.old_db_connection_name == "Old Name"
+
+
+@with_config({"SYNC_DB_PERMISSIONS_IN_ASYNC_MODE": True})
+def test_sync_permissions_command_async_mode_new_db_name(
+    mocker: MockerFixture, database_with_catalog: MagicMock
+):
+    """
+    Test ``SyncPermissionsCommand`` in async mode when the
+    database name changed.
+    """
+    mocker.patch(
+        "superset.commands.database.sync_permissions.security_manager.get_user_by_username"
+    )
+    async_task_mock = mocker.patch(
+        "superset.commands.database.sync_permissions.sync_database_permissions_task"
+    )
+    cmmd = SyncPermissionsCommand(
+        1,
+        "admin",
+        old_db_connection_name="Old Name",
+        db_connection=database_with_catalog,
+    )
+    cmmd.run()
+
+    async_task_mock.delay.assert_called_once_with(1, "admin", "Old Name")
+
+
+def test_resync_permissions_command_get_catalogs(database_with_catalog: MagicMock):
+    """
+    Test the ``_get_catalog_names`` method.
+    """
+    cmmd = SyncPermissionsCommand(1, None, db_connection=database_with_catalog)
+    assert cmmd._get_catalog_names() == ["catalog1", "catalog2"]
+
+
+@pytest.mark.parametrize(
+    ("inner_exception, outer_exception"),
+    [
+        (
+            OAuth2RedirectError("Missing token", "mock_tab", "mock_url"),
+            OAuth2RedirectError,
+        ),
+        (GenericDBException, DatabaseConnectionFailedError),
+    ],
+)
+def test_resync_permissions_command_raise_on_getting_catalogs(
+    inner_exception: Exception,
+    outer_exception: Exception,
+    database_with_catalog: MagicMock,
+):
+    """
+    Test the ``_get_catalog_names`` method when raising an exception.
+    """
+    database_with_catalog.get_all_catalog_names.side_effect = inner_exception
+    cmmd = SyncPermissionsCommand(1, None, db_connection=database_with_catalog)
+    with pytest.raises(outer_exception):
+        cmmd._get_catalog_names()
+
+
+def test_resync_permissions_command_get_schemas(database_with_catalog: MagicMock):
+    """
+    Test the ``_get_schema_names`` method.
+    """
+    cmmd = SyncPermissionsCommand(1, None, db_connection=database_with_catalog)
+    assert cmmd._get_schema_names("catalog1") == ["schema1", "schema2"]
+    assert cmmd._get_schema_names("catalog2") == ["schema3", "schema4"]
+
+
+@pytest.mark.parametrize(
+    ("inner_exception, outer_exception"),
+    [
+        (
+            OAuth2RedirectError("Missing token", "mock_tab", "mock_url"),
+            OAuth2RedirectError,
+        ),
+        (GenericDBException, DatabaseConnectionFailedError),
+    ],
+)
+def test_resync_permissions_command_raise_on_getting_schemas(
+    inner_exception: Exception,
+    outer_exception: Exception,
+    database_with_catalog: MagicMock,
+):
+    """
+    Test the ``_get_schema_names`` method when raising an exception.
+    """
+    database_with_catalog.get_all_schema_names.side_effect = inner_exception
+    cmmd = SyncPermissionsCommand(1, None, db_connection=database_with_catalog)
+    with pytest.raises(outer_exception):
+        cmmd._get_schema_names("blah")
+
+
+def test_resync_permissions_command_refresh_schemas(
+    mocker: MockerFixture, database_with_catalog: MagicMock
+):
+    """
+    Test the ``_refresh_schemas`` method.
+    """
+    find_pvm_mock = mocker.patch(
+        "superset.commands.database.sync_permissions.security_manager.find_permission_view_menu"
+    )
+    find_pvm_mock.side_effect = [mocker.MagicMock(), None]
+    add_pvm_mock = mocker.patch("superset.commands.database.sync_permissions.add_pvm")
+
+    cmmd = SyncPermissionsCommand(1, None, db_connection=database_with_catalog)
+    cmmd._refresh_schemas("catalog1", ["schema1", "schema2"])
+
+    add_pvm_mock.assert_called_once_with(
+        db.session,
+        security_manager,
+        "schema_access",
+        f"[{database_with_catalog.name}].[catalog1].[schema2]",
+    )
+
+
+def test_resync_permissions_command_rename_db_in_perms(
+    mocker: MockerFixture, database_with_catalog: MagicMock
+):
+    """
+    Test the ``_rename_database_in_permissions`` method.
+    """
+    find_pvm_mock = mocker.patch(
+        "superset.commands.database.sync_permissions.security_manager.find_permission_view_menu"
+    )
+    mock_catalog_perm = mocker.MagicMock()
+    mock_catalog_perm.view_menu.name = "[old_name].[catalog]"
+    mock_schema_perm = mocker.MagicMock()
+    mock_schema_perm.view_menu.name = "[old_name].[catalog].[schema1]"
+    find_pvm_mock.side_effect = [
+        mock_catalog_perm,
+        mock_schema_perm,
+        None,
+    ]
+
+    mock_dataset = mocker.MagicMock()
+    mock_dataset.id = 1
+    mock_dataset.catalog_perm = "[old_name].[catalog1]"
+    mock_dataset.schema_perm = "[old_name].[catalog1].[schema1]"
+    mock_chart = mocker.MagicMock()
+    mock_chart.catalog_perm = "[old_name].[catalog1]"
+    mock_chart.schema_perm = "[old_name].[catalog1].[schema1]"
+
+    mock_database_dao = mocker.patch(
+        "superset.commands.database.sync_permissions.DatabaseDAO"
+    )
+    mock_database_dao.get_datasets.side_effect = [
+        [mock_dataset],
+        [],
+    ]
+    mock_dataset_dao = mocker.patch(
+        "superset.commands.database.sync_permissions.DatasetDAO"
+    )
+    mock_dataset_dao.get_related_objects.return_value = {"charts": [mock_chart]}
+
+    cmmd = SyncPermissionsCommand(
+        1, None, old_db_connection_name="old_name", db_connection=database_with_catalog
+    )
+    cmmd._rename_database_in_permissions("catalog1", ["schema1", "schema2"])
+
+    find_pvm_mock.assert_has_calls(
+        [
+            mocker.call("catalog_access", "[old_name].[catalog1]"),
+            mocker.call("schema_access", "[old_name].[catalog1].[schema1]"),
+            mocker.call("schema_access", "[old_name].[catalog1].[schema2]"),
+        ]
+    )
+
+    assert (
+        mock_catalog_perm.view_menu.name == f"[{database_with_catalog.name}].[catalog1]"
+    )
+    assert (
+        mock_schema_perm.view_menu.name
+        == f"[{database_with_catalog.name}].[catalog1].[schema1]"
+    )
+    assert mock_dataset.catalog_perm == f"[{database_with_catalog.name}].[catalog1]"
+    assert (
+        mock_dataset.schema_perm
+        == f"[{database_with_catalog.name}].[catalog1].[schema1]"
+    )
+    assert mock_chart.catalog_perm == f"[{database_with_catalog.name}].[catalog1]"
+    assert (
+        mock_chart.schema_perm == f"[{database_with_catalog.name}].[catalog1].[schema1]"
+    )
--- a/tests/unit_tests/commands/databases/update_test.py
+++ b/tests/unit_tests/commands/databases/update_test.py
@@ -17,84 +17,19 @@

 from unittest.mock import MagicMock

-import pytest
 from pytest_mock import MockerFixture

+from superset import db
 from superset.commands.database.update import UpdateDatabaseCommand
-from superset.db_engine_specs.base import BaseEngineSpec
-from superset.exceptions import OAuth2RedirectError
 from superset.extensions import security_manager
 from superset.utils import json
-
-oauth2_client_info = {
-    "id": "client_id",
-    "secret": "client_secret",
-    "scope": "scope-a",
-    "redirect_uri": "redirect_uri",
-    "authorization_request_uri": "auth_uri",
-    "token_request_uri": "token_uri",
-    "request_content_type": "json",
-}
-
-
-@pytest.fixture
-def database_with_catalog(mocker: MockerFixture) -> MagicMock:
-    """
-    Mock a database with catalogs and schemas.
-    """
-    database = mocker.MagicMock()
-    database.database_name = "my_db"
-    database.db_engine_spec.__name__ = "test_engine"
-    database.db_engine_spec.supports_catalog = True
-    database.get_all_catalog_names.return_value = ["catalog1", "catalog2"]
-    database.get_all_schema_names.side_effect = [
-        ["schema1", "schema2"],
-        ["schema3", "schema4"],
-    ]
-    database.get_default_catalog.return_value = "catalog2"
-
-    return database
-
-
-@pytest.fixture
-def database_without_catalog(mocker: MockerFixture) -> MagicMock:
-    """
-    Mock a database without catalogs.
-    """
-    database = mocker.MagicMock()
-    database.database_name = "my_db"
-    database.db_engine_spec.__name__ = "test_engine"
-    database.db_engine_spec.supports_catalog = False
-    database.get_all_schema_names.return_value = ["schema1", "schema2"]
-
-    return database
-
-
-@pytest.fixture
-def database_needs_oauth2(mocker: MockerFixture) -> MagicMock:
-    """
-    Mock a database without catalogs that needs OAuth2.
-    """
-    database = mocker.MagicMock()
-    database.database_name = "my_db"
-    database.db_engine_spec.__name__ = "test_engine"
-    database.db_engine_spec.supports_catalog = False
-    database.get_all_schema_names.side_effect = OAuth2RedirectError(
-        "url",
-        "tab_id",
-        "redirect_uri",
-    )
-    database.encrypted_extra = json.dumps({"oauth2_client_info": oauth2_client_info})
-    database.db_engine_spec.unmask_encrypted_extra = (
-        BaseEngineSpec.unmask_encrypted_extra
-    )
-
-    return database
+from tests.conftest import with_config
+from tests.unit_tests.commands.databases.conftest import oauth2_client_info


 def test_update_with_catalog(
    mocker: MockerFixture,
-    database_with_catalog: MockerFixture,
+    database_with_catalog: MagicMock,
 ) -> None:
    """
    Test that permissions are updated correctly.
@@ -111,9 +46,15 @@ def test_update_with_catalog(
    When update is called, only `catalog2.schema3` has permissions associated with it,
    so `catalog1.*` and `catalog2.schema4` are added.
    """
-    DatabaseDAO = mocker.patch("superset.commands.database.update.DatabaseDAO")  # noqa: N806
-    DatabaseDAO.find_by_id.return_value = database_with_catalog
-    DatabaseDAO.update.return_value = database_with_catalog
+    database_dao = mocker.patch("superset.commands.database.update.DatabaseDAO")
+    database_dao.find_by_id.return_value = database_with_catalog
+    database_dao.update.return_value = database_with_catalog
+    sync_db_perms_dao = mocker.patch(
+        "superset.commands.database.sync_permissions.DatabaseDAO"
+    )
+    sync_db_perms_dao.find_by_id.return_value = database_with_catalog
+    mocker.patch("superset.commands.database.update.get_username")
+    mocker.patch("superset.security_manager.get_user_by_username")

    find_permission_view_menu = mocker.patch.object(
        security_manager,
@@ -128,25 +69,66 @@ def test_update_with_catalog(
        None,
        None,
    ]
-    add_permission_view_menu = mocker.patch.object(
-        security_manager,
-        "add_permission_view_menu",
-    )
+    add_pvm = mocker.patch("superset.commands.database.sync_permissions.add_pvm")

    UpdateDatabaseCommand(1, {}).run()

-    add_permission_view_menu.assert_has_calls(
+    add_pvm.assert_has_calls(
        [
            # first catalog is added with all schemas
-            mocker.call("catalog_access", "[my_db].[catalog1]"),
-            mocker.call("schema_access", "[my_db].[catalog1].[schema1]"),
-            mocker.call("schema_access", "[my_db].[catalog1].[schema2]"),
+            mocker.call(
+                db.session, security_manager, "catalog_access", "[my_db].[catalog1]"
+            ),
+            mocker.call(
+                db.session,
+                security_manager,
+                "schema_access",
+                "[my_db].[catalog1].[schema1]",
+            ),
+            mocker.call(
+                db.session,
+                security_manager,
+                "schema_access",
+                "[my_db].[catalog1].[schema2]",
+            ),
            # second catalog already exists, only `schema4` is added
-            mocker.call("schema_access", "[my_db].[catalog2].[schema4]"),
+            mocker.call(
+                db.session,
+                security_manager,
+                "schema_access",
+                f"[{database_with_catalog.name}].[catalog2].[schema4]",
+            ),
        ],
    )


+@with_config({"SYNC_DB_PERMISSIONS_IN_ASYNC_MODE": True})
+def test_update_sync_perms_in_async_mode(
+    mocker: MockerFixture,
+    database_with_catalog: MagicMock,
+) -> None:
+    """
+    Test that updating a DB connection with async mode enables
+    triggers the celery task to syn perms.
+    """
+    database_dao = mocker.patch("superset.commands.database.update.DatabaseDAO")
+    database_dao.find_by_id.return_value = database_with_catalog
+    database_dao.update.return_value = database_with_catalog
+    sync_db_perms_dao = mocker.patch(
+        "superset.commands.database.sync_permissions.DatabaseDAO"
+    )
+    sync_db_perms_dao.find_by_id.return_value = database_with_catalog
+    sync_task = mocker.patch(
+        "superset.commands.database.sync_permissions.sync_database_permissions_task.delay"
+    )
+    mocker.patch("superset.commands.database.update.get_username", return_value="admin")
+    mocker.patch("superset.security_manager.get_user_by_username")
+
+    UpdateDatabaseCommand(1, {}).run()
+
+    sync_task.assert_called_once_with(1, "admin", "my_db")
+
+
 def test_update_without_catalog(
    mocker: MockerFixture,
    database_without_catalog: MockerFixture,
@@ -162,9 +144,15 @@ def test_update_without_catalog(
    When update is called, only `schema2` has permissions associated with it, so `schema1`
    is added.
    """  # noqa: E501
-    DatabaseDAO = mocker.patch("superset.commands.database.update.DatabaseDAO")  # noqa: N806
-    DatabaseDAO.find_by_id.return_value = database_without_catalog
-    DatabaseDAO.update.return_value = database_without_catalog
+    database_dao = mocker.patch("superset.commands.database.update.DatabaseDAO")
+    database_dao.find_by_id.return_value = database_without_catalog
+    database_dao.update.return_value = database_without_catalog
+    sync_db_perms_dao = mocker.patch(
+        "superset.commands.database.sync_permissions.DatabaseDAO"
+    )
+    sync_db_perms_dao.find_by_id.return_value = database_without_catalog
+    mocker.patch("superset.commands.database.update.get_username")
+    mocker.patch("superset.security_manager.get_user_by_username")

    find_permission_view_menu = mocker.patch.object(
        security_manager,
@@ -174,22 +162,21 @@ def test_update_without_catalog(
        None,  # schema1 has no permissions
        "[my_db].[schema2]",  # second schema already exists
    ]
-    add_permission_view_menu = mocker.patch.object(
-        security_manager,
-        "add_permission_view_menu",
-    )
+    add_pvm = mocker.patch("superset.commands.database.sync_permissions.add_pvm")

    UpdateDatabaseCommand(1, {}).run()

-    add_permission_view_menu.assert_called_with(
+    add_pvm.assert_called_with(
+        db.session,
+        security_manager,
        "schema_access",
-        "[my_db].[schema1]",
+        f"[{database_without_catalog.name}].[schema1]",
    )


 def test_rename_with_catalog(
    mocker: MockerFixture,
-    database_with_catalog: MockerFixture,
+    database_with_catalog: MagicMock,
 ) -> None:
    """
    Test that permissions are renamed correctly.
@@ -207,25 +194,33 @@ def test_rename_with_catalog(
    so `catalog1.*` and `catalog2.schema4` are added. Additionally, the database has
    been renamed from `my_db` to `my_other_db`.
    """
-    DatabaseDAO = mocker.patch("superset.commands.database.update.DatabaseDAO")  # noqa: N806
    original_database = mocker.MagicMock()
    original_database.database_name = "my_db"
-    DatabaseDAO.find_by_id.return_value = original_database
+    database_dao = mocker.patch("superset.commands.database.update.DatabaseDAO")
+    database_dao.find_by_id.return_value = original_database
    database_with_catalog.database_name = "my_other_db"
-    DatabaseDAO.update.return_value = database_with_catalog
+    database_dao.update.return_value = database_with_catalog
+    sync_db_perms_dao = mocker.patch(
+        "superset.commands.database.sync_permissions.DatabaseDAO"
+    )
+    sync_db_perms_dao.find_by_id.return_value = database_with_catalog
+    mocker.patch("superset.commands.database.update.get_username")
+    mocker.patch("superset.security_manager.get_user_by_username")

    dataset = mocker.MagicMock()
    chart = mocker.MagicMock()
-    DatabaseDAO.get_datasets.return_value = [dataset]
-    DatasetDAO = mocker.patch("superset.commands.database.update.DatasetDAO")  # noqa: N806
-    DatasetDAO.get_related_objects.return_value = {"charts": [chart]}
+    sync_db_perms_dao.get_datasets.return_value = [dataset]
+    dataset_dao = mocker.patch("superset.commands.database.sync_permissions.DatasetDAO")
+    dataset_dao.get_related_objects.return_value = {"charts": [chart]}

    find_permission_view_menu = mocker.patch.object(
        security_manager,
        "find_permission_view_menu",
    )
    catalog2_pvm = mocker.MagicMock()
+    catalog2_pvm.view_menu.name = "[my_db].[catalog2]"
    catalog2_schema3_pvm = mocker.MagicMock()
+    catalog2_schema3_pvm.view_menu.name = "[my_db].[catalog2].[schema3]"
    find_permission_view_menu.side_effect = [
        # these are called when adding the permissions:
        None,  # first catalog is new
@@ -237,31 +232,52 @@ def test_rename_with_catalog(
        catalog2_schema3_pvm,  # old [my_db].[catalog2].[schema3]
        None,  # [my_db].[catalog2].[schema4] doesn't exist
    ]
-    add_permission_view_menu = mocker.patch.object(
-        security_manager,
-        "add_permission_view_menu",
-    )
+    add_pvm = mocker.patch("superset.commands.database.sync_permissions.add_pvm")
+    add_vm = mocker.patch("superset.commands.database.sync_permissions.add_vm")

    UpdateDatabaseCommand(1, {}).run()

-    add_permission_view_menu.assert_has_calls(
+    add_pvm.assert_has_calls(
        [
            # first catalog is added with all schemas with the new DB name
-            mocker.call("catalog_access", "[my_other_db].[catalog1]"),
-            mocker.call("schema_access", "[my_other_db].[catalog1].[schema1]"),
-            mocker.call("schema_access", "[my_other_db].[catalog1].[schema2]"),
+            mocker.call(
+                db.session,
+                security_manager,
+                "catalog_access",
+                "[my_other_db].[catalog1]",
+            ),
+            mocker.call(
+                db.session,
+                security_manager,
+                "schema_access",
+                "[my_other_db].[catalog1].[schema1]",
+            ),
+            mocker.call(
+                db.session,
+                security_manager,
+                "schema_access",
+                "[my_other_db].[catalog1].[schema2]",
+            ),
            # second catalog already exists, only `schema4` is added
-            mocker.call("schema_access", "[my_other_db].[catalog2].[schema4]"),
+            mocker.call(
+                db.session,
+                security_manager,
+                "schema_access",
+                f"[{database_with_catalog.name}].[catalog2].[schema4]",
+            ),
        ],
    )

-    assert catalog2_pvm.view_menu.name == "[my_other_db].[catalog2]"
-    assert catalog2_schema3_pvm.view_menu.name == "[my_other_db].[catalog2].[schema3]"
+    assert catalog2_pvm.view_menu == add_vm.return_value
+    assert (
+        catalog2_schema3_pvm.view_menu.name
+        == f"[{database_with_catalog.name}].[catalog2].[schema3]"
+    )

-    assert dataset.catalog_perm == "[my_other_db].[catalog2]"
-    assert dataset.schema_perm == "[my_other_db].[catalog2].[schema4]"
-    assert chart.catalog_perm == "[my_other_db].[catalog2]"
-    assert chart.schema_perm == "[my_other_db].[catalog2].[schema4]"
+    assert dataset.catalog_perm == f"[{database_with_catalog.name}].[catalog2]"
+    assert dataset.schema_perm == f"[{database_with_catalog.name}].[catalog2].[schema4]"
+    assert chart.catalog_perm == f"[{database_with_catalog.name}].[catalog2]"
+    assert chart.schema_perm == f"[{database_with_catalog.name}].[catalog2].[schema4]"


 def test_rename_without_catalog(
@@ -279,38 +295,44 @@ def test_rename_without_catalog(
    When update is called, only `schema2` has permissions associated with it, so `schema1`
    is added. Additionally, the database has been renamed from `my_db` to `my_other_db`.
    """  # noqa: E501
-    DatabaseDAO = mocker.patch("superset.commands.database.update.DatabaseDAO")  # noqa: N806
+    database_dao = mocker.patch("superset.commands.database.update.DatabaseDAO")
    original_database = mocker.MagicMock()
    original_database.database_name = "my_db"
-    DatabaseDAO.find_by_id.return_value = original_database
    database_without_catalog.database_name = "my_other_db"
-    DatabaseDAO.update.return_value = database_without_catalog
-    DatabaseDAO.get_datasets.return_value = []
+    database_dao.update.return_value = database_without_catalog
+    database_dao.find_by_id.return_value = original_database
+    sync_db_perms_dao = mocker.patch(
+        "superset.commands.database.sync_permissions.DatabaseDAO"
+    )
+    sync_db_perms_dao.find_by_id.return_value = database_without_catalog
+    sync_db_perms_dao.get_datasets.return_value = []
+    mocker.patch("superset.commands.database.update.get_username")
+    mocker.patch("superset.security_manager.get_user_by_username")

    find_permission_view_menu = mocker.patch.object(
        security_manager,
        "find_permission_view_menu",
    )
    schema2_pvm = mocker.MagicMock()
+    schema2_pvm.view_menu.name = "[my_db].[schema2]"
    find_permission_view_menu.side_effect = [
        None,  # schema1 has no permissions
        "[my_db].[schema2]",  # second schema already exists
        None,  # [my_db].[schema1] doesn't exist
        schema2_pvm,  # old [my_db].[schema2]
    ]
-    add_permission_view_menu = mocker.patch.object(
-        security_manager,
-        "add_permission_view_menu",
-    )
+    add_pvm = mocker.patch("superset.commands.database.sync_permissions.add_pvm")

    UpdateDatabaseCommand(1, {}).run()

-    add_permission_view_menu.assert_called_with(
+    add_pvm.assert_called_with(
+        db.session,
+        security_manager,
        "schema_access",
-        "[my_other_db].[schema1]",
+        f"[{database_without_catalog.name}].[schema1]",
    )

-    assert schema2_pvm.view_menu.name == "[my_other_db].[schema2]"
+    assert schema2_pvm.view_menu.name == f"[{database_without_catalog.name}].[schema2]"


 def test_update_with_oauth2(
@@ -320,9 +342,15 @@ def test_update_with_oauth2(
    """
    Test that the database can be updated even if OAuth2 is needed to connect.
    """
-    DatabaseDAO = mocker.patch("superset.commands.database.update.DatabaseDAO")  # noqa: N806
-    DatabaseDAO.find_by_id.return_value = database_needs_oauth2
-    DatabaseDAO.update.return_value = database_needs_oauth2
+    database_dao = mocker.patch("superset.commands.database.update.DatabaseDAO")
+    database_dao.find_by_id.return_value = database_needs_oauth2
+    database_dao.update.return_value = database_needs_oauth2
+    sync_db_perms_dao = mocker.patch(
+        "superset.commands.database.sync_permissions.DatabaseDAO"
+    )
+    sync_db_perms_dao.find_by_id.return_value = database_needs_oauth2
+    mocker.patch("superset.commands.database.update.get_username")
+    mocker.patch("superset.security_manager.get_user_by_username")

    find_permission_view_menu = mocker.patch.object(
        security_manager,
@@ -332,14 +360,11 @@ def test_update_with_oauth2(
        None,  # schema1 has no permissions
        "[my_db].[schema2]",  # second schema already exists
    ]
-    add_permission_view_menu = mocker.patch.object(
-        security_manager,
-        "add_permission_view_menu",
-    )
+    add_pvm = mocker.patch("superset.commands.database.sync_permissions.add_pvm")

    UpdateDatabaseCommand(1, {}).run()

-    add_permission_view_menu.assert_not_called()
+    add_pvm.assert_not_called()
    database_needs_oauth2.purge_oauth2_tokens.assert_not_called()


@@ -350,9 +375,15 @@ def test_update_with_oauth2_changed(
    """
    Test that the database can be updated even if OAuth2 is needed to connect.
    """
-    DatabaseDAO = mocker.patch("superset.commands.database.update.DatabaseDAO")  # noqa: N806
-    DatabaseDAO.find_by_id.return_value = database_needs_oauth2
-    DatabaseDAO.update.return_value = database_needs_oauth2
+    database_dao = mocker.patch("superset.commands.database.update.DatabaseDAO")
+    database_dao.find_by_id.return_value = database_needs_oauth2
+    database_dao.update.return_value = database_needs_oauth2
+    sync_db_perms_dao = mocker.patch(
+        "superset.commands.database.sync_permissions.DatabaseDAO"
+    )
+    sync_db_perms_dao.find_by_id.return_value = database_needs_oauth2
+    mocker.patch("superset.commands.database.update.get_username")
+    mocker.patch("superset.security_manager.get_user_by_username")

    find_permission_view_menu = mocker.patch.object(
        security_manager,
@@ -362,10 +393,7 @@ def test_update_with_oauth2_changed(
        None,  # schema1 has no permissions
        "[my_db].[schema2]",  # second schema already exists
    ]
-    add_permission_view_menu = mocker.patch.object(
-        security_manager,
-        "add_permission_view_menu",
-    )
+    add_pvm = mocker.patch("superset.commands.database.sync_permissions.add_pvm")

    modified_oauth2_client_info = oauth2_client_info.copy()
    modified_oauth2_client_info["scope"] = "scope-b"
@@ -379,7 +407,7 @@ def test_update_with_oauth2_changed(
        },
    ).run()

-    add_permission_view_menu.assert_not_called()
+    add_pvm.assert_not_called()
    database_needs_oauth2.purge_oauth2_tokens.assert_called()


@@ -390,9 +418,15 @@ def test_remove_oauth_config_purges_tokens(
    """
    Test that removing the OAuth config from a database purges existing tokens.
    """
-    DatabaseDAO = mocker.patch("superset.commands.database.update.DatabaseDAO")  # noqa: N806
-    DatabaseDAO.find_by_id.return_value = database_needs_oauth2
-    DatabaseDAO.update.return_value = database_needs_oauth2
+    database_dao = mocker.patch("superset.commands.database.update.DatabaseDAO")
+    database_dao.find_by_id.return_value = database_needs_oauth2
+    database_dao.update.return_value = database_needs_oauth2
+    sync_db_perms_dao = mocker.patch(
+        "superset.commands.database.sync_permissions.DatabaseDAO"
+    )
+    sync_db_perms_dao.find_by_id.return_value = database_needs_oauth2
+    mocker.patch("superset.commands.database.update.get_username")
+    mocker.patch("superset.security_manager.get_user_by_username")

    find_permission_view_menu = mocker.patch.object(
        security_manager,
@@ -402,19 +436,16 @@ def test_remove_oauth_config_purges_tokens(
        None,
        "[my_db].[schema2]",
    ]
-    add_permission_view_menu = mocker.patch.object(
-        security_manager,
-        "add_permission_view_menu",
-    )
+    add_pvm = mocker.patch("superset.commands.database.sync_permissions.add_pvm")

    UpdateDatabaseCommand(1, {"masked_encrypted_extra": None}).run()

-    add_permission_view_menu.assert_not_called()
+    add_pvm.assert_not_called()
    database_needs_oauth2.purge_oauth2_tokens.assert_called()

    UpdateDatabaseCommand(1, {"masked_encrypted_extra": "{}"}).run()

-    add_permission_view_menu.assert_not_called()
+    add_pvm.assert_not_called()
    database_needs_oauth2.purge_oauth2_tokens.assert_called()


@@ -425,9 +456,15 @@ def test_update_oauth2_removes_masked_encrypted_extra_key(
    """
    Test that the ``masked_encrypted_extra`` key is properly purged from the properties.
    """
-    DatabaseDAO = mocker.patch("superset.commands.database.update.DatabaseDAO")  # noqa: N806
-    DatabaseDAO.find_by_id.return_value = database_needs_oauth2
-    DatabaseDAO.update.return_value = database_needs_oauth2
+    database_dao = mocker.patch("superset.commands.database.update.DatabaseDAO")
+    database_dao.find_by_id.return_value = database_needs_oauth2
+    database_dao.update.return_value = database_needs_oauth2
+    sync_db_perms_dao = mocker.patch(
+        "superset.commands.database.sync_permissions.DatabaseDAO"
+    )
+    sync_db_perms_dao.find_by_id.return_value = database_needs_oauth2
+    mocker.patch("superset.commands.database.update.get_username")
+    mocker.patch("superset.security_manager.get_user_by_username")

    find_permission_view_menu = mocker.patch.object(
        security_manager,
@@ -437,10 +474,7 @@ def test_update_oauth2_removes_masked_encrypted_extra_key(
        None,
        "[my_db].[schema2]",
    ]
-    add_permission_view_menu = mocker.patch.object(
-        security_manager,
-        "add_permission_view_menu",
-    )
+    add_pvm = mocker.patch("superset.commands.database.sync_permissions.add_pvm")

    modified_oauth2_client_info = oauth2_client_info.copy()
    modified_oauth2_client_info["scope"] = "scope-b"
@@ -454,9 +488,9 @@ def test_update_oauth2_removes_masked_encrypted_extra_key(
        },
    ).run()

-    add_permission_view_menu.assert_not_called()
+    add_pvm.assert_not_called()
    database_needs_oauth2.purge_oauth2_tokens.assert_called()
-    DatabaseDAO.update.assert_called_with(
+    database_dao.update.assert_called_with(
        database_needs_oauth2,
        {
            "encrypted_extra": json.dumps(
@@ -474,9 +508,15 @@ def test_update_other_fields_dont_affect_oauth(
    Test that not including ``masked_encrypted_extra`` in the payload does not
    touch the OAuth config.
    """
-    DatabaseDAO = mocker.patch("superset.commands.database.update.DatabaseDAO")  # noqa: N806
-    DatabaseDAO.find_by_id.return_value = database_needs_oauth2
-    DatabaseDAO.update.return_value = database_needs_oauth2
+    database_dao = mocker.patch("superset.commands.database.update.DatabaseDAO")
+    database_dao.find_by_id.return_value = database_needs_oauth2
+    database_dao.update.return_value = database_needs_oauth2
+    sync_db_perms_dao = mocker.patch(
+        "superset.commands.database.sync_permissions.DatabaseDAO"
+    )
+    sync_db_perms_dao.find_by_id.return_value = database_needs_oauth2
+    mocker.patch("superset.commands.database.update.get_username")
+    mocker.patch("superset.security_manager.get_user_by_username")

    find_permission_view_menu = mocker.patch.object(
        security_manager,
@@ -486,12 +526,9 @@ def test_update_other_fields_dont_affect_oauth(
        None,
        "[my_db].[schema2]",
    ]
-    add_permission_view_menu = mocker.patch.object(
-        security_manager,
-        "add_permission_view_menu",
-    )
+    add_pvm = mocker.patch("superset.commands.database.sync_permissions.add_pvm")

    UpdateDatabaseCommand(1, {"database_name": "New DB name"}).run()

-    add_permission_view_menu.assert_not_called()
+    add_pvm.assert_not_called()
    database_needs_oauth2.purge_oauth2_tokens.assert_not_called()
--- a/tests/unit_tests/commands/databases/utils_test.py
+++ b/tests/unit_tests/commands/databases/utils_test.py
@@ -0,0 +1,212 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations
+
+import datetime
+import sqlite3
+from unittest.mock import MagicMock
+
+import pytest
+from flask_appbuilder.security.sqla.models import (
+    Permission,
+    PermissionView,
+    ViewMenu,
+)
+from pytest_mock import MockerFixture
+from sqlalchemy.orm import Session
+
+from superset.commands.database.utils import (
+    add_perm,
+    add_pvm,
+    add_vm,
+    ping,
+)
+from tests.conftest import with_config
+
+
+@pytest.fixture
+def mock_engine(mocker: MockerFixture) -> tuple[MagicMock, MagicMock, MagicMock]:
+    mock_connection = mocker.MagicMock()
+    mock_engine = mocker.MagicMock()
+    mock_dialect = mocker.MagicMock()
+    mock_engine.raw_connection.return_value = mock_connection
+    mock_engine.dialect = mock_dialect
+    return mock_engine, mock_connection, mock_dialect
+
+
+@with_config({"TEST_DATABASE_CONNECTION_TIMEOUT": datetime.timedelta(seconds=10)})
+def test_ping_success(mock_engine: MockerFixture):
+    """
+    Test the ``ping`` method.
+    """
+    mock_engine, mock_connection, mock_dialect = mock_engine
+    mock_dialect.do_ping.return_value = True
+
+    result = ping(mock_engine)
+
+    assert result is True
+
+    mock_engine.raw_connection.assert_called_once()
+    mock_dialect.do_ping.assert_called_once_with(mock_connection)
+
+
+@with_config({"TEST_DATABASE_CONNECTION_TIMEOUT": datetime.timedelta(seconds=10)})
+def test_ping_sqlite_exception(mocker: MockerFixture, mock_engine: MockerFixture):
+    """
+    Test the ``ping`` method when a sqlite3.ProgrammingError is raised.
+    """
+    mock_engine, mock_connection, mock_dialect = mock_engine
+    mock_dialect.do_ping.side_effect = [sqlite3.ProgrammingError, True]
+
+    result = ping(mock_engine)
+
+    assert result is True
+
+    mock_dialect.do_ping.assert_has_calls(
+        [mocker.call(mock_connection), mocker.call(mock_engine)]
+    )
+
+
+def test_ping_runtime_exception(mocker: MockerFixture, mock_engine: MockerFixture):
+    """
+    Test the ``ping`` method when a RuntimeError is raised.
+    """
+    mock_engine, _, mock_dialect = mock_engine
+    mock_timeout = mocker.patch("superset.commands.database.utils.timeout")
+    mock_timeout.side_effect = RuntimeError("timeout")
+    mock_dialect.do_ping.return_value = True
+
+    result = ping(mock_engine)
+
+    assert result is True
+    mock_dialect.do_ping.assert_called_once_with(mock_engine)
+
+
+@pytest.fixture
+def db_session(mocker: MockerFixture) -> Session:
+    return mocker.MagicMock(spec=Session)
+
+
+def test_add_vm(db_session: Session, mocker: MockerFixture):
+    """
+    Thest ``add_vm`` when the ViewMenu does not exist.
+    """
+    sm = mocker.MagicMock()
+    sm.find_view_menu.return_value = None
+    sm.viewmenu_model = ViewMenu
+
+    result = add_vm(db_session, sm, "new_view_menu")
+
+    assert result.name == "new_view_menu"
+    sm.find_view_menu.assert_called_once_with("new_view_menu")
+    db_session.add.assert_called_once_with(result)
+
+
+def test_add_vm_existing(db_session: Session, mocker: MockerFixture):
+    """
+    Thest ``add_vm`` when the ViewMenu already exists.
+    """
+    mock_vm = mocker.MagicMock()
+    sm = mocker.MagicMock()
+    sm.find_view_menu.return_value = mock_vm
+
+    result = add_vm(db_session, sm, "existing_view_menu")
+
+    assert result == mock_vm
+    sm.find_view_menu.assert_called_once_with("existing_view_menu")
+    db_session.add.assert_not_called()
+
+
+def test_add_perm(db_session: Session, mocker: MockerFixture):
+    """
+    Thest ``add_perm`` when the Permission does not exist.
+    """
+    sm = mocker.MagicMock()
+    sm.find_permission.return_value = None
+    sm.permission_model = Permission
+
+    result = add_perm(db_session, sm, "new_perm")
+
+    assert result.name == "new_perm"
+    sm.find_permission.assert_called_once_with("new_perm")
+    db_session.add.assert_called_once_with(result)
+
+
+def test_add_perm_existing(db_session: Session, mocker: MockerFixture):
+    """
+    Thest ``add_perm`` when the Permission already exists.
+    """
+    mock_perm = mocker.MagicMock()
+    sm = mocker.MagicMock()
+    sm.find_permission.return_value = mock_perm
+
+    result = add_perm(db_session, sm, "existing_perm")
+
+    assert result == mock_perm
+    sm.find_permission.assert_called_once_with("existing_perm")
+    db_session.add.assert_not_called()
+
+
+def test_add_pvm(db_session: Session, mocker: MockerFixture):
+    """
+    Thest ``add_pvm`` when the PermissionView does not exist.
+    """
+    sm = mocker.MagicMock()
+    sm.find_permission_view_menu.return_value = None
+    sm.permissionview_model = PermissionView
+
+    mock_vm = mocker.MagicMock()
+    mock_perm = mocker.MagicMock()
+    mock_add_vm = mocker.patch("superset.commands.database.utils.add_vm")
+    mock_add_vm.return_value = mock_vm
+    mock_add_perm = mocker.patch("superset.commands.database.utils.add_perm")
+    mock_add_perm.return_value = mock_perm
+
+    result = add_pvm(db_session, sm, "new_perm", "new_view_menu")
+
+    assert result is not None
+    assert result.view_menu == mock_vm
+    assert result.permission == mock_perm
+    sm.find_permission_view_menu.assert_called_once_with("new_perm", "new_view_menu")
+    mock_add_vm.assert_called_once_with(db_session, sm, "new_view_menu")
+    mock_add_perm.assert_called_once_with(db_session, sm, "new_perm")
+    db_session.add.assert_called_once_with(result)
+
+
+def test_add_pvm_missing_data(db_session: Session, mocker: MockerFixture):
+    """
+    Thest ``add_pvm`` when permission_name and view_menu_name are empty.
+    """
+    sm = mocker.MagicMock()
+    result = add_pvm(db_session, sm, None, None)
+
+    assert result is None
+
+
+def test_add_pvm_existing(db_session: Session, mocker: MockerFixture):
+    """
+    Thest ``add_pvm`` when the PermissionView already exists.
+    """
+    mock_pvm = mocker.MagicMock()
+    sm = mocker.MagicMock()
+    sm.find_permission_view_menu.return_value = mock_pvm
+
+    result = add_pvm(db_session, sm, "existinf_perm", "existing_vm")
+
+    assert result == mock_pvm
+    sm.find_permission_view_menu.assert_called_once_with("existinf_perm", "existing_vm")
+    db_session.add.assert_not_called()