dashboard_tests.py

# 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.
# isort:skip_file
"""Unit tests for Superset"""
from datetime import datetime
import json
import unittest
from random import random

import pytest
from flask import escape, url_for
from sqlalchemy import func

from tests.fixtures.unicode_dashboard import load_unicode_dashboard_with_position
from tests.test_app import app
from superset import db, security_manager
from superset.connectors.sqla.models import SqlaTable
from superset.models import core as models
from superset.models.dashboard import Dashboard
from superset.models.slice import Slice
from tests.fixtures.energy_dashboard import load_energy_table_with_slice

from .base_tests import SupersetTestCase


class TestDashboard(SupersetTestCase):
    @pytest.fixture
    def cleanup_copied_dash(self):
        with app.app_context():
            original_dashboard = (
                db.session.query(Dashboard).filter_by(slug="births").first()
            )
            original_dashboard_id = original_dashboard.id
            yield
            copied_dashboard = (
                db.session.query(Dashboard)
                .filter(
                    Dashboard.dashboard_title == "Copy Of Births",
                    Dashboard.id != original_dashboard_id,
                )
                .first()
            )

            db.session.merge(original_dashboard)
            if copied_dashboard:
                db.session.delete(copied_dashboard)
            db.session.commit()

    @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()}"
            pytest.published_dash_slug = f"published_dash_{random()}"

            # 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 get_mock_positions(self, dash):
        positions = {"DASHBOARD_VERSION_KEY": "v2"}
        for i, slc in enumerate(dash.slices):
            id = "DASHBOARD_CHART_TYPE-{}".format(i)
            d = {
                "type": "CHART",
                "id": id,
                "children": [],
                "meta": {"width": 4, "height": 50, "chartId": slc.id},
            }
            positions[id] = d
        return positions

    def test_dashboard(self):
        self.login(username="admin")
        urls = {}
        for dash in db.session.query(Dashboard).all():
            urls[dash.dashboard_title] = dash.url
        for title, url in urls.items():
            assert escape(title) in self.client.get(url).data.decode("utf-8")

    def test_superset_dashboard_url(self):
        url_for("Superset.dashboard", dashboard_id_or_slug=1)

    def test_new_dashboard(self):
        self.login(username="admin")
        dash_count_before = db.session.query(func.count(Dashboard.id)).first()[0]
        url = "/dashboard/new/"
        resp = self.get_resp(url)
        self.assertIn("[ untitled dashboard ]", resp)
        dash_count_after = db.session.query(func.count(Dashboard.id)).first()[0]
        self.assertEqual(dash_count_before + 1, dash_count_after)

    def test_dashboard_modes(self):
        self.login(username="admin")
        dash = db.session.query(Dashboard).filter_by(slug="births").first()
        url = dash.url
        if dash.url.find("?") == -1:
            url += "?"
        else:
            url += "&"
        resp = self.get_resp(url + "edit=true&standalone=true")
        self.assertIn("editMode": true", resp)
        self.assertIn("standalone_mode": true", resp)
        self.assertIn('<body class="standalone">', resp)

    def test_save_dash(self, username="admin"):
        self.login(username=username)
        dash = db.session.query(Dashboard).filter_by(slug="births").first()
        positions = self.get_mock_positions(dash)
        data = {
            "css": "",
            "expanded_slices": {},
            "positions": positions,
            "dashboard_title": dash.dashboard_title,
            # set a further modified_time for unit test
            "last_modified_time": datetime.now().timestamp() + 1000,
        }
        url = "/superset/save_dash/{}/".format(dash.id)
        resp = self.get_resp(url, data=dict(data=json.dumps(data)))
        self.assertIn("SUCCESS", resp)

    def test_save_dash_with_filter(self, username="admin"):
        self.login(username=username)
        dash = db.session.query(Dashboard).filter_by(slug="world_health").first()

        positions = self.get_mock_positions(dash)
        filters = {str(dash.slices[0].id): {"region": ["North America"]}}
        default_filters = json.dumps(filters)
        data = {
            "css": "",
            "expanded_slices": {},
            "positions": positions,
            "dashboard_title": dash.dashboard_title,
            "default_filters": default_filters,
            # set a further modified_time for unit test
            "last_modified_time": datetime.now().timestamp() + 1000,
        }

        url = "/superset/save_dash/{}/".format(dash.id)
        resp = self.get_resp(url, data=dict(data=json.dumps(data)))
        self.assertIn("SUCCESS", resp)

        updatedDash = db.session.query(Dashboard).filter_by(slug="world_health").first()
        new_url = updatedDash.url
        self.assertIn("world_health", new_url)
        self.assertNotIn("preselect_filters", new_url)

        resp = self.get_resp(new_url)
        self.assertIn("North America", resp)

    def test_save_dash_with_invalid_filters(self, username="admin"):
        self.login(username=username)
        dash = db.session.query(Dashboard).filter_by(slug="world_health").first()

        # add an invalid filter slice
        positions = self.get_mock_positions(dash)
        filters = {str(99999): {"region": ["North America"]}}
        default_filters = json.dumps(filters)
        data = {
            "css": "",
            "expanded_slices": {},
            "positions": positions,
            "dashboard_title": dash.dashboard_title,
            "default_filters": default_filters,
            # set a further modified_time for unit test
            "last_modified_time": datetime.now().timestamp() + 1000,
        }

        url = "/superset/save_dash/{}/".format(dash.id)
        resp = self.get_resp(url, data=dict(data=json.dumps(data)))
        self.assertIn("SUCCESS", resp)

        updatedDash = db.session.query(Dashboard).filter_by(slug="world_health").first()
        new_url = updatedDash.url
        self.assertNotIn("region", new_url)

    def test_save_dash_with_dashboard_title(self, username="admin"):
        self.login(username=username)
        dash = db.session.query(Dashboard).filter_by(slug="births").first()
        origin_title = dash.dashboard_title
        positions = self.get_mock_positions(dash)
        data = {
            "css": "",
            "expanded_slices": {},
            "positions": positions,
            "dashboard_title": "new title",
            # set a further modified_time for unit test
            "last_modified_time": datetime.now().timestamp() + 1000,
        }
        url = "/superset/save_dash/{}/".format(dash.id)
        self.get_resp(url, data=dict(data=json.dumps(data)))
        updatedDash = db.session.query(Dashboard).filter_by(slug="births").first()
        self.assertEqual(updatedDash.dashboard_title, "new title")
        # bring back dashboard original title
        data["dashboard_title"] = origin_title
        self.get_resp(url, data=dict(data=json.dumps(data)))

    def test_save_dash_with_colors(self, username="admin"):
        self.login(username=username)
        dash = db.session.query(Dashboard).filter_by(slug="births").first()
        positions = self.get_mock_positions(dash)
        new_label_colors = {"data value": "random color"}
        data = {
            "css": "",
            "expanded_slices": {},
            "positions": positions,
            "dashboard_title": dash.dashboard_title,
            "color_namespace": "Color Namespace Test",
            "color_scheme": "Color Scheme Test",
            "label_colors": new_label_colors,
            # set a further modified_time for unit test
            "last_modified_time": datetime.now().timestamp() + 1000,
        }
        url = "/superset/save_dash/{}/".format(dash.id)
        self.get_resp(url, data=dict(data=json.dumps(data)))
        updatedDash = db.session.query(Dashboard).filter_by(slug="births").first()
        self.assertIn("color_namespace", updatedDash.json_metadata)
        self.assertIn("color_scheme", updatedDash.json_metadata)
        self.assertIn("label_colors", updatedDash.json_metadata)
        # bring back original dashboard
        del data["color_namespace"]
        del data["color_scheme"]
        del data["label_colors"]
        self.get_resp(url, data=dict(data=json.dumps(data)))

    @pytest.mark.usefixtures(
        "cleanup_copied_dash", "load_unicode_dashboard_with_position"
    )
    def test_copy_dash(self, username="admin"):
        self.login(username=username)
        dash = db.session.query(Dashboard).filter_by(slug="births").first()
        positions = self.get_mock_positions(dash)
        new_label_colors = {"data value": "random color"}
        data = {
            "css": "",
            "duplicate_slices": False,
            "expanded_slices": {},
            "positions": positions,
            "dashboard_title": "Copy Of Births",
            "color_namespace": "Color Namespace Test",
            "color_scheme": "Color Scheme Test",
            "label_colors": new_label_colors,
            # set a further modified_time for unit test
            "last_modified_time": datetime.now().timestamp() + 1000,
        }

        # Save changes to Births dashboard and retrieve updated dash
        dash_id = dash.id
        url = "/superset/save_dash/{}/".format(dash_id)
        self.client.post(url, data=dict(data=json.dumps(data)))
        dash = db.session.query(Dashboard).filter_by(id=dash_id).first()
        orig_json_data = dash.data

        # Verify that copy matches original
        url = "/superset/copy_dash/{}/".format(dash_id)
        resp = self.get_json_resp(url, data=dict(data=json.dumps(data)))
        self.assertEqual(resp["dashboard_title"], "Copy Of Births")
        self.assertEqual(resp["position_json"], orig_json_data["position_json"])
        self.assertEqual(resp["metadata"], orig_json_data["metadata"])
        # check every attribute in each dashboard's slices list,
        # exclude modified and changed_on attribute
        for index, slc in enumerate(orig_json_data["slices"]):
            for key in slc:
                if key not in ["modified", "changed_on", "changed_on_humanized"]:
                    self.assertEqual(slc[key], resp["slices"][index][key])

    @pytest.mark.usefixtures("load_energy_table_with_slice")
    def test_add_slices(self, username="admin"):
        self.login(username=username)
        dash = db.session.query(Dashboard).filter_by(slug="births").first()
        new_slice = (
            db.session.query(Slice).filter_by(slice_name="Energy Force Layout").first()
        )
        existing_slice = (
            db.session.query(Slice).filter_by(slice_name="Girl Name Cloud").first()
        )
        data = {
            "slice_ids": [new_slice.data["slice_id"], existing_slice.data["slice_id"]]
        }
        url = "/superset/add_slices/{}/".format(dash.id)
        resp = self.client.post(url, data=dict(data=json.dumps(data)))
        assert "SLICES ADDED" in resp.data.decode("utf-8")

        dash = db.session.query(Dashboard).filter_by(slug="births").first()
        new_slice = (
            db.session.query(Slice).filter_by(slice_name="Energy Force Layout").first()
        )
        assert new_slice in dash.slices
        assert len(set(dash.slices)) == len(dash.slices)

        # cleaning up
        dash = db.session.query(Dashboard).filter_by(slug="births").first()
        dash.slices = [o for o in dash.slices if o.slice_name != "Energy Force Layout"]
        db.session.commit()

    def test_remove_slices(self, username="admin"):
        self.login(username=username)
        dash = db.session.query(Dashboard).filter_by(slug="births").first()
        origin_slices_length = len(dash.slices)

        positions = self.get_mock_positions(dash)
        # remove one chart
        chart_keys = []
        for key in positions.keys():
            if key.startswith("DASHBOARD_CHART_TYPE"):
                chart_keys.append(key)
        positions.pop(chart_keys[0])

        data = {
            "css": "",
            "expanded_slices": {},
            "positions": positions,
            "dashboard_title": dash.dashboard_title,
            # set a further modified_time for unit test
            "last_modified_time": datetime.now().timestamp() + 1000,
        }

        # save dash
        dash_id = dash.id
        url = "/superset/save_dash/{}/".format(dash_id)
        self.client.post(url, data=dict(data=json.dumps(data)))
        dash = db.session.query(Dashboard).filter_by(id=dash_id).first()

        # verify slices data
        data = dash.data
        self.assertEqual(len(data["slices"]), origin_slices_length - 1)

    def test_public_user_dashboard_access(self):
        table = db.session.query(SqlaTable).filter_by(table_name="birth_names").one()

        # Make the births dash published so it can be seen
        births_dash = db.session.query(Dashboard).filter_by(slug="births").one()
        births_dash.published = True

        db.session.merge(births_dash)
        db.session.commit()

        # Try access before adding appropriate permissions.
        self.revoke_public_access_to_table(table)
        self.logout()

        resp = self.get_resp("/api/v1/chart/")
        self.assertNotIn("birth_names", resp)

        resp = self.get_resp("/api/v1/dashboard/")
        self.assertNotIn("/superset/dashboard/births/", resp)

        self.grant_public_access_to_table(table)

        # Try access after adding appropriate permissions.
        self.assertIn("birth_names", self.get_resp("/api/v1/chart/"))

        resp = self.get_resp("/api/v1/dashboard/")
        self.assertIn("/superset/dashboard/births/", resp)

        self.assertIn("Births", self.get_resp("/superset/dashboard/births/"))

        # Confirm that public doesn't have access to other datasets.
        resp = self.get_resp("/api/v1/chart/")
        self.assertNotIn("wb_health_population", resp)

        resp = self.get_resp("/api/v1/dashboard/")
        self.assertNotIn("/superset/dashboard/world_health/", resp)

        # Cleanup
        self.revoke_public_access_to_table(table)

    def test_dashboard_with_created_by_can_be_accessed_by_public_users(self):
        self.logout()
        table = db.session.query(SqlaTable).filter_by(table_name="birth_names").one()
        self.grant_public_access_to_table(table)

        dash = db.session.query(Dashboard).filter_by(slug="births").first()
        dash.owners = [security_manager.find_user("admin")]
        dash.created_by = security_manager.find_user("admin")
        db.session.merge(dash)
        db.session.commit()

        assert "Births" in self.get_resp("/superset/dashboard/births/")
        # Cleanup
        self.revoke_public_access_to_table(table)

    def test_only_owners_can_save(self):
        dash = db.session.query(Dashboard).filter_by(slug="births").first()
        dash.owners = []
        db.session.merge(dash)
        db.session.commit()
        self.test_save_dash("admin")

        self.logout()
        self.assertRaises(Exception, self.test_save_dash, "alpha")

        alpha = security_manager.find_user("alpha")

        dash = db.session.query(Dashboard).filter_by(slug="births").first()
        dash.owners = [alpha]
        db.session.merge(dash)
        db.session.commit()
        self.test_save_dash("alpha")

    def test_owners_can_view_empty_dashboard(self):
        dash = db.session.query(Dashboard).filter_by(slug="empty_dashboard").first()
        if not dash:
            dash = Dashboard()
            dash.dashboard_title = "Empty Dashboard"
            dash.slug = "empty_dashboard"
        else:
            dash.slices = []
            dash.owners = []
        db.session.merge(dash)
        db.session.commit()

        gamma_user = security_manager.find_user("gamma")
        self.login(gamma_user.username)

        resp = self.get_resp("/api/v1/dashboard/")
        self.assertNotIn("/superset/dashboard/empty_dashboard/", resp)

    @pytest.mark.usefixtures("load_energy_table_with_slice", "load_dashboard")
    def test_users_can_view_published_dashboard(self):
        resp = self.get_resp("/api/v1/dashboard/")
        self.assertNotIn(f"/superset/dashboard/{pytest.hidden_dash_slug}/", resp)
        self.assertIn(f"/superset/dashboard/{pytest.published_dash_slug}/", resp)

    def test_users_can_view_own_dashboard(self):
        user = security_manager.find_user("gamma")
        my_dash_slug = f"my_dash_{random()}"
        not_my_dash_slug = f"not_my_dash_{random()}"

        # Create one dashboard I own and another that I don't
        dash = Dashboard()
        dash.dashboard_title = "My Dashboard"
        dash.slug = my_dash_slug
        dash.owners = [user]
        dash.slices = []

        hidden_dash = Dashboard()
        hidden_dash.dashboard_title = "Not My Dashboard"
        hidden_dash.slug = not_my_dash_slug
        hidden_dash.slices = []
        hidden_dash.owners = []

        db.session.merge(dash)
        db.session.merge(hidden_dash)
        db.session.commit()

        self.login(user.username)

        resp = self.get_resp("/api/v1/dashboard/")
        self.assertIn(f"/superset/dashboard/{my_dash_slug}/", resp)
        self.assertNotIn(f"/superset/dashboard/{not_my_dash_slug}/", resp)

    def test_users_can_view_favorited_dashboards(self):
        user = security_manager.find_user("gamma")
        fav_dash_slug = f"my_favorite_dash_{random()}"
        regular_dash_slug = f"regular_dash_{random()}"

        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)

        resp = self.get_resp("/api/v1/dashboard/")
        self.assertIn(f"/superset/dashboard/{fav_dash_slug}/", resp)

    def test_user_can_not_view_unpublished_dash(self):
        admin_user = security_manager.find_user("admin")
        gamma_user = security_manager.find_user("gamma")
        slug = f"admin_owned_unpublished_dash_{random()}"

        # Create a dashboard owned by admin and unpublished
        dash = Dashboard()
        dash.dashboard_title = "My Dashboard"
        dash.slug = slug
        dash.owners = [admin_user]
        dash.slices = []
        dash.published = False
        db.session.merge(dash)
        db.session.commit()

        # list dashboards as a gamma user
        self.login(gamma_user.username)
        resp = self.get_resp("/api/v1/dashboard/")
        self.assertNotIn(f"/superset/dashboard/{slug}/", resp)


if __name__ == "__main__":
    unittest.main()