# 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
from datetime import date, datetime
import os
import re
from typing import Any, Optional
from unittest.mock import Mock, patch  # noqa: F401

from superset.commands.database.exceptions import DatabaseInvalidError
from tests.integration_tests.fixtures.birth_names_dashboard import (
    load_birth_names_dashboard_with_slices,  # noqa: F401
    load_birth_names_data,  # noqa: F401
)

from flask import current_app, Flask, g  # noqa: F401
import pandas as pd
import pytest
import marshmallow
from sqlalchemy.exc import ArgumentError  # noqa: F401

import tests.integration_tests.test_app  # noqa: F401
from superset import db, security_manager
from superset.constants import NO_TIME_RANGE
from superset.exceptions import CertificateException, SupersetException  # noqa: F401
from superset.models.core import Database, Log
from superset.models.dashboard import Dashboard  # noqa: F401
from superset.models.slice import Slice  # noqa: F401
from superset.utils.core import (
    cast_to_num,
    convert_legacy_filters_into_adhoc,
    create_ssl_cert_file,
    DTTM_ALIAS,
    extract_dataframe_dtypes,
    GenericDataType,
    get_form_data_token,
    as_list,
    recipients_string_to_list,
    merge_extra_filters,
    merge_extra_form_data,
    normalize_dttm_col,
    parse_ssl_cert,
    split,
    DateColumn,
)
from superset.utils import json
from superset.utils.database import get_or_create_db
from superset.utils import schema
from superset.utils.hashing import md5_sha_from_str
from superset.views.utils import build_extra_filters, get_form_data  # noqa: F401
from tests.integration_tests.base_tests import SupersetTestCase
from tests.integration_tests.constants import ADMIN_USERNAME
from tests.integration_tests.fixtures.world_bank_dashboard import (
    load_world_bank_dashboard_with_slices,  # noqa: F401
    load_world_bank_data,  # noqa: F401
)

from .fixtures.certificates import ssl_certificate


