# 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 superset import db
# from superset.models.dashboard import Dashboard

import json
import urllib.request
from io import BytesIO
from typing import Tuple
from unittest import skipUnless
from unittest.mock import ANY, call, MagicMock, patch

import pytest
from flask_testing import LiveServerTestCase
from sqlalchemy.sql import func

from superset import db, is_feature_enabled, security_manager
from superset.extensions import machine_auth_provider_factory
from superset.models.dashboard import Dashboard
from superset.models.slice import Slice
from superset.tasks.types import ExecutorType
from superset.utils.screenshots import ChartScreenshot, DashboardScreenshot
from superset.utils.urls import get_url_path
from superset.utils.webdriver import find_unexpected_errors, WebDriverProxy
from tests.integration_tests.conftest import with_feature_flags
from tests.integration_tests.fixtures.birth_names_dashboard import (
    load_birth_names_dashboard_with_slices,
    load_birth_names_data,
)
from tests.integration_tests.test_app import app

from .base_tests import SupersetTestCase

CHART_URL = "/api/v1/chart/"
DASHBOARD_URL = "/api/v1/dashboard/"


class TestThumbnailsSeleniumLive(LiveServerTestCase):
    def create_app(self):
        return app

    def url_open_auth(self, username: str, url: str):
        admin_user = security_manager.find_user(username=username)
        cookies = machine_auth_provider_factory.instance.get_auth_cookies(admin_user)
        opener = urllib.request.build_opener()
        opener.addheaders.append(("Cookie", f"session={cookies['session']}"))
        return opener.open(f"{self.get_server_url()}/{url}")

    @skipUnless((is_feature_enabled("THUMBNAILS")), "Thumbnails feature")
    def test_get_async_dashboard_screenshot(self):
        """
        Thumbnails: Simple get async dashboard screenshot
        """
        with patch("superset.dashboards.api.DashboardRestApi.get") as mock_get:
            rv = self.client.get(DASHBOARD_URL)
            resp = json.loads(rv.data.decode("utf-8"))
            thumbnail_url = resp["result"][0]["thumbnail_url"]

            response = self.url_open_auth(
                "admin",
                thumbnail_url,
            )
            self.assertEqual(response.getcode(), 202)


class TestWebDriverScreenshotErrorDetector(SupersetTestCase):
    @patch("superset.utils.webdriver.WebDriverWait")
    @patch("superset.utils.webdriver.firefox")
    @patch("superset.utils.webdriver.find_unexpected_errors")
    def test_not_call_find_unexpected_errors_if_feature_disabled(
        self, mock_find_unexpected_errors, mock_firefox, mock_webdriver_wait
    ):
        webdriver_proxy = WebDriverProxy("firefox")
        user = security_manager.get_user_by_username(
            app.config["THUMBNAIL_SELENIUM_USER"]
        )
        url = get_url_path("Superset.dashboard", dashboard_id_or_slug=1)
        webdriver_proxy.get_screenshot(url, "grid-container", user=user)

        assert not mock_find_unexpected_errors.called

    @patch("superset.utils.webdriver.WebDriverWait")
    @patch("superset.utils.webdriver.firefox")
    @patch("superset.utils.webdriver.find_unexpected_errors")
    def test_call_find_unexpected_errors_if_feature_enabled(
        self, mock_find_unexpected_errors, mock_firefox, mock_webdriver_wait
    ):
        app.config["SCREENSHOT_REPLACE_UNEXPECTED_ERRORS"] = True
        webdriver_proxy = WebDriverProxy("firefox")
        user = security_manager.get_user_by_username(
            app.config["THUMBNAIL_SELENIUM_USER"]
        )
        url = get_url_path("Superset.dashboard", dashboard_id_or_slug=1)
        webdriver_proxy.get_screenshot(url, "grid-container", user=user)

        assert mock_find_unexpected_errors.called

        app.config["SCREENSHOT_REPLACE_UNEXPECTED_ERRORS"] = False

    def test_find_unexpected_errors_no_alert(self):
        webdriver = MagicMock()

        webdriver.find_elements.return_value = []

        unexpected_errors = find_unexpected_errors(driver=webdriver)
        assert len(unexpected_errors) == 0

        assert "alert" in webdriver.find_elements.call_args_list[0][0][1]

    @patch("superset.utils.webdriver.WebDriverWait")
    def test_find_unexpected_errors(self, mock_webdriver_wait):
        webdriver = MagicMock()
        alert_div = MagicMock()

        webdriver.find_elements.return_value = [alert_div]
        alert_div.find_elements.return_value = MagicMock()

        unexpected_errors = find_unexpected_errors(driver=webdriver)
        assert len(unexpected_errors) == 1

        # attempt to find alerts
        assert "alert" in webdriver.find_elements.call_args_list[0][0][1]
        # attempt to click on "See more" buttons
        assert "button" in alert_div.find_element.call_args_list[0][0][1]
        # Wait for error modal to show up and to hide
        assert 2 == len(mock_webdriver_wait.call_args_list)
        # replace the text in alert div, eg, "unexpected errors"
        assert alert_div == webdriver.execute_script.call_args_list[0][0][1]


