DataCenterCDC-DS/dolphinscheduler-python/pydolphinscheduler/tests/example/test_example.py

# 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

# import os
# import os
from pathlib import Path
from unittest.mock import patch

import pytest

from tests.testing.task import Task

process_definition_name = set()


def get_all_example_define():
    """Get all examples files in examples directory."""
    return (
        path
        for path in Path(__file__).parent.parent.parent.joinpath("examples").iterdir()
        if path.is_file()
    )


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


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


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


def submit_add_process_definition(self):
    """Side effect for adding process definition name to global variable."""
    process_definition_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_example_define():
        # 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 is end with keyword "_example"
        if ex.stem != "tutorial":
            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.process_definition.ProcessDefinition.start")
@patch(
    "pydolphinscheduler.core.process_definition.ProcessDefinition.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_process_definition_without_same_name(
    mock_code_version, mock_submit, mock_start
):
    """Test all examples file without same process definition's name.

    Our process definition would compete with others if we have same process definition name. It will make
    different between actually workflow and our workflow-as-code file which make users feel strange.
    """
    for ex in get_all_example_define():
        # 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.process_definition.ProcessDefinition.start")
@patch(
    "pydolphinscheduler.core.process_definition.ProcessDefinition.submit",
    side_effect=submit_add_process_definition,
    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_process_definition(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 process definition.
    """
    global process_definition_name
    for ex in get_all_example_define():
        # Skip bulk_create_example check, cause it contain multiple workflow and
        # without one named bulk_create_example
        if ex.stem == "bulk_create_example":
            continue
        process_definition_name = set()
        assert ex.stem not in process_definition_name
        import_module(ex.name, str(ex))
        assert ex.stem in process_definition_name