QT/superset2
1
0
Fork 0
mirror of https://github.com/apache/superset.git synced 2026-04-19 08:04:53 +00:00
Code Issues Packages Projects Releases Wiki Activity

chore(tags): Handle tagging as part of asset update call (#28570)

Browse Source
This commit is contained in:
Vitor Avila
2024-05-28 12:41:31 -03:00
committed by GitHub
parent 475364248f
commit 0fdb4b7c23
17 changed files with 1075 additions and 167 deletions
Download Patch File Download Diff File Expand all files Collapse all files

367
tests/unit_tests/commands/test_utils.py
View File

@@ -15,19 +15,74 @@
# specific language governing permissions and limitations
# under the License.
from unittest.mock import MagicMock, patch
from superset.commands.utils import compute_owner_list, populate_owner_list, User
from unittest.mock import call, MagicMock, patch
import pytest
from superset.commands.exceptions import TagForbiddenError, TagNotFoundValidationError
from superset.commands.utils import (
compute_owner_list,
populate_owner_list,
Tag,
TagType,
update_tags,
User,
validate_tags,
)
from superset.tags.models import ObjectType
OBJECT_TYPES = {ObjectType.chart, ObjectType.chart}
MOCK_TAGS = [
Tag(
id=1,
name="first",
type=TagType.custom,
),
Tag(
id=2,
name="second",
type=TagType.custom,
),
Tag(
id=3,
name="third",
type=TagType.custom,
),
Tag(
id=4,
name="type:dashboard",
type=TagType.type,
),
Tag(
id=4,
name="owner:1",
type=TagType.owner,
),
Tag(
id=4,
name="avorited_by:2",
type=TagType.favorited_by,
),
]
@patch("superset.commands.utils.g")
def test_populate_owner_list_default_to_user(mock_user):
"""
Test the ``populate_owner_list`` method when no owners are provided
and default_to_user is True (non-admin).
"""
owner_list = populate_owner_list([], True)
assert owner_list == [mock_user.user]
@patch("superset.commands.utils.g")
def test_populate_owner_list_default_to_user_handle_none(mock_user):
"""
Test the ``populate_owner_list`` method when owners is None
and default_to_user is True (non-admin).
"""
owner_list = populate_owner_list(None, True)
assert owner_list == [mock_user.user]
@@ -36,6 +91,10 @@ def test_populate_owner_list_default_to_user_handle_none(mock_user):
@patch("superset.commands.utils.security_manager")
@patch("superset.commands.utils.get_user_id")
def test_populate_owner_list_admin_user(mock_user_id, mock_sm, mock_g):
"""
Test the ``populate_owner_list`` method when an admin is setting
another user as an owner and default_to_user is False.
"""
test_user = User(id=1, first_name="First", last_name="Last")
mock_g.user = User(id=4, first_name="Admin", last_name="User")
mock_user_id.return_value = 4
@@ -50,6 +109,10 @@ def test_populate_owner_list_admin_user(mock_user_id, mock_sm, mock_g):
@patch("superset.commands.utils.security_manager")
@patch("superset.commands.utils.get_user_id")
def test_populate_owner_list_admin_user_empty_list(mock_user_id, mock_sm, mock_g):
"""
Test the ``populate_owner_list`` method when an admin is setting an empty list
of owners.
"""
mock_g.user = User(id=4, first_name="Admin", last_name="User")
mock_user_id.return_value = 4
mock_sm.is_admin = MagicMock(return_value=True)
@@ -61,6 +124,10 @@ def test_populate_owner_list_admin_user_empty_list(mock_user_id, mock_sm, mock_g
@patch("superset.commands.utils.security_manager")
@patch("superset.commands.utils.get_user_id")
def test_populate_owner_list_non_admin(mock_user_id, mock_sm, mock_g):
"""
Test the ``populate_owner_list`` method when a non admin is adding
another user as an owner and default_to_user is False (both get added).
"""
test_user = User(id=1, first_name="First", last_name="Last")
mock_g.user = User(id=4, first_name="Non", last_name="admin")
mock_user_id.return_value = 4
@@ -73,6 +140,9 @@ def test_populate_owner_list_non_admin(mock_user_id, mock_sm, mock_g):
@patch("superset.commands.utils.populate_owner_list")
def test_compute_owner_list_new_owners(mock_populate_owner_list):
"""
Test the ``compute_owner_list`` method when replacing the owner list.
"""
current_owners = [User(id=1), User(id=2), User(id=3)]
new_owners = [4, 5, 6]
@@ -82,6 +152,9 @@ def test_compute_owner_list_new_owners(mock_populate_owner_list):
@patch("superset.commands.utils.populate_owner_list")
def test_compute_owner_list_no_new_owners(mock_populate_owner_list):
"""
Test the ``compute_owner_list`` method when replacing new_owners is None.
"""
current_owners = [User(id=1), User(id=2), User(id=3)]
new_owners = None
@@ -91,6 +164,9 @@ def test_compute_owner_list_no_new_owners(mock_populate_owner_list):
@patch("superset.commands.utils.populate_owner_list")
def test_compute_owner_list_new_owner_empty_list(mock_populate_owner_list):
"""
Test the ``compute_owner_list`` method when new_owners is an empty list.
"""
current_owners = [User(id=1), User(id=2), User(id=3)]
new_owners = []
@@ -100,6 +176,9 @@ def test_compute_owner_list_new_owner_empty_list(mock_populate_owner_list):
@patch("superset.commands.utils.populate_owner_list")
def test_compute_owner_list_no_owners(mock_populate_owner_list):
"""
Test the ``compute_owner_list`` method when current ownership is an empty list.
"""
current_owners = []
new_owners = [4, 5, 6]
@@ -109,8 +188,292 @@ def test_compute_owner_list_no_owners(mock_populate_owner_list):
@patch("superset.commands.utils.populate_owner_list")
def test_compute_owner_list_no_owners_handle_none(mock_populate_owner_list):
"""
Test the ``compute_owner_list`` method when current ownership is None.
"""
current_owners = None
new_owners = [4, 5, 6]
compute_owner_list(current_owners, new_owners)
mock_populate_owner_list.assert_called_once_with(new_owners, default_to_user=False)
@pytest.mark.parametrize("object_type", OBJECT_TYPES)
@patch("superset.commands.utils.security_manager")
def test_validate_tags_new_tags_is_none(mock_sm, object_type):
"""
Test the ``validate_tags`` method when new_tags is None.
"""
validate_tags(object_type, MOCK_TAGS, None)
mock_sm.can_access.assert_not_called()
@pytest.mark.parametrize("object_type", OBJECT_TYPES)
@patch("superset.commands.utils.security_manager")
def test_validate_tags_empty_list_can_write_on_tag(mock_sm, object_type):
"""
Test the ``validate_tags`` method when new_tags is an empty list and
user has permission to write on tag.
"""
mock_sm.can_access = MagicMock(return_value=True)
validate_tags(object_type, MOCK_TAGS, [])
mock_sm.can_access.assert_called_once_with("can_write", "Tag")
@pytest.mark.parametrize("object_type", OBJECT_TYPES)
@patch("superset.commands.utils.security_manager")
def test_validate_tags_empty_list_can_tag_on_object(mock_sm, object_type):
"""
Test the ``validate_tags`` method when new_tags is an empty list and
user has permission to tag objects.
"""
mock_sm.can_access = MagicMock(side_effect=[False, True])
validate_tags(object_type, MOCK_TAGS, [])
mock_sm.can_access.assert_has_calls(
[call("can_write", "Tag"), call("can_tag", object_type.name.capitalize())]
)
@pytest.mark.parametrize("object_type", OBJECT_TYPES)
@patch("superset.commands.utils.security_manager")
def test_validate_tags_empty_list_missing_permission(mock_sm, object_type):
"""
Test the ``validate_tags`` method when new_tags is an empty list and
the user doesn't have the required permission.
"""
mock_sm.can_access = MagicMock(side_effect=[False, False])
with pytest.raises(TagForbiddenError):
validate_tags(object_type, MOCK_TAGS, [])
mock_sm.can_access.assert_has_calls(
[call("can_write", "Tag"), call("can_tag", object_type.name.capitalize())]
)
@pytest.mark.parametrize("object_type", OBJECT_TYPES)
@patch("superset.commands.utils.security_manager")
def test_validate_tags_no_changes_can_write_on_tag(mock_sm, object_type):
"""
Test the ``validate_tags`` method when new_tags is equal to existing tags
and user has permission to write on tag.
"""
new_tags = [tag.id for tag in MOCK_TAGS if tag.type == TagType.custom]
validate_tags(object_type, MOCK_TAGS, new_tags)
mock_sm.can_access.assert_not_called()
@pytest.mark.parametrize("object_type", OBJECT_TYPES)
@patch("superset.commands.utils.security_manager")
def test_validate_tags_no_changes_can_tag_on_object(mock_sm, object_type):
"""
Test the ``validate_tags`` method when new_tags is equal to existing tags
and user has permission to tag objects.
"""
new_tags = [tag.id for tag in MOCK_TAGS if tag.type == TagType.custom]
validate_tags(object_type, MOCK_TAGS, new_tags)
mock_sm.can_access.assert_not_called()
@pytest.mark.parametrize("object_type", OBJECT_TYPES)
@patch("superset.commands.utils.security_manager")
def test_validate_tags_no_changes_missing_permission(mock_sm, object_type):
"""
Test the ``validate_tags`` method when new_tags is equal to existing tags
the user doens't have the required perms.
"""
new_tags = [tag.id for tag in MOCK_TAGS if tag.type == TagType.custom]
validate_tags(object_type, MOCK_TAGS, new_tags)
mock_sm.can_access.assert_not_called()
@pytest.mark.parametrize("object_type", OBJECT_TYPES)
@patch("superset.commands.utils.security_manager")
@patch("superset.commands.utils.TagDAO.find_by_id")
def test_validate_tags_add_new_tags_can_write_on_tag(
mock_tag_find_by_id, mock_sm, object_type
):
"""
Test the ``validate_tags`` method when new_tags are added and user has
permission to write on tag.
"""
new_tag_ids = [tag.id for tag in MOCK_TAGS if tag.type == TagType.custom]
new_tag = {
"id": 10,
"name": "New test tag",
"type": TagType.custom,
}
new_tag_ids.append(new_tag["id"])
mock_tag_find_by_id.return_value = new_tag
mock_sm.can_access = MagicMock(return_value=True)
validate_tags(object_type, MOCK_TAGS, new_tag_ids)
mock_sm.can_access.assert_called_once_with("can_write", "Tag")
@pytest.mark.parametrize("object_type", OBJECT_TYPES)
@patch("superset.commands.utils.security_manager")
@patch("superset.commands.utils.TagDAO.find_by_id")
def test_validate_tags_add_new_tags_can_tag_on_object(
mock_tag_find_by_id, mock_sm, object_type
):
"""
Test the ``validate_tags`` method when new_tags are added and user has
permission to tag objects.
"""
current_tags = [tag for tag in MOCK_TAGS if tag.type == TagType.custom]
new_tag = current_tags.pop()
new_tag_ids = [tag.id for tag in current_tags]
new_tag_ids.append(new_tag.id)
mock_sm.can_access = MagicMock(side_effect=[False, True])
mock_tag_find_by_id.return_value = new_tag
validate_tags(object_type, current_tags, new_tag_ids)
mock_sm.can_access.assert_has_calls(
[call("can_write", "Tag"), call("can_tag", object_type.name.capitalize())]
)
@pytest.mark.parametrize("object_type", OBJECT_TYPES)
@patch("superset.commands.utils.security_manager")
@patch("superset.commands.utils.TagDAO.find_by_name")
def test_validate_tags_can_write_on_tag_unable_to_find_tag(
mock_tag_find_by_id, mock_sm, object_type
):
"""
Test the ``validate_tags`` method when an un-existing tag is being
added and user has permission to write on tag.
"""
fake_id = 100
mock_sm.can_access = MagicMock(return_value=True)
mock_tag_find_by_id.return_value = None
with pytest.raises(TagNotFoundValidationError):
validate_tags(object_type, MOCK_TAGS, [fake_id])
mock_sm.can_access.assert_called_once_with("can_write", "Tag")
@pytest.mark.parametrize("object_type", OBJECT_TYPES)
@patch("superset.commands.utils.security_manager")
@patch("superset.commands.utils.TagDAO.find_by_name")
def test_validate_tags_can_tag_on_object_unable_to_find_tag(
mock_tag_find_by_id, mock_sm, object_type
):
"""
Test the ``validate_tags`` method when an un-existing tag is being
added and user has permission to tag on object.
"""
fake_id = 100
mock_sm.can_access = MagicMock(side_effect=[False, True])
mock_tag_find_by_id.return_value = None
with pytest.raises(TagNotFoundValidationError):
validate_tags(object_type, MOCK_TAGS, [fake_id])
mock_sm.can_access.assert_has_calls(
[call("can_write", "Tag"), call("can_tag", object_type.name.capitalize())]
)
@pytest.mark.parametrize("object_type", OBJECT_TYPES)
@patch("superset.commands.utils.TagDAO")
def test_update_tags_adding_tags(mock_tag_dao, object_type):
"""
Test the ``update_tags`` method when adding tags.
"""
current_tags = [tag for tag in MOCK_TAGS if tag.type == TagType.custom]
new_tag = current_tags.pop()
new_tags = [tag for tag in MOCK_TAGS if tag.type == TagType.custom]
new_tag_ids = [tag.id for tag in new_tags]
mock_tag_dao.find_by_ids.return_value = [new_tag]
update_tags(object_type, 1, current_tags, new_tag_ids)
mock_tag_dao.find_by_ids.assert_called_once_with([new_tag.id])
mock_tag_dao.delete_tagged_object.assert_not_called()
mock_tag_dao.create_custom_tagged_objects.assert_called_once_with(
object_type, 1, [new_tag.name]
)
@pytest.mark.parametrize("object_type", OBJECT_TYPES)
@patch("superset.commands.utils.TagDAO")
def test_update_tags_removing_tags(mock_tag_dao, object_type):
"""
Test the ``update_tags`` method when removing existing tags.
"""
new_tags = [tag for tag in MOCK_TAGS if tag.type == TagType.custom]
tag_to_be_removed = new_tags.pop()
new_tag_ids = [tag.id for tag in new_tags]
update_tags(object_type, 1, MOCK_TAGS, new_tag_ids)
mock_tag_dao.delete_tagged_object.assert_called_once_with(
object_type, 1, tag_to_be_removed.name
)
mock_tag_dao.create_custom_tagged_objects.assert_not_called()
@pytest.mark.parametrize("object_type", OBJECT_TYPES)
@patch("superset.commands.utils.TagDAO")
def test_update_tags_adding_and_removing_tags(mock_tag_dao, object_type):
"""
Test the ``update_tags`` method when adding and removing existing tags.
"""
new_tags = [tag for tag in MOCK_TAGS if tag.type == TagType.custom]
tag_to_be_removed = new_tags.pop()
new_tag = Tag(id=10, name="my new tag", type=TagType.custom)
new_tags.append(new_tag)
new_tag_ids = [tag.id for tag in new_tags]
mock_tag_dao.find_by_ids.return_value = [new_tag]
update_tags(object_type, 1, MOCK_TAGS, new_tag_ids)
mock_tag_dao.delete_tagged_object.assert_called_once_with(
object_type, 1, tag_to_be_removed.name
)
mock_tag_dao.find_by_ids.assert_called_once_with([new_tag.id])
mock_tag_dao.create_custom_tagged_objects.assert_called_once_with(
object_type, 1, ["my new tag"]
)
@pytest.mark.parametrize("object_type", OBJECT_TYPES)
@patch("superset.commands.utils.TagDAO")
def test_update_tags_removing_all_tags(mock_tag_dao, object_type):
"""
Test the ``update_tags`` method when removing all tags.
"""
update_tags(object_type, 1, MOCK_TAGS, [])
mock_tag_dao.delete_tagged_object.assert_has_calls(
[
call(object_type, 1, tag.name)
for tag in MOCK_TAGS
if tag.type == TagType.custom
]
)
mock_tag_dao.create_custom_tagged_objects.assert_not_called()
@pytest.mark.parametrize("object_type", OBJECT_TYPES)
@patch("superset.commands.utils.TagDAO")
def test_update_tags_no_tags(mock_tag_dao, object_type):
"""
Test the ``update_tags`` method when the asset only has system tags.
"""
system_tags = [tag for tag in MOCK_TAGS if tag.type != TagType.custom]
new_tags = [tag for tag in MOCK_TAGS if tag.type == TagType.custom]
new_tag_ids = [tag.id for tag in new_tags]
new_tag_names = [tag.name for tag in new_tags]
mock_tag_dao.find_by_ids.return_value = new_tags
update_tags(object_type, 1, system_tags, new_tag_ids)
mock_tag_dao.delete_tagged_object.assert_not_called()
mock_tag_dao.find_by_ids.assert_called_once_with(new_tag_ids)
mock_tag_dao.create_custom_tagged_objects.assert_called_once_with(
object_type, 1, new_tag_names
)