class TestWebDriverProxy(SupersetTestCase):
    @patch("superset.utils.webdriver.WebDriverWait")
    @patch("superset.utils.webdriver.firefox")
    @patch("superset.utils.webdriver.sleep")
    def test_screenshot_selenium_headstart(
        self, mock_sleep, mock_webdriver, mock_webdriver_wait
    ):
        webdriver = WebDriverProxy("firefox")
        user = security_manager.get_user_by_username(
            app.config["THUMBNAIL_SELENIUM_USER"]
        )
        url = get_url_path("Superset.slice", slice_id=1, standalone="true")
        app.config["SCREENSHOT_SELENIUM_HEADSTART"] = 5
        webdriver.get_screenshot(url, "chart-container", user=user)
        assert mock_sleep.call_args_list[0] == call(5)

    @patch("superset.utils.webdriver.WebDriverWait")
    @patch("superset.utils.webdriver.firefox")
    def test_screenshot_selenium_locate_wait(self, mock_webdriver, mock_webdriver_wait):
        app.config["SCREENSHOT_LOCATE_WAIT"] = 15
        webdriver = WebDriverProxy("firefox")
        user = security_manager.get_user_by_username(
            app.config["THUMBNAIL_SELENIUM_USER"]
        )
        url = get_url_path("Superset.slice", slice_id=1, standalone="true")
        webdriver.get_screenshot(url, "chart-container", user=user)
        assert mock_webdriver_wait.call_args_list[0] == call(ANY, 15)

    @patch("superset.utils.webdriver.WebDriverWait")
    @patch("superset.utils.webdriver.firefox")
    def test_screenshot_selenium_load_wait(self, mock_webdriver, mock_webdriver_wait):
        app.config["SCREENSHOT_LOAD_WAIT"] = 15
        webdriver = WebDriverProxy("firefox")
        user = security_manager.get_user_by_username(
            app.config["THUMBNAIL_SELENIUM_USER"]
        )
        url = get_url_path("Superset.slice", slice_id=1, standalone="true")
        webdriver.get_screenshot(url, "chart-container", user=user)
        assert mock_webdriver_wait.call_args_list[2] == call(ANY, 15)

    @patch("superset.utils.webdriver.WebDriverWait")
    @patch("superset.utils.webdriver.firefox")
    @patch("superset.utils.webdriver.sleep")
    def test_screenshot_selenium_animation_wait(
        self, mock_sleep, mock_webdriver, mock_webdriver_wait
    ):
        webdriver = WebDriverProxy("firefox")
        user = security_manager.get_user_by_username(
            app.config["THUMBNAIL_SELENIUM_USER"]
        )
        url = get_url_path("Superset.slice", slice_id=1, standalone="true")
        app.config["SCREENSHOT_SELENIUM_ANIMATION_WAIT"] = 4
        webdriver.get_screenshot(url, "chart-container", user=user)
        assert mock_sleep.call_args_list[1] == call(4)


