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