# 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
from superset.models import core as models
from tests.integration_tests.dashboards.base_case import DashboardTestCase
from tests.integration_tests.dashboards.consts import *
from tests.integration_tests.dashboards.dashboard_test_utils import *
from tests.integration_tests.dashboards.superset_factory_util import *
from tests.integration_tests.fixtures.energy_dashboard import (
    load_energy_table_data,
    load_energy_table_with_slice,
)


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) for url in dashboard_title_by_url.keys()
        }

        # assert
        for dashboard_url, get_dashboard_response in responses_by_url.items():
            self.assert200(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",
            published=False,
            owners=[user],
        )

        not_my_owned_dashboard = create_dashboard_to_db(
            dashboard_title="Not My Dashboard",
            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)
        admin_and_draft_dashboard = create_dashboard_to_db(
            dashboard_title="admin_owned_unpublished_dash", 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_draft_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], should_create_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",
            published=True,
            slices=[slice_to_add_to_dashboards],
        )

        second_dash = create_dashboard_to_db(
            dashboard_title="Hidden Dashboard",
            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)

    def test_get_dashboards_api_no_data_access(self):
        """
        Dashboard API: Test get dashboards no data access
        """
        admin = self.get_user("admin")
        title = f"title{random_str()}"
        create_dashboard_to_db(title, "slug1", owners=[admin])

        self.login(username="gamma")
        arguments = {
            "filters": [{"col": "dashboard_title", "opr": "sw", "value": title[0:8]}]
        }
        uri = DASHBOARDS_API_URL_WITH_QUERY_FORMAT.format(prison.dumps(arguments))
        rv = self.client.get(uri)
        self.assert200(rv)
        data = json.loads(rv.data.decode("utf-8"))
        self.assertEqual(0, data["count"])