class TestThumbnails(SupersetTestCase):
    mock_image = b"bytes mock image"
    digest_return_value = "foo_bar"
    digest_hash = "5c7d96a3dd7a87850a2ef34087565a6e"

    def _get_id_and_thumbnail_url(self, url: str) -> Tuple[int, str]:
        rv = self.client.get(url)
        resp = json.loads(rv.data.decode("utf-8"))
        obj = resp["result"][0]
        return obj["id"], obj["thumbnail_url"]

    @pytest.mark.usefixtures("load_birth_names_dashboard_with_slices")
    @with_feature_flags(THUMBNAILS=False)
    def test_dashboard_thumbnail_disabled(self):
        """
        Thumbnails: Dashboard thumbnail disabled
        """
        self.login(username="admin")
        _, thumbnail_url = self._get_id_and_thumbnail_url(DASHBOARD_URL)
        rv = self.client.get(thumbnail_url)
        self.assertEqual(rv.status_code, 404)

    @pytest.mark.usefixtures("load_birth_names_dashboard_with_slices")
    @with_feature_flags(THUMBNAILS=False)
    def test_chart_thumbnail_disabled(self):
        """
        Thumbnails: Chart thumbnail disabled
        """
        self.login(username="admin")
        _, thumbnail_url = self._get_id_and_thumbnail_url(CHART_URL)
        rv = self.client.get(thumbnail_url)
        self.assertEqual(rv.status_code, 404)

    @pytest.mark.usefixtures("load_birth_names_dashboard_with_slices")
    @with_feature_flags(THUMBNAILS=True)
    def test_get_async_dashboard_screenshot_as_selenium(self):
        """
        Thumbnails: Simple get async dashboard screenshot as selenium user
        """
        self.login(username="alpha")
        with patch.dict(
            "superset.thumbnails.digest.current_app.config",
            {
                "THUMBNAIL_EXECUTE_AS": [ExecutorType.SELENIUM],
            },
        ), patch(
            "superset.thumbnails.digest._adjust_string_for_executor"
        ) as mock_adjust_string:
            mock_adjust_string.return_value = self.digest_return_value
            _, thumbnail_url = self._get_id_and_thumbnail_url(DASHBOARD_URL)
            assert self.digest_hash in thumbnail_url
            assert mock_adjust_string.call_args[0][1] == ExecutorType.SELENIUM
            assert mock_adjust_string.call_args[0][2] == "admin"

            rv = self.client.get(thumbnail_url)
            self.assertEqual(rv.status_code, 202)

    @pytest.mark.usefixtures("load_birth_names_dashboard_with_slices")
    @with_feature_flags(THUMBNAILS=True)
    def test_get_async_dashboard_screenshot_as_current_user(self):
        """
        Thumbnails: Simple get async dashboard screenshot as current user
        """
        username = "alpha"
        self.login(username=username)
        with patch.dict(
            "superset.thumbnails.digest.current_app.config",
            {
                "THUMBNAIL_EXECUTE_AS": [ExecutorType.CURRENT_USER],
            },
        ), patch(
            "superset.thumbnails.digest._adjust_string_for_executor"
        ) as mock_adjust_string:
            mock_adjust_string.return_value = self.digest_return_value
            _, thumbnail_url = self._get_id_and_thumbnail_url(DASHBOARD_URL)
            assert self.digest_hash in thumbnail_url
            assert mock_adjust_string.call_args[0][1] == ExecutorType.CURRENT_USER
            assert mock_adjust_string.call_args[0][2] == username

            rv = self.client.get(thumbnail_url)
            self.assertEqual(rv.status_code, 202)

    @pytest.mark.usefixtures("load_birth_names_dashboard_with_slices")
    @with_feature_flags(THUMBNAILS=True)
    def test_get_async_dashboard_notfound(self):
        """
        Thumbnails: Simple get async dashboard not found
        """
        max_id = db.session.query(func.max(Dashboard.id)).scalar()
        self.login(username="admin")
        uri = f"api/v1/dashboard/{max_id + 1}/thumbnail/1234/"
        rv = self.client.get(uri)
        self.assertEqual(rv.status_code, 404)

    @pytest.mark.usefixtures("load_birth_names_dashboard_with_slices")
    @skipUnless((is_feature_enabled("THUMBNAILS")), "Thumbnails feature")
    def test_get_async_dashboard_not_allowed(self):
        """
        Thumbnails: Simple get async dashboard not allowed
        """
        self.login(username="gamma")
        _, thumbnail_url = self._get_id_and_thumbnail_url(DASHBOARD_URL)
        rv = self.client.get(thumbnail_url)
        self.assertEqual(rv.status_code, 404)

    @pytest.mark.usefixtures("load_birth_names_dashboard_with_slices")
    @with_feature_flags(THUMBNAILS=True)
    def test_get_async_chart_screenshot_as_selenium(self):
        """
        Thumbnails: Simple get async chart screenshot as selenium user
        """
        self.login(username="alpha")
        with patch.dict(
            "superset.thumbnails.digest.current_app.config",
            {
                "THUMBNAIL_EXECUTE_AS": [ExecutorType.SELENIUM],
            },
        ), patch(
            "superset.thumbnails.digest._adjust_string_for_executor"
        ) as mock_adjust_string:
            mock_adjust_string.return_value = self.digest_return_value
            _, thumbnail_url = self._get_id_and_thumbnail_url(CHART_URL)
            assert self.digest_hash in thumbnail_url
            assert mock_adjust_string.call_args[0][1] == ExecutorType.SELENIUM
            assert mock_adjust_string.call_args[0][2] == "admin"

            rv = self.client.get(thumbnail_url)
            self.assertEqual(rv.status_code, 202)

    @pytest.mark.usefixtures("load_birth_names_dashboard_with_slices")
    @with_feature_flags(THUMBNAILS=True)
    def test_get_async_chart_screenshot_as_current_user(self):
        """
        Thumbnails: Simple get async chart screenshot as current user
        """
        username = "alpha"
        self.login(username=username)
        with patch.dict(
            "superset.thumbnails.digest.current_app.config",
            {
                "THUMBNAIL_EXECUTE_AS": [ExecutorType.CURRENT_USER],
            },
        ), patch(
            "superset.thumbnails.digest._adjust_string_for_executor"
        ) as mock_adjust_string:
            mock_adjust_string.return_value = self.digest_return_value
            _, thumbnail_url = self._get_id_and_thumbnail_url(CHART_URL)
            assert self.digest_hash in thumbnail_url
            assert mock_adjust_string.call_args[0][1] == ExecutorType.CURRENT_USER
            assert mock_adjust_string.call_args[0][2] == username

            rv = self.client.get(thumbnail_url)
            self.assertEqual(rv.status_code, 202)

    @pytest.mark.usefixtures("load_birth_names_dashboard_with_slices")
    @with_feature_flags(THUMBNAILS=True)
    def test_get_async_chart_notfound(self):
        """
        Thumbnails: Simple get async chart not found
        """
        max_id = db.session.query(func.max(Slice.id)).scalar()
        self.login(username="admin")
        uri = f"api/v1/chart/{max_id + 1}/thumbnail/1234/"
        rv = self.client.get(uri)
        self.assertEqual(rv.status_code, 404)

    @pytest.mark.usefixtures("load_birth_names_dashboard_with_slices")
    @with_feature_flags(THUMBNAILS=True)
    def test_get_cached_chart_wrong_digest(self):
        """
        Thumbnails: Simple get chart with wrong digest
        """
        with patch.object(
            ChartScreenshot, "get_from_cache", return_value=BytesIO(self.mock_image)
        ):
            self.login(username="admin")
            id_, thumbnail_url = self._get_id_and_thumbnail_url(CHART_URL)
            rv = self.client.get(f"api/v1/chart/{id_}/thumbnail/1234/")
            self.assertEqual(rv.status_code, 302)
            self.assertEqual(rv.headers["Location"], thumbnail_url)

    @pytest.mark.usefixtures("load_birth_names_dashboard_with_slices")
    @with_feature_flags(THUMBNAILS=True)
    def test_get_cached_dashboard_screenshot(self):
        """
        Thumbnails: Simple get cached dashboard screenshot
        """
        with patch.object(
            DashboardScreenshot, "get_from_cache", return_value=BytesIO(self.mock_image)
        ):
            self.login(username="admin")
            _, thumbnail_url = self._get_id_and_thumbnail_url(DASHBOARD_URL)
            rv = self.client.get(thumbnail_url)
            self.assertEqual(rv.status_code, 200)
            self.assertEqual(rv.data, self.mock_image)

    @pytest.mark.usefixtures("load_birth_names_dashboard_with_slices")
    @with_feature_flags(THUMBNAILS=True)
    def test_get_cached_chart_screenshot(self):
        """
        Thumbnails: Simple get cached chart screenshot
        """
        with patch.object(
            ChartScreenshot, "get_from_cache", return_value=BytesIO(self.mock_image)
        ):
            self.login(username="admin")
            id_, thumbnail_url = self._get_id_and_thumbnail_url(CHART_URL)
            rv = self.client.get(thumbnail_url)
            self.assertEqual(rv.status_code, 200)
            self.assertEqual(rv.data, self.mock_image)

    @pytest.mark.usefixtures("load_birth_names_dashboard_with_slices")
    @with_feature_flags(THUMBNAILS=True)
    def test_get_cached_dashboard_wrong_digest(self):
        """
        Thumbnails: Simple get dashboard with wrong digest
        """
        with patch.object(
            DashboardScreenshot, "get_from_cache", return_value=BytesIO(self.mock_image)
        ):
            self.login(username="admin")
            id_, thumbnail_url = self._get_id_and_thumbnail_url(DASHBOARD_URL)
            rv = self.client.get(f"api/v1/dashboard/{id_}/thumbnail/1234/")
            self.assertEqual(rv.status_code, 302)
            self.assertEqual(rv.headers["Location"], thumbnail_url)