[python] Add test for examples (#7759)

* Make sure all file with `.py` extension and all file
  end with "_example"
* Make sure all examples have `__doc__`
* Make sure not have duplicate process definition name
* Make sure process definition same as filename

close: #7729

dolphinscheduler-python/pydolphinscheduler/examples/task_datax_example.py

@@ -32,7 +32,7 @@ from pydolphinscheduler.tasks.datax import CustomDataX, DataX
 JSON_TEMPLATE = ""
 
 with ProcessDefinition(
-    name="task_datax",
+    name="task_datax_example",
     tenant="tenant_exists",
 ) as pd:
     # This task synchronizes the data in `t_ds_project`

dolphinscheduler-python/pydolphinscheduler/examples/task_dependent_example.py

@@ -49,7 +49,7 @@ with ProcessDefinition(
     pd.submit()
 
 with ProcessDefinition(
-    name="task_dependent",
+    name="task_dependent_example",
     tenant="tenant_exists",
 ) as pd:
     task = Dependent(

dolphinscheduler-python/pydolphinscheduler/examples/task_switch_example.py

@@ -34,7 +34,7 @@ from pydolphinscheduler.tasks.shell import Shell
 from pydolphinscheduler.tasks.switch import Branch, Default, Switch, SwitchCondition
 
 with ProcessDefinition(
-    name="task_dependent_external",
+    name="task_switch_example",
     tenant="tenant_exists",
 ) as pd:
     parent = Shell(name="parent", command="echo parent")

dolphinscheduler-python/pydolphinscheduler/tests/example/__init__.py

@@ -0,0 +1,18 @@
+# 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.
+
+"""Init example package tests."""

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

@@ -0,0 +1,159 @@
+# 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