class TestUtils(SupersetTestCase):
    def test_convert_legacy_filters_into_adhoc_where(self):
        form_data = {"where": "a = 1"}
        expected = {
            "adhoc_filters": [
                {
                    "clause": "WHERE",
                    "expressionType": "SQL",
                    "filterOptionName": "46fb6d7891e23596e42ae38da94a57e0",
                    "sqlExpression": "a = 1",
                }
            ]
        }
        convert_legacy_filters_into_adhoc(form_data)
        assert form_data == expected

    def test_convert_legacy_filters_into_adhoc_filters(self):
        form_data = {"filters": [{"col": "a", "op": "in", "val": "someval"}]}
        expected = {
            "adhoc_filters": [
                {
                    "clause": "WHERE",
                    "comparator": "someval",
                    "expressionType": "SIMPLE",
                    "filterOptionName": "135c7ee246666b840a3d7a9c3a30cf38",
                    "operator": "in",
                    "subject": "a",
                }
            ]
        }
        convert_legacy_filters_into_adhoc(form_data)
        assert form_data == expected

    def test_convert_legacy_filters_into_adhoc_present_and_empty(self):
        form_data = {"adhoc_filters": [], "where": "a = 1"}
        expected = {
            "adhoc_filters": [
                {
                    "clause": "WHERE",
                    "expressionType": "SQL",
                    "filterOptionName": "46fb6d7891e23596e42ae38da94a57e0",
                    "sqlExpression": "a = 1",
                }
            ]
        }
        convert_legacy_filters_into_adhoc(form_data)
        assert form_data == expected

    def test_convert_legacy_filters_into_adhoc_having(self):
        form_data = {"having": "COUNT(1) = 1"}
        expected = {
            "adhoc_filters": [
                {
                    "clause": "HAVING",
                    "expressionType": "SQL",
                    "filterOptionName": "683f1c26466ab912f75a00842e0f2f7b",
                    "sqlExpression": "COUNT(1) = 1",
                }
            ]
        }
        convert_legacy_filters_into_adhoc(form_data)
        assert form_data == expected

    def test_convert_legacy_filters_into_adhoc_present_and_nonempty(self):
        form_data = {
            "adhoc_filters": [
                {"clause": "WHERE", "expressionType": "SQL", "sqlExpression": "a = 1"}
            ],
            "filters": [{"col": "a", "op": "in", "val": "someval"}],
            "having": "COUNT(1) = 1",
        }
        expected = {
            "adhoc_filters": [
                {"clause": "WHERE", "expressionType": "SQL", "sqlExpression": "a = 1"}
            ]
        }
        convert_legacy_filters_into_adhoc(form_data)
        assert form_data == expected

    def test_split(self):
        assert list(split("a b")) == ["a", "b"]
        assert list(split("a,b", delimiter=",")) == ["a", "b"]
        assert list(split("a,(b,a)", delimiter=",")) == ["a", "(b,a)"]
        assert list(split('a,(b,a),"foo , bar"', delimiter=",")) == [
            "a",
            "(b,a)",
            '"foo , bar"',
        ]
        assert list(split("a,'b,c'", delimiter=",", quote="'")) == ["a", "'b,c'"]
        assert list(split('a "b c"')) == ["a", '"b c"']
        assert list(split('a "b \\" c"')) == ["a", '"b \\" c"']

    def test_get_or_create_db(self):
        get_or_create_db("test_db", "sqlite:///superset.db")
        database = db.session.query(Database).filter_by(database_name="test_db").one()
        assert database is not None
        assert database.sqlalchemy_uri == "sqlite:///superset.db"
        assert (
            security_manager.find_permission_view_menu("database_access", database.perm)
            is not None
        )
        # Test change URI
        get_or_create_db("test_db", "sqlite:///changed.db")
        database = db.session.query(Database).filter_by(database_name="test_db").one()
        assert database.sqlalchemy_uri == "sqlite:///changed.db"
        db.session.delete(database)
        db.session.commit()

    def test_get_or_create_db_invalid_uri(self):
        with self.assertRaises(DatabaseInvalidError):  # noqa: PT027
            get_or_create_db("test_db", "yoursql:superset.db/()")

    def test_get_or_create_db_existing_invalid_uri(self):
        database = get_or_create_db("test_db", "sqlite:///superset.db")
        database.sqlalchemy_uri = "None"
        db.session.commit()
        database = get_or_create_db("test_db", "sqlite:///superset.db")
        assert database.sqlalchemy_uri == "sqlite:///superset.db"

    def test_as_list(self):
        self.assertListEqual(as_list(123), [123])  # noqa: PT009
        self.assertListEqual(as_list([123]), [123])  # noqa: PT009
        self.assertListEqual(as_list("foo"), ["foo"])  # noqa: PT009

    def test_merge_extra_filters_with_no_extras(self):
        form_data = {
            "time_range": "Last 10 days",
        }
        merge_extra_form_data(form_data)
        assert form_data == {"time_range": "Last 10 days", "adhoc_filters": []}

    def test_merge_extra_filters_with_unset_legacy_time_range(self):
        """
        Make sure native filter is applied if filter time range is unset.
        """
        form_data = {
            "time_range": "Last 10 days",
            "extra_filters": [
                {"col": "__time_range", "op": "==", "val": NO_TIME_RANGE},
            ],
            "extra_form_data": {"time_range": "Last year"},
        }
        merge_extra_filters(form_data)
        assert form_data == {
            "time_range": "Last year",
            "applied_time_extras": {},
            "adhoc_filters": [],
        }

    def test_merge_extra_filters_with_extras(self):
        form_data = {
            "time_range": "Last 10 days",
            "extra_form_data": {
                "filters": [{"col": "foo", "op": "IN", "val": ["bar"]}],
                "adhoc_filters": [
                    {
                        "expressionType": "SQL",
                        "clause": "WHERE",
                        "sqlExpression": "1 = 0",
                    }
                ],
                "time_range": "Last 100 years",
                "time_grain_sqla": "PT1M",
                "relative_start": "now",
            },
        }
        merge_extra_form_data(form_data)
        adhoc_filters = form_data["adhoc_filters"]
        assert adhoc_filters[0] == {
            "clause": "WHERE",
            "expressionType": "SQL",
            "isExtra": True,
            "sqlExpression": "1 = 0",
        }
        converted_filter = adhoc_filters[1]
        del converted_filter["filterOptionName"]
        assert converted_filter == {
            "clause": "WHERE",
            "comparator": ["bar"],
            "expressionType": "SIMPLE",
            "isExtra": True,
            "operator": "IN",
            "subject": "foo",
        }
        assert form_data["time_range"] == "Last 100 years"
        assert form_data["time_grain_sqla"] == "PT1M"
        assert form_data["extras"]["relative_start"] == "now"

    def test_ssl_certificate_parse(self):
        parsed_certificate = parse_ssl_cert(ssl_certificate)
        assert parsed_certificate.serial_number == 12355228710836649848

    def test_ssl_certificate_file_creation(self):
        path = create_ssl_cert_file(ssl_certificate)
        expected_filename = md5_sha_from_str(ssl_certificate)
        assert expected_filename in path
        assert os.path.exists(path)

    def test_recipients_string_to_list(self):
        assert recipients_string_to_list("a@a") == ["a@a"]
        assert recipients_string_to_list(" a@a ") == ["a@a"]
        assert recipients_string_to_list("a@a\n") == ["a@a"]
        assert recipients_string_to_list(",a@a;") == ["a@a"]
        assert recipients_string_to_list(",a@a; b@b c@c a-c@c; d@d, f@f") == [
            "a@a",
            "b@b",
            "c@c",
            "a-c@c",
            "d@d",
            "f@f",
        ]

    def test_get_form_data_default(self) -> None:
        form_data, slc = get_form_data()
        assert slc is None

    def test_get_form_data_request_args(self) -> None:
        with current_app.test_request_context(
            query_string={"form_data": json.dumps({"foo": "bar"})}
        ):
            form_data, slc = get_form_data()
            assert form_data == {"foo": "bar"}
            assert slc is None

    def test_get_form_data_request_form(self) -> None:
        with current_app.test_request_context(
            data={"form_data": json.dumps({"foo": "bar"})}
        ):
            form_data, slc = get_form_data()
            assert form_data == {"foo": "bar"}
            assert slc is None

    def test_get_form_data_request_form_with_queries(self) -> None:
        # the CSV export uses for requests, even when sending requests to
        # /api/v1/chart/data
        with current_app.test_request_context(
            data={
                "form_data": json.dumps({"queries": [{"url_params": {"foo": "bar"}}]})
            }
        ):
            form_data, slc = get_form_data()
            assert form_data == {"url_params": {"foo": "bar"}}
            assert slc is None

    def test_get_form_data_request_args_and_form(self) -> None:
        with current_app.test_request_context(
            data={"form_data": json.dumps({"foo": "bar"})},
            query_string={"form_data": json.dumps({"baz": "bar"})},
        ):
            form_data, slc = get_form_data()
            assert form_data == {"baz": "bar", "foo": "bar"}
            assert slc is None

    def test_get_form_data_globals(self) -> None:
        with current_app.test_request_context():
            g.form_data = {"foo": "bar"}
            form_data, slc = get_form_data()
            delattr(g, "form_data")
            assert form_data == {"foo": "bar"}
            assert slc is None

    def test_get_form_data_corrupted_json(self) -> None:
        with current_app.test_request_context(
            data={"form_data": "{x: '2324'}"},
            query_string={"form_data": '{"baz": "bar"'},
        ):
            form_data, slc = get_form_data()
            assert form_data == {}
            assert slc is None

    @pytest.mark.usefixtures("load_world_bank_dashboard_with_slices")
    def test_log_this(self) -> None:
        # TODO: Add additional scenarios.
        self.login(ADMIN_USERNAME)
        slc = self.get_slice("Life Expectancy VS Rural %")
        dashboard_id = 1

        assert slc.viz is not None
        resp = self.get_json_resp(  # noqa: F841
            f"/superset/explore_json/{slc.datasource_type}/{slc.datasource_id}/"
            + f'?form_data={{"slice_id": {slc.id}}}&dashboard_id={dashboard_id}',
            {"form_data": json.dumps(slc.viz.form_data)},
        )

        record = (
            db.session.query(Log)
            .filter_by(action="explore_json", slice_id=slc.id)
            .order_by(Log.dttm.desc())
            .first()
        )

        assert record.dashboard_id == dashboard_id
        assert json.loads(record.json)["dashboard_id"] == str(dashboard_id)
        assert json.loads(record.json)["form_data"]["slice_id"] == slc.id

        assert (
            json.loads(record.json)["form_data"]["viz_type"]
            == slc.viz.form_data["viz_type"]
        )

    def test_schema_validate_json(self):
        valid = '{"a": 5, "b": [1, 5, ["g", "h"]]}'
        assert schema.validate_json(valid) is None
        invalid = '{"a": 5, "b": [1, 5, ["g", "h]]}'
        self.assertRaises(marshmallow.ValidationError, schema.validate_json, invalid)  # noqa: PT027

    def test_schema_one_of_case_insensitive(self):
        validator = schema.OneOfCaseInsensitive(choices=[1, 2, 3, "FoO", "BAR", "baz"])
        assert 1 == validator(1)
        assert 2 == validator(2)
        assert "FoO" == validator("FoO")
        assert "FOO" == validator("FOO")
        assert "bar" == validator("bar")
        assert "BaZ" == validator("BaZ")
        self.assertRaises(marshmallow.ValidationError, validator, "qwerty")  # noqa: PT027
        self.assertRaises(marshmallow.ValidationError, validator, 4)  # noqa: PT027

    def test_cast_to_num(self) -> None:
        assert cast_to_num("5") == 5
        assert cast_to_num("5.2") == 5.2
        assert cast_to_num(10) == 10
        assert cast_to_num(10.1) == 10.1
        assert cast_to_num(None) is None
        assert cast_to_num("this is not a string") is None

    def test_get_form_data_token(self):
        assert get_form_data_token({"token": "token_abcdefg1"}) == "token_abcdefg1"
        generated_token = get_form_data_token({})
        assert re.match(r"^token_[a-z0-9]{8}$", generated_token) is not None

    @pytest.mark.usefixtures("load_birth_names_dashboard_with_slices")
    def test_extract_dataframe_dtypes(self):
        slc = self.get_slice("Girls")
        cols: tuple[tuple[str, GenericDataType, list[Any]], ...] = (
            ("dt", GenericDataType.TEMPORAL, [date(2021, 2, 4), date(2021, 2, 4)]),
            (
                "dttm",
                GenericDataType.TEMPORAL,
                [datetime(2021, 2, 4, 1, 1, 1), datetime(2021, 2, 4, 1, 1, 1)],
            ),
            ("str", GenericDataType.STRING, ["foo", "foo"]),
            ("int", GenericDataType.NUMERIC, [1, 1]),
            ("float", GenericDataType.NUMERIC, [0.5, 0.5]),
            ("mixed-int-float", GenericDataType.NUMERIC, [0.5, 1.0]),
            ("bool", GenericDataType.BOOLEAN, [True, False]),
            ("mixed-str-int", GenericDataType.STRING, ["abc", 1.0]),
            ("obj", GenericDataType.STRING, [{"a": 1}, {"a": 1}]),
            ("dt_null", GenericDataType.TEMPORAL, [None, date(2021, 2, 4)]),
            (
                "dttm_null",
                GenericDataType.TEMPORAL,
                [None, datetime(2021, 2, 4, 1, 1, 1)],
            ),
            ("str_null", GenericDataType.STRING, [None, "foo"]),
            ("int_null", GenericDataType.NUMERIC, [None, 1]),
            ("float_null", GenericDataType.NUMERIC, [None, 0.5]),
            ("bool_null", GenericDataType.BOOLEAN, [None, False]),
            ("obj_null", GenericDataType.STRING, [None, {"a": 1}]),
            # Non-timestamp columns should be identified as temporal if
            # `is_dttm` is set to `True` in the underlying datasource
            ("ds", GenericDataType.TEMPORAL, [None, {"ds": "2017-01-01"}]),
        )

        df = pd.DataFrame(data={col[0]: col[2] for col in cols})
        assert extract_dataframe_dtypes(df, slc.datasource) == [col[1] for col in cols]

    def test_normalize_dttm_col(self):
        def normalize_col(
            df: pd.DataFrame,
            timestamp_format: Optional[str],
            offset: int,
            time_shift: Optional[str],
        ) -> pd.DataFrame:
            df = df.copy()
            normalize_dttm_col(
                df,
                tuple(  # noqa: C409
                    [
                        DateColumn.get_legacy_time_column(
                            timestamp_format=timestamp_format,
                            offset=offset,
                            time_shift=time_shift,
                        )
                    ]
                ),
            )
            return df

        ts = pd.Timestamp(2021, 2, 15, 19, 0, 0, 0)
        df = pd.DataFrame([{"__timestamp": ts, "a": 1}])

        # test regular (non-numeric) format
        assert normalize_col(df, None, 0, None)[DTTM_ALIAS][0] == ts
        assert normalize_col(df, "epoch_ms", 0, None)[DTTM_ALIAS][0] == ts
        assert normalize_col(df, "epoch_s", 0, None)[DTTM_ALIAS][0] == ts

        # test offset
        assert normalize_col(df, None, 1, None)[DTTM_ALIAS][0] == pd.Timestamp(
            2021, 2, 15, 20, 0, 0, 0
        )

        # test offset and timedelta
        assert normalize_col(df, None, 1, "30 minutes")[DTTM_ALIAS][0] == pd.Timestamp(
            2021, 2, 15, 20, 30, 0, 0
        )

        # test numeric epoch_s format
        df = pd.DataFrame([{"__timestamp": ts.timestamp(), "a": 1}])
        assert normalize_col(df, "epoch_s", 0, None)[DTTM_ALIAS][0] == ts

        # test numeric epoch_ms format
        df = pd.DataFrame([{"__timestamp": ts.timestamp() * 1000, "a": 1}])
        assert normalize_col(df, "epoch_ms", 0, None)[DTTM_ALIAS][0] == ts