ktmud commented on a change in pull request #12680: URL: https://github.com/apache/superset/pull/12680#discussion_r566479017
########## File path: tests/dashboards/security/security_dataset_tests.py ########## @@ -0,0 +1,260 @@ +# 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. +"""Unit tests for Superset""" +import json + +import prison +import pytest +from flask import escape + +from superset import app, security_manager +from superset.models import core as models +from tests.dashboards.base_case import DashboardTestCase +from tests.dashboards.consts import * +from tests.dashboards.dashboard_test_utils import * +from tests.dashboards.superset_factory_util import * +from tests.fixtures.energy_dashboard import load_energy_table_with_slice + + +# @mark.amit +class TestDashboardDatasetSecurity(DashboardTestCase): + @pytest.fixture + def load_dashboard(self): + with app.app_context(): + table = ( + db.session.query(SqlaTable).filter_by(table_name="energy_usage").one() + ) + # get a slice from the allowed table + slice = db.session.query(Slice).filter_by(slice_name="Energy Sankey").one() + + self.grant_public_access_to_table(table) + + pytest.hidden_dash_slug = f"hidden_dash_{random_slug()}" + pytest.published_dash_slug = f"published_dash_{random_slug()}" + + # Create a published and hidden dashboard and add them to the database + published_dash = Dashboard() + published_dash.dashboard_title = "Published Dashboard" + published_dash.slug = pytest.published_dash_slug + published_dash.slices = [slice] + published_dash.published = True + + hidden_dash = Dashboard() + hidden_dash.dashboard_title = "Hidden Dashboard" + hidden_dash.slug = pytest.hidden_dash_slug + hidden_dash.slices = [slice] + hidden_dash.published = False + + db.session.merge(published_dash) + db.session.merge(hidden_dash) + yield db.session.commit() + + self.revoke_public_access_to_table(table) + db.session.delete(published_dash) + db.session.delete(hidden_dash) + db.session.commit() + + def test_dashboard_access__admin_can_access_all(self): + # arrange + self.login(username=ADMIN_USERNAME) + dashboard_title_by_url = { + dash.url: dash.dashboard_title for dash in get_all_dashboards() + } + + # act + responses_by_url = { + url: self.client.get(url).data.decode("utf-8") + for url in dashboard_title_by_url.keys() + } + + # assert + for dashboard_url, get_dashboard_response in responses_by_url.items(): + assert ( + escape(dashboard_title_by_url[dashboard_url]) in get_dashboard_response + ) + + def test_get_dashboards__users_are_dashboards_owners(self): + # arrange + username = "gamma" + user = security_manager.find_user(username) + my_owned_dashboard = create_dashboard_to_db( + dashboard_title="My Dashboard", + slug=f"my_dash_{random_slug()}", + published=False, + owners=[user], + ) + + not_my_owned_dashboard = create_dashboard_to_db( + dashboard_title="Not My Dashboard", + slug=f"not_my_dash_{random_slug()}", + published=False, + ) + + self.login(user.username) + + # act + get_dashboards_response = self.get_resp(DASHBOARDS_API_URL) + + # assert + self.assertIn(my_owned_dashboard.url, get_dashboards_response) + self.assertNotIn(not_my_owned_dashboard.url, get_dashboards_response) + + def test_get_dashboards__owners_can_view_empty_dashboard(self): + # arrange + dash = create_dashboard_to_db("Empty Dashboard", slug="empty_dashboard") + dashboard_url = dash.url + gamma_user = security_manager.find_user("gamma") + self.login(gamma_user.username) + + # act + get_dashboards_response = self.get_resp(DASHBOARDS_API_URL) + + # assert + self.assertNotIn(dashboard_url, get_dashboards_response) + + def test_get_dashboards__users_can_view_favorites_dashboards(self): + # arrange + user = security_manager.find_user("gamma") + fav_dash_slug = f"my_favorite_dash_{random_slug()}" + regular_dash_slug = f"regular_dash_{random_slug()}" + + favorite_dash = Dashboard() + favorite_dash.dashboard_title = "My Favorite Dashboard" + favorite_dash.slug = fav_dash_slug + + regular_dash = Dashboard() + regular_dash.dashboard_title = "A Plain Ol Dashboard" + regular_dash.slug = regular_dash_slug + + db.session.merge(favorite_dash) + db.session.merge(regular_dash) + db.session.commit() + + dash = db.session.query(Dashboard).filter_by(slug=fav_dash_slug).first() + + favorites = models.FavStar() + favorites.obj_id = dash.id + favorites.class_name = "Dashboard" + favorites.user_id = user.id + + db.session.merge(favorites) + db.session.commit() + + self.login(user.username) + + # act + get_dashboards_response = self.get_resp(DASHBOARDS_API_URL) + + # assert + self.assertIn(f"/superset/dashboard/{fav_dash_slug}/", get_dashboards_response) + + def test_get_dashboards__user_can_not_view_unpublished_dash(self): + # arrange + admin_user = security_manager.find_user(ADMIN_USERNAME) + gamma_user = security_manager.find_user(GAMMA_USERNAME) + slug = f"admin_owned_unpublished_dash_{random_slug()}" + + admin_and_not_published_dashboard = create_dashboard_to_db( + "My Dashboard", slug=slug, owners=[admin_user] + ) + + self.login(gamma_user.username) + + # act - list dashboards as a gamma user + get_dashboards_response_as_gamma = self.get_resp(DASHBOARDS_API_URL) + + # assert + self.assertNotIn( + admin_and_not_published_dashboard.url, get_dashboards_response_as_gamma + ) + + @pytest.mark.usefixtures("load_energy_table_with_slice", "load_dashboard") + def test_get_dashboards__users_can_view_permitted_dashboard(self): + # arrange + username = random_str() + new_role = f"role_{random_str()}" + self.create_user_with_roles(username, [new_role], copy_roles=True) + accessed_table = get_sql_table_by_name("energy_usage") + self.grant_role_access_to_table(accessed_table, new_role) + # get a slice from the allowed table + slice_to_add_to_dashboards = get_slice_by_name("Energy Sankey") + # Create a published and hidden dashboard and add them to the database + first_dash = create_dashboard_to_db( + dashboard_title="Published Dashboard", + slug=f"first_dash_{random_slug()}", + published=True, + slices=[slice_to_add_to_dashboards], + ) + + second_dash = create_dashboard_to_db( + dashboard_title="Hidden Dashboard", + slug=f"second_dash_{random_slug()}", + published=True, + slices=[slice_to_add_to_dashboards], + ) + + try: + self.login(username) + # act + get_dashboards_response = self.get_resp(DASHBOARDS_API_URL) + + # assert + self.assertIn(second_dash.url, get_dashboards_response) + self.assertIn(first_dash.url, get_dashboards_response) + finally: + self.revoke_public_access_to_table(accessed_table) + + @staticmethod + def __add_dashboard_mode_params(dashboard_url): Review comment: Was this function used anywhere? I couldn't find it. And why was it defined in two places? ---------------------------------------------------------------- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
