# 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.

"""Test utils.path_dict module."""

from typing import Dict

import pytest
from ruamel.yaml import YAML

from pydolphinscheduler.utils.yaml_parser import YamlParser
from tests.testing.path import path_default_config_yaml

yaml = YAML()

expects = [
    {
        # yaml.load("no need test") is a flag about skipping it because it to different to maintainer
        "name": yaml.load("no need test"),
        "name.family": ("Smith", "SmithEdit"),
        "name.given": ("Alice", "AliceEdit"),
        "name.mark": yaml.load("no need test"),
        "name.mark.name_mark": yaml.load("no need test"),
        "name.mark.name_mark.key": ("value", "valueEdit"),
    },
    {
        # yaml.load("no need test") is a flag about skipping it because it to different to maintainer
        "java_gateway": yaml.load("no need test"),
        "java_gateway.address": ("127.0.0.1", "127.1.1.1"),
        "java_gateway.port": (25333, 25555),
        "java_gateway.auto_convert": (True, False),
        "default": yaml.load("no need test"),
        "default.user": yaml.load("no need test"),
        "default.user.name": ("userPythonGateway", "userPythonGatewayEdit"),
        "default.user.password": ("userPythonGateway", "userPythonGatewayEdit"),
        "default.user.email": (
            "userPythonGateway@dolphinscheduler.com",
            "userEdit@dolphinscheduler.com",
        ),
        "default.user.tenant": ("tenant_pydolphin", "tenant_pydolphinEdit"),
        "default.user.phone": (11111111111, 22222222222),
        "default.user.state": (1, 0),
        "default.workflow": yaml.load("no need test"),
        "default.workflow.project": ("project-pydolphin", "project-pydolphinEdit"),
        "default.workflow.tenant": ("tenant_pydolphin", "SmithEdit"),
        "default.workflow.user": ("userPythonGateway", "SmithEdit"),
        "default.workflow.queue": ("queuePythonGateway", "SmithEdit"),
        "default.workflow.worker_group": ("default", "SmithEdit"),
        "default.workflow.time_zone": ("Asia/Shanghai", "SmithEdit"),
    },
]

param = [
    """#example
name:
  # details
  family: Smith   # very common
  given: Alice    # one of the siblings
  mark:
    name_mark:
      key: value
"""
]

with open(path_default_config_yaml, "r") as f:
    param.append(f.read())


@pytest.mark.parametrize(
    "src, delimiter, expect",
    [
        (
            param[0],
            "|",
            expects[0],
        ),
        (
            param[1],
            "/",
            expects[1],
        ),
    ],
)
def test_yaml_parser_specific_delimiter(src: str, delimiter: str, expect: Dict):
    """Test specific delimiter for :class:`YamlParser`."""

    def ch_dl(key):
        return key.replace(".", delimiter)

    yaml_parser = YamlParser(src, delimiter=delimiter)
    assert all(
        [
            expect[key][0] == yaml_parser[ch_dl(key)]
            for key in expect
            if expect[key] != "no need test"
        ]
    )
    assert all(
        [
            expect[key][0] == yaml_parser.get(ch_dl(key))
            for key in expect
            if expect[key] != "no need test"
        ]
    )


@pytest.mark.parametrize(
    "src, expect",
    [
        (
            param[0],
            expects[0],
        ),
        (
            param[1],
            expects[1],
        ),
    ],
)
def test_yaml_parser_contains(src: str, expect: Dict):
    """Test magic function :func:`YamlParser.__contain__` also with `key in obj` syntax."""
    yaml_parser = YamlParser(src)
    assert len(expect.keys()) == len(
        yaml_parser.dict_parser.keys()
    ), "Parser keys length not equal to expect keys length"
    assert all(
        [key in yaml_parser for key in expect]
    ), "Parser keys not equal to expect keys"


@pytest.mark.parametrize(
    "src, expect",
    [
        (
            param[0],
            expects[0],
        ),
        (
            param[1],
            expects[1],
        ),
    ],
)
def test_yaml_parser_get(src: str, expect: Dict):
    """Test magic function :func:`YamlParser.__getitem__` also with `obj[key]` syntax."""
    yaml_parser = YamlParser(src)
    assert all(
        [
            expect[key][0] == yaml_parser[key]
            for key in expect
            if expect[key] != "no need test"
        ]
    )
    assert all(
        [
            expect[key][0] == yaml_parser.get(key)
            for key in expect
            if expect[key] != "no need test"
        ]
    )


@pytest.mark.parametrize(
    "src, expect",
    [
        (
            param[0],
            expects[0],
        ),
        (
            param[1],
            expects[1],
        ),
    ],
)
def test_yaml_parser_set(src: str, expect: Dict):
    """Test magic function :func:`YamlParser.__setitem__` also with `obj[key] = val` syntax."""
    yaml_parser = YamlParser(src)
    for key in expect:
        assert key in yaml_parser.dict_parser.keys()
        if expect[key] == "no need test":
            continue
        assert expect[key][0] == yaml_parser.dict_parser[key]
        assert expect[key][1] != yaml_parser.dict_parser[key]

        yaml_parser[key] = expect[key][1]
        assert expect[key][0] != yaml_parser.dict_parser[key]
        assert expect[key][1] == yaml_parser.dict_parser[key]


@pytest.mark.parametrize(
    "src, setter, expect",
    [
        (
            param[0],
            {"name.mark.name_mark.key": "edit"},
            """#example
name:
  # details
  family: Smith   # very common
  given: Alice    # one of the siblings
  mark:
    name_mark:
      key: edit
""",
        ),
        (
            param[0],
            {
                "name.family": "SmithEdit",
                "name.given": "AliceEdit",
                "name.mark.name_mark.key": "edit",
            },
            """#example
name:
  # details
  family: SmithEdit # very common
  given: AliceEdit # one of the siblings
  mark:
    name_mark:
      key: edit
""",
        ),
    ],
)
def test_yaml_parser_str_repr(src: str, setter: Dict, expect: str):
    """Test function :func:`YamlParser.to_string`."""
    yaml_parser = YamlParser(src)

    # Equal before change
    assert f"YamlParser({src})" == repr(yaml_parser)
    assert src == str(yaml_parser)

    for key, val in setter.items():
        yaml_parser[key] = val

    # Equal after changed
    assert expect == str(yaml_parser)
    assert f"YamlParser({expect})" == repr(yaml_parser)


@pytest.mark.parametrize(
    "src, key, expect",
    [
        (param[1], "java_gateway.port", 25333),
        (param[1], "default.user.phone", 11111111111),
        (param[1], "default.user.state", 1),
    ],
)
def test_yaml_parser_get_int(src: str, key: str, expect: int):
    """Test function :func:`YamlParser.get_int`."""
    yaml_parser = YamlParser(src)
    assert expect == yaml_parser.get_int(key)


@pytest.mark.parametrize(
    "src, key, expect",
    [
        (param[1], "java_gateway.auto_convert", True),
    ],
)
def test_yaml_parser_get_bool(src: str, key: str, expect: bool):
    """Test function :func:`YamlParser.get_bool`."""
    yaml_parser = YamlParser(src)
    assert expect == yaml_parser.get_bool(key)