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

import ast
import importlib
from unittest.mock import patch

import pytest

from tests.testing.constants import task_without_example
from tests.testing.path import get_all_examples, get_tasks
from tests.testing.task import Task

workflow_name = set()


def import_module(script_name, script_path):
    """Import and run example module in examples directory."""
    spec = importlib.util.spec_from_file_location(script_name, script_path)
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    return module


def test_task_without_example():
    """Test task which without example.

    Avoiding add new type of tasks but without adding example describe how to use it.
    """
    # We use example/tutorial.py as shell task example
    # Task sub_process is deprecated and will be removed in the future, so we set it to ignore
    ignore_name = {"__init__.py", "shell.py", "func_wrap.py", "sub_process.py"}
    all_tasks = {task.stem for task in get_tasks(ignore_name=ignore_name)}

    have_example_tasks = set()
    start = "task_"
    end = "_example"
    for ex in get_all_examples():
        stem = ex.stem
        if stem.startswith(start) and stem.endswith(end):
            task_name = stem.replace(start, "").replace(end, "")
            have_example_tasks.add(task_name)

    assert all_tasks.difference(have_example_tasks) == task_without_example


@pytest.fixture
def setup_and_teardown_for_stuff():
    """Fixture of py.test handle setup and teardown."""
    yield
    global workflow_name
    workflow_name = set()


def submit_check_without_same_name(self):
    """Side effect for verifying workflow name and adding it to global variable."""
    if self.name in workflow_name:
        raise ValueError(
            "Example workflow should not have same name, but get duplicate name: %s",
            self.name,
        )
    submit_add_workflow(self)


def submit_add_workflow(self):
    """Side effect for adding workflow name to global variable."""
    workflow_name.add(self.name)


def test_example_basic():
    """Test example basic information.

    Which including:
    * File extension name is `.py`
    * All example except `tutorial.py` is end with keyword "_example"
    * All example must have not empty `__doc__`.
    """
    for ex in get_all_examples():
        # All files in example is python script
        assert (
            ex.suffix == ".py"
        ), f"We expect all examples is python script, but get {ex.name}."

        # All except tutorial and __init__ is end with keyword "_example"
        if (
            ex.stem
            not in ("tutorial", "tutorial_decorator", "tutorial_resource_plugin")
            and ex.stem != "__init__"
        ):
            assert ex.stem.endswith(
                "_example"
            ), f"We expect all examples script end with keyword '_example', but get {ex.stem}."

        # All files have __doc__
        tree = ast.parse(ex.read_text())
        example_doc = ast.get_docstring(tree, clean=False)
        assert (
            example_doc is not None
        ), f"We expect all examples have __doc__, but {ex.name} do not."


@patch("pydolphinscheduler.core.workflow.Workflow.start")
@patch(
    "pydolphinscheduler.core.workflow.Workflow.submit",
    side_effect=submit_check_without_same_name,
    autospec=True,
)
@patch(
    "pydolphinscheduler.core.task.Task.gen_code_and_version",
    # Example bulk_create_example.py would create workflow dynamic by :func:`get_one_task_by_name`
    # and would raise error in :func:`get_one_task_by_name` if we return constant value
    # using :arg:`return_value`
    side_effect=Task("test_example", "test_example").gen_code_and_version,
)
def test_example_workflow_without_same_name(mock_code_version, mock_submit, mock_start):
    """Test all examples file without same workflow's name.

    Our workflow would compete with others if we have same workflow name. It will make
    different between actually workflow and our workflow-as-code file which make users feel strange.
    """
    for ex in get_all_examples():
        # We use side_effect `submit_check_without_same_name` overwrite :func:`submit`
        # and check whether it have duplicate name or not
        import_module(ex.name, str(ex))
    assert True


@patch("pydolphinscheduler.core.workflow.Workflow.start")
@patch(
    "pydolphinscheduler.core.workflow.Workflow.submit",
    side_effect=submit_add_workflow,
    autospec=True,
)
@patch(
    "pydolphinscheduler.core.task.Task.gen_code_and_version",
    # Example bulk_create_example.py would create workflow dynamic by :func:`get_one_task_by_name`
    # and would raise error in :func:`get_one_task_by_name` if we return constant value
    # using :arg:`return_value`
    side_effect=Task("test_example", "test_example").gen_code_and_version,
)
def test_file_name_in_workflow(mock_code_version, mock_submit, mock_start):
    """Test example file name in example definition name.

    We should not directly assert equal, because some of the examples contain
    more than one workflow.
    """
    global workflow_name
    for ex in get_all_examples():
        # Skip __init__ file
        if ex.stem == "__init__":
            continue
        # Skip bulk_create_example check, cause it contain multiple workflow and
        # without one named bulk_create_example
        if ex.stem == "bulk_create_example":
            continue
        workflow_name = set()
        assert ex.stem not in workflow_name
        import_module(ex.name, str(ex))
        assert ex.stem in workflow_name