[python] refactor yaml file parser (#8701)

* [python] refactor yaml file parser

* Change yaml parser package to ruamel.yaml
* Refactor configuration.py module
* And file.py to write file locally
* Add more tests on it

close: #8593

* Fix UT error

* Remove pypyaml from tests

* Fix file error when param create is False

* Fix error logic
* And tests to avoid regression

dolphinscheduler-python/pydolphinscheduler/UPDATING.md

@@ -0,0 +1,27 @@
+<!--
+ 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.
+-->
+
+# UPDATING
+
+Updating is try to document non-backward compatible updates which notice users the detail changes about pydolphinscheduler.
+It started after version 2.0.5 released
+
+## dev
+
+* Use package ``ruamel.yaml`` replace ``pyyaml`` for write yaml file with comment.

dolphinscheduler-python/pydolphinscheduler/setup.py

@@ -38,7 +38,7 @@ version = "2.0.4"
 prod = [
     "click>=8.0.0",
     "py4j~=0.10",
-    "pyyaml",
+    "ruamel.yaml",
 ]
 
 build = [

dolphinscheduler-python/pydolphinscheduler/src/pydolphinscheduler/core/configuration.py

@@ -16,137 +16,133 @@
 # under the License.
 
 """Configuration module for pydolphinscheduler."""
-
-import copy
 import os
 from pathlib import Path
-from typing import Any, Dict
+from typing import Any
-
-import yaml
 
-from pydolphinscheduler.exceptions import PyDSConfException, PyDSParamException
+from pydolphinscheduler.exceptions import PyDSConfException
-from pydolphinscheduler.utils.path_dict import PathDict
+from pydolphinscheduler.utils import file
+from pydolphinscheduler.utils.yaml_parser import YamlParser
 
-DEFAULT_CONFIG_PATH = Path(__file__).resolve().parent.joinpath("default_config.yaml")
+BUILD_IN_CONFIG_PATH = Path(__file__).resolve().parent.joinpath("default_config.yaml")
 
 
-def get_config_file_path() -> Path:
+def config_path() -> Path:
     """Get the path of pydolphinscheduler configuration file."""
     pyds_home = os.environ.get("PYDOLPHINSCHEDULER_HOME", "~/pydolphinscheduler")
     config_file_path = Path(pyds_home).joinpath("config.yaml").expanduser()
     return config_file_path
 
 
-def read_yaml(path: str) -> Dict:
+def get_configs() -> YamlParser:
-    """Read configs dict from configuration file.
+    """Get all configuration settings from configuration file.
 
-    :param path: The path of configuration file.
+    Will use custom configuration file first if it exists, otherwise default configuration file in
+    default path.
     """
-    with open(path, "r") as f:
+    path = str(config_path()) if config_path().exists() else BUILD_IN_CONFIG_PATH
-        return yaml.safe_load(f)
+    with open(path, mode="r") as f:
-
+        return YamlParser(f.read())
-
-def write_yaml(context: Dict, path: str) -> None:
-    """Write configs dict to configuration file.
-
-    :param context: The configs dict write to configuration file.
-    :param path: The path of configuration file.
-    """
-    parent = Path(path).parent
-    if not parent.exists():
-        parent.mkdir(parents=True)
-    with open(path, mode="w") as f:
-        f.write(yaml.dump(context))
-
-
-def default_yaml_config() -> Dict:
-    """Get default configs in ``DEFAULT_CONFIG_PATH``."""
-    with open(DEFAULT_CONFIG_PATH, "r") as f:
-        return yaml.safe_load(f)
-
-
-def _whether_exists_config() -> bool:
-    """Check whether config file already exists in :func:`get_config_file_path`."""
-    return True if get_config_file_path().exists() else False
-
-
-def get_all_configs():
-    """Get all configs from configuration file."""
-    exists = _whether_exists_config()
-    if exists:
-        return read_yaml(str(get_config_file_path()))
-    else:
-        return default_yaml_config()
-
-
-# Add configs as module variables to avoid read configuration multiple times when
-#  Get common configuration setting
-#  Set or get multiple configs in single time
-configs = get_all_configs()
 
 
 def init_config_file() -> None:
-    """Initialize configuration file to :func:`get_config_file_path`."""
+    """Initialize configuration file by default configs."""
-    if _whether_exists_config():
+    if config_path().exists():
         raise PyDSConfException(
             "Initialize configuration false to avoid overwrite configure by accident, file already exists "
             "in %s, if you wan to overwrite the exists configure please remove the exists file manually.",
-            str(get_config_file_path()),
+            str(config_path()),
         )
-    write_yaml(context=default_yaml_config(), path=str(get_config_file_path()))
+    file.write(content=get_configs().to_string(), to_path=str(config_path()))
 
 
 def get_single_config(key: str) -> Any:
     """Get single config to configuration file.
 
-    :param key: The config path want get.
+    Support get from nested keys by delimiter ``.``.
+
+    For example, yaml config as below:
+
+    .. code-block:: yaml
+
+        one:
+          two1:
+            three: value1
+          two2: value2
+
+    you could get ``value1`` and ``value2`` by nested path
+
+    .. code-block:: python
+
+        value1 = get_single_config("one.two1.three")
+        value2 = get_single_config("one.two2")
+
+    :param key: The config key want to get it value.
     """
-    global configs
+    config = get_configs()
-    config_path_dict = PathDict(configs)
+    if key not in config:
-    if key not in config_path_dict:
+        raise PyDSConfException(
-        raise PyDSParamException(
             "Configuration path %s do not exists. Can not get configuration.", key
         )
-    return config_path_dict.__getattr__(key)
+    return config[key]
 
 
 def set_single_config(key: str, value: Any) -> None:
     """Change single config to configuration file.
 
-    :param key: The config path want change.
+    For example, yaml config as below:
+
+    .. code-block:: yaml
+
+        one:
+          two1:
+            three: value1
+          two2: value2
+
+    you could change ``value1`` to ``value3``, also change ``value2`` to ``value4`` by nested path assigned
+
+    .. code-block:: python
+
+        set_single_config["one.two1.three"] = "value3"
+        set_single_config["one.two2"] = "value4"
+
+    :param key: The config key want change.
     :param value: The new value want to set.
     """
-    global configs
+    config = get_configs()
-    config_path_dict = PathDict(configs)
+    if key not in config:
-    if key not in config_path_dict:
+        raise PyDSConfException(
-        raise PyDSParamException(
             "Configuration path %s do not exists. Can not set configuration.", key
         )
-    config_path_dict.__setattr__(key, value)
+    config[key] = value
-    write_yaml(context=dict(config_path_dict), path=str(get_config_file_path()))
+    file.write(content=config.to_string(), to_path=str(config_path()), overwrite=True)
 
 
 # Start Common Configuration Settings
-path_configs = PathDict(copy.deepcopy(configs))
+
+# Add configs as module variables to avoid read configuration multiple times when
+#  Get common configuration setting
+#  Set or get multiple configs in single time
+configs: YamlParser = get_configs()
 
 # Java Gateway Settings
-JAVA_GATEWAY_ADDRESS = str(getattr(path_configs, "java_gateway.address"))
+JAVA_GATEWAY_ADDRESS = configs.get("java_gateway.address")
-JAVA_GATEWAY_PORT = str(getattr(path_configs, "java_gateway.port"))
+JAVA_GATEWAY_PORT = configs.get_int("java_gateway.port")
-JAVA_GATEWAY_AUTO_CONVERT = str(getattr(path_configs, "java_gateway.auto_convert"))
+JAVA_GATEWAY_AUTO_CONVERT = configs.get_bool("java_gateway.auto_convert")
 
 # User Settings
-USER_NAME = str(getattr(path_configs, "default.user.name"))
+USER_NAME = configs.get("default.user.name")
-USER_PASSWORD = str(getattr(path_configs, "default.user.password"))
+USER_PASSWORD = configs.get("default.user.password")
-USER_EMAIL = str(getattr(path_configs, "default.user.email"))
+USER_EMAIL = configs.get("default.user.email")
-USER_PHONE = str(getattr(path_configs, "default.user.phone"))
+USER_PHONE = configs.get("default.user.phone")
-USER_STATE = str(getattr(path_configs, "default.user.state"))
+USER_STATE = configs.get("default.user.state")
 
 # Workflow Settings
-WORKFLOW_PROJECT = str(getattr(path_configs, "default.workflow.project"))
+WORKFLOW_PROJECT = configs.get("default.workflow.project")
-WORKFLOW_TENANT = str(getattr(path_configs, "default.workflow.tenant"))
+WORKFLOW_TENANT = configs.get("default.workflow.tenant")
-WORKFLOW_USER = str(getattr(path_configs, "default.workflow.user"))
+WORKFLOW_USER = configs.get("default.workflow.user")
-WORKFLOW_QUEUE = str(getattr(path_configs, "default.workflow.queue"))
+WORKFLOW_QUEUE = configs.get("default.workflow.queue")
-WORKFLOW_WORKER_GROUP = str(getattr(path_configs, "default.workflow.worker_group"))
+WORKFLOW_WORKER_GROUP = configs.get("default.workflow.worker_group")
-WORKFLOW_TIME_ZONE = str(getattr(path_configs, "default.workflow.time_zone"))
+WORKFLOW_TIME_ZONE = configs.get("default.workflow.time_zone")
 
 # End Common Configuration Setting

dolphinscheduler-python/pydolphinscheduler/src/pydolphinscheduler/core/default_config.yaml

@@ -15,25 +15,33 @@
 # specific language governing permissions and limitations
 # under the License.
 
+# Setting about Java gateway server
 java_gateway:
   # The address of Python gateway server start. Set its value to `0.0.0.0` if your Python API run in different
   # between Python gateway server. It could be be specific to other address like `127.0.0.1` or `localhost`
   address: 127.0.0.1
+
   # The port of Python gateway server start. Define which port you could connect to Python gateway server from
   # Python API side.
   port: 25333
 
+  # Whether automatically convert Python objects to Java Objects. Default value is ``True``. There is some
+  # performance lost when set to ``True`` but for now pydolphinscheduler do not handle the convert issue between
+  # java and Python, mark it as TODO item in the future.
   auto_convert: true
 
+# Setting about dolphinscheduler default value, will use the value set below if property do not set, which
+# including ``user``, ``workflow`` 
 default:
+  # Default value for dolphinscheduler's user object
   user:
     name: userPythonGateway
-    # TODO simple set password same as username
     password: userPythonGateway
     email: userPythonGateway@dolphinscheduler.com
     tenant: tenant_pydolphin
     phone: 11111111111
     state: 1
+  # Default value for dolphinscheduler's workflow object
   workflow:
     project: project-pydolphin
     tenant: tenant_pydolphin

dolphinscheduler-python/pydolphinscheduler/src/pydolphinscheduler/utils/file.py

@@ -0,0 +1,57 @@
+# 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.
+
+"""File util for pydolphinscheduler."""
+
+from pathlib import Path
+from typing import Optional
+
+
+def write(
+    content: str,
+    to_path: str,
+    create: Optional[bool] = True,
+    overwrite: Optional[bool] = False,
+) -> None:
+    """Write configs dict to configuration file.
+
+    :param content: The source string want to write to :param:`to_path`.
+    :param to_path: The path want to write content.
+    :param create: Whether create the file parent directory or not if it does not exist.
+      If set ``True`` will create file with :param:`to_path` if path not exists, otherwise
+      ``False`` will not create. Default ``True``.
+    :param overwrite: Whether overwrite the file or not if it exists. If set ``True``
+      will overwrite the exists content, otherwise ``False`` will not overwrite it. Default ``True``.
+    """
+    path = Path(to_path)
+    if not path.parent.exists():
+        if create:
+            path.parent.mkdir(parents=True)
+        else:
+            raise ValueError(
+                "Parent directory do not exists and set param `create` to `False`."
+            )
+    if not path.exists():
+        with path.open(mode="w") as f:
+            f.write(content)
+    elif overwrite:
+        with path.open(mode="w") as f:
+            f.write(content)
+    else:
+        raise FileExistsError(
+            "File %s already exists and you choose not overwrite mode.", to_path
+        )

dolphinscheduler-python/pydolphinscheduler/src/pydolphinscheduler/utils/path_dict.py

@@ -1,85 +0,0 @@
-# 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.
-
-"""Path dict allow users access value by key chain, like `var.key1.key2.key3`."""
-
-
-# TODO maybe we should rewrite it by `collections.abc.MutableMapping` later,
-#  according to https://stackoverflow.com/q/3387691/7152658
-class PathDict(dict):
-    """Path dict allow users access value by key chain, like `var.key1.key2.key3`."""
-
-    def __init__(self, original=None):
-        super().__init__()
-        if original is None:
-            pass
-        elif isinstance(original, dict):
-            for key in original:
-                self.__setitem__(key, original[key])
-        else:
-            raise TypeError(
-                "Parameter original expected dict type but get %s", type(original)
-            )
-
-    def __getitem__(self, key):
-        if "." not in key:
-            # try:
-            return dict.__getitem__(self, key)
-        # except KeyError:
-        #     # cPickle would get error when key without value pairs, in this case we just skip it
-        #     return
-        my_key, rest_of_key = key.split(".", 1)
-        target = dict.__getitem__(self, my_key)
-        if not isinstance(target, PathDict):
-            raise KeyError(
-                'Cannot get "%s" to (%s) as sub-key of "%s".'
-                % (rest_of_key, repr(target), my_key)
-            )
-        return target[rest_of_key]
-
-    def __setitem__(self, key, value):
-        if "." in key:
-            my_key, rest_of_key = key.split(".", 1)
-            target = self.setdefault(my_key, PathDict())
-            if not isinstance(target, PathDict):
-                raise KeyError(
-                    'Cannot set "%s" from (%s) as sub-key of "%s"'
-                    % (rest_of_key, repr(target), my_key)
-                )
-            target[rest_of_key] = value
-        else:
-            if isinstance(value, dict) and not isinstance(value, PathDict):
-                value = PathDict(value)
-            dict.__setitem__(self, key, value)
-
-    def __contains__(self, key):
-        if "." not in key:
-            return dict.__contains__(self, key)
-        my_key, rest_of_key = key.split(".", 1)
-        target = dict.__getitem__(self, my_key)
-        if not isinstance(target, PathDict):
-            return False
-        return rest_of_key in target
-
-    def setdefault(self, key, default):
-        """Overwrite method dict.setdefault."""
-        if key not in self:
-            self[key] = default
-        return self[key]
-
-    __setattr__ = __setitem__
-    __getattr__ = __getitem__

dolphinscheduler-python/pydolphinscheduler/src/pydolphinscheduler/utils/yaml_parser.py

@@ -0,0 +1,169 @@
+# 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.
+
+"""YAML parser utils, parser yaml string to ``ruamel.yaml`` object and nested key dict."""
+
+import copy
+import io
+from typing import Any, Dict, Optional
+
+from ruamel.yaml import YAML
+from ruamel.yaml.comments import CommentedMap
+
+
+class YamlParser:
+    """A parser to parse Yaml file and provider easier way to access or change value.
+
+    This parser provider delimiter string key to get or set :class:`ruamel.yaml.YAML` object
+
+    For example, yaml config named ``test.yaml`` and its content as below:
+
+    .. code-block:: yaml
+
+        one:
+          two1:
+            three: value1
+          two2: value2
+
+    you could get ``value1`` and ``value2`` by nested path
+
+    .. code-block:: python
+
+        yaml_parser = YamlParser("test.yaml")
+
+        # Use function ``get`` to get value
+        value1 = yaml_parser.get("one.two1.three")
+        # Or use build-in ``__getitem__`` to get value
+        value2 = yaml_parser["one.two2"]
+
+    or you could change ``value1`` to ``value3``, also change ``value2`` to ``value4`` by nested path assigned
+
+    .. code-block:: python
+
+        yaml_parser["one.two1.three"] = "value3"
+        yaml_parser["one.two2"] = "value4"
+    """
+
+    def __init__(self, content: str, delimiter: Optional[str] = "."):
+        self.src_parser = content
+        self._delimiter = delimiter
+
+    @property
+    def src_parser(self) -> CommentedMap:
+        """Get src_parser property."""
+        return self._src_parser
+
+    @src_parser.setter
+    def src_parser(self, content: str) -> None:
+        """Set src_parser property."""
+        self._yaml = YAML()
+        self._src_parser = self._yaml.load(content)
+
+    def parse_nested_dict(
+        self, result: Dict, commented_map: CommentedMap, key: str
+    ) -> None:
+        """Parse :class:`ruamel.yaml.comments.CommentedMap` to nested dict using :param:`delimiter`."""
+        if not isinstance(commented_map, CommentedMap):
+            return
+        for sub_key in set(commented_map.keys()):
+            next_key = f"{key}{self._delimiter}{sub_key}"
+            result[next_key] = commented_map[sub_key]
+            self.parse_nested_dict(result, commented_map[sub_key], next_key)
+
+    @property
+    def dict_parser(self) -> Dict:
+        """Get :class:`CommentedMap` to nested dict using :param:`delimiter` as key delimiter.
+
+        Use Depth-First-Search get all nested key and value, and all key connect by :param:`delimiter`.
+        It make users could easier access or change :class:`CommentedMap` object.
+
+        For example, yaml config named ``test.yaml`` and its content as below:
+
+        .. code-block:: yaml
+
+            one:
+              two1:
+                three: value1
+              two2: value2
+
+        It could parser to nested dict as
+
+        .. code-block:: python
+
+            {
+                "one": ordereddict([('two1', ordereddict([('three', 'value1')])), ('two2', 'value2')]),
+                "one.two1": ordereddict([('three', 'value1')]),
+                "one.two1.three": "value1",
+                "one.two2": "value2",
+            }
+        """
+        res = dict()
+        src_parser_copy = copy.deepcopy(self.src_parser)
+
+        base_keys = set(src_parser_copy.keys())
+        if not base_keys:
+            return res
+        else:
+            for key in base_keys:
+                res[key] = src_parser_copy[key]
+                self.parse_nested_dict(res, src_parser_copy[key], key)
+            return res
+
+    def __contains__(self, key) -> bool:
+        return key in self.dict_parser
+
+    def __getitem__(self, key: str) -> Any:
+        return self.dict_parser[key]
+
+    def __setitem__(self, key: str, val: Any) -> None:
+        if key not in self.dict_parser:
+            raise KeyError("Key %s do not exists.", key)
+
+        mid = None
+        keys = key.split(self._delimiter)
+        for idx, k in enumerate(keys, 1):
+            if idx == len(keys):
+                mid[k] = val
+            else:
+                mid = mid[k] if mid else self.src_parser[k]
+
+    def get(self, key: str) -> Any:
+        """Get value by key, is call ``__getitem__``."""
+        return self[key]
+
+    def get_int(self, key: str) -> int:
+        """Get value and covert it to int."""
+        return int(self.get(key))
+
+    def get_bool(self, key: str) -> bool:
+        """Get value and covert it to boolean."""
+        val = self.get(key)
+        if isinstance(val, str):
+            return val.lower() in {"true", "t"}
+        elif isinstance(val, int):
+            return val != 0
+        else:
+            return val
+
+    def to_string(self) -> str:
+        """Transfer :class:`YamlParser` to string object.
+
+        It is useful when users want to output the :class:`YamlParser` object they change just now.
+        """
+        buf = io.StringIO()
+        self._yaml.dump(self.src_parser, buf)
+        return buf.getvalue()

dolphinscheduler-python/pydolphinscheduler/tests/cli/test_config.py

@@ -23,26 +23,24 @@ from pathlib import Path
 import pytest
 
 from pydolphinscheduler.cli.commands import cli
-from pydolphinscheduler.core.configuration import get_config_file_path
+from pydolphinscheduler.core.configuration import BUILD_IN_CONFIG_PATH, config_path
 from tests.testing.cli import CliTestWrapper
-from tests.testing.constants import DEV_MODE
+from tests.testing.constants import DEV_MODE, ENV_PYDS_HOME
+from tests.testing.file import get_file_content
 
-default_config_path = "~/pydolphinscheduler"
 config_file = "config.yaml"
 
 
 @pytest.fixture
-def delete_tmp_config_file():
+def teardown_file_env():
-    """Util for deleting temp configuration file after test finish."""
+    """Util for deleting temp configuration file and pop env var after test finish."""
     yield
-    config_file_path = get_config_file_path()
+    config_file_path = config_path()
-    config_file_path.unlink()
+    if config_file_path.exists():
+        config_file_path.unlink()
+    os.environ.pop(ENV_PYDS_HOME, None)
 
 
-@pytest.mark.skipif(
-    DEV_MODE,
-    reason="Avoid delete ~/pydolphinscheduler/config.yaml by accident when test locally.",
-)
 @pytest.mark.parametrize(
     "home",
     [
@@ -51,24 +49,23 @@ def delete_tmp_config_file():
         "/tmp/test_abc",
     ],
 )
-def test_config_init(delete_tmp_config_file, home):
+def test_config_init(teardown_file_env, home):
     """Test command line interface `config --init`."""
     if home:
-        os.environ["PYDOLPHINSCHEDULER_HOME"] = home
+        os.environ[ENV_PYDS_HOME] = home
-        config_path = home
+    elif DEV_MODE:
-    else:
+        pytest.skip(
-        config_path = default_config_path
+            "Avoid delete ~/pydolphinscheduler/config.yaml by accident when test locally."
+        )
 
-    path = Path(config_path).joinpath(config_file).expanduser()
+    config_file_path = config_path()
-    assert not path.exists()
+    assert not config_file_path.exists()
 
     cli_test = CliTestWrapper(cli, ["config", "--init"])
     cli_test.assert_success()
 
-    assert path.exists()
+    assert config_file_path.exists()
-    # TODO We have a bug here, yaml dump do not support comment
+    assert get_file_content(config_file_path) == get_file_content(BUILD_IN_CONFIG_PATH)
-    # with path.open(mode="r") as cli_crt, open(path_default_config_yaml, "r") as src:
-    #     assert src.read() == cli_crt.read()
 
 
 @pytest.mark.parametrize(
@@ -80,9 +77,9 @@ def test_config_init(delete_tmp_config_file, home):
         ("default.workflow.project", "project-pydolphin"),
     ],
 )
-def test_config_get(delete_tmp_config_file, key: str, expect: str):
+def test_config_get(teardown_file_env, key: str, expect: str):
     """Test command line interface `config --get XXX`."""
-    os.environ["PYDOLPHINSCHEDULER_HOME"] = "/tmp/pydolphinscheduler"
+    os.environ[ENV_PYDS_HOME] = "/tmp/pydolphinscheduler"
     cli_test = CliTestWrapper(cli, ["config", "--init"])
     cli_test.assert_success()
 
@@ -109,9 +106,9 @@ def test_config_get(delete_tmp_config_file, key: str, expect: str):
         ),
     ],
 )
-def test_config_get_multiple(delete_tmp_config_file, keys: str, expects: str):
+def test_config_get_multiple(teardown_file_env, keys: str, expects: str):
     """Test command line interface `config --get KEY1 --get KEY2 ...`."""
-    os.environ["PYDOLPHINSCHEDULER_HOME"] = "/tmp/pydolphinscheduler"
+    os.environ[ENV_PYDS_HOME] = "/tmp/pydolphinscheduler"
     cli_test = CliTestWrapper(cli, ["config", "--init"])
     cli_test.assert_success()
 
@@ -125,8 +122,6 @@ def test_config_get_multiple(delete_tmp_config_file, keys: str, expects: str):
         cli_test.assert_success(output=f"{keys[idx]} = {expect}", fuzzy=True)
 
 
-# TODO fix command `config --set KEY VAL`
-@pytest.mark.skip(reason="We still have bug in command `config --set KEY VAL`")
 @pytest.mark.parametrize(
     "key, value",
     [
@@ -136,9 +131,11 @@ def test_config_get_multiple(delete_tmp_config_file, keys: str, expects: str):
         ("default.workflow.project", "edit-project-pydolphin"),
     ],
 )
-def test_config_set(delete_tmp_config_file, key: str, value: str):
+def test_config_set(teardown_file_env, key: str, value: str):
     """Test command line interface `config --set KEY VALUE`."""
-    os.environ["PYDOLPHINSCHEDULER_HOME"] = "/tmp/pydolphinscheduler"
+    path = "/tmp/pydolphinscheduler"
+    assert not Path(path).joinpath(config_file).exists()
+    os.environ[ENV_PYDS_HOME] = path
     cli_test = CliTestWrapper(cli, ["config", "--init"])
     cli_test.assert_success()
 
@@ -153,10 +150,6 @@ def test_config_set(delete_tmp_config_file, key: str, value: str):
     assert f"{key} = {value}" in cli_test.result.output
 
 
-# TODO do not skip `config --set KEY1 VAL1 --set KEY2 VAL2`
-@pytest.mark.skip(
-    reason="We still have bug in command `config --set KEY1 VAL1 --set KEY2 VAL2`"
-)
 @pytest.mark.parametrize(
     "keys, values",
     [
@@ -176,9 +169,11 @@ def test_config_set(delete_tmp_config_file, key: str, value: str):
         ),
     ],
 )
-def test_config_set_multiple(delete_tmp_config_file, keys: str, values: str):
+def test_config_set_multiple(teardown_file_env, keys: str, values: str):
     """Test command line interface `config --set KEY1 VAL1 --set KEY2 VAL2`."""
-    os.environ["PYDOLPHINSCHEDULER_HOME"] = "/tmp/pydolphinscheduler"
+    path = "/tmp/pydolphinscheduler"
+    assert not Path(path).joinpath(config_file).exists()
+    os.environ[ENV_PYDS_HOME] = path
     cli_test = CliTestWrapper(cli, ["config", "--init"])
     cli_test.assert_success()
 

dolphinscheduler-python/pydolphinscheduler/tests/core/test_configuration.py

@@ -19,27 +19,160 @@
 
 import os
 from pathlib import Path
+from typing import Any
 
 import pytest
 
 from pydolphinscheduler.core import configuration
+from pydolphinscheduler.core.configuration import (
+    BUILD_IN_CONFIG_PATH,
+    config_path,
+    get_single_config,
+    set_single_config,
+)
+from pydolphinscheduler.exceptions import PyDSConfException
+from pydolphinscheduler.utils.yaml_parser import YamlParser
+from tests.testing.constants import DEV_MODE, ENV_PYDS_HOME
+from tests.testing.file import get_file_content
+
+
+@pytest.fixture
+def teardown_file_env():
+    """Util for deleting temp configuration file and pop env var after test finish."""
+    yield
+    config_file_path = config_path()
+    if config_file_path.exists():
+        config_file_path.unlink()
+    os.environ.pop(ENV_PYDS_HOME, None)
+
+
+@pytest.mark.parametrize(
+    "home, expect",
+    [
+        (None, "~/pydolphinscheduler/config.yaml"),
+        ("/tmp/pydolphinscheduler", "/tmp/pydolphinscheduler/config.yaml"),
+        ("/tmp/test_abc", "/tmp/test_abc/config.yaml"),
+    ],
+)
+def test_config_path(home: Any, expect: str):
+    """Test function :func:`config_path`."""
+    if home:
+        os.environ[ENV_PYDS_HOME] = home
+    assert Path(expect).expanduser() == configuration.config_path()
+
+
+@pytest.mark.parametrize(
+    "home",
+    [
+        None,
+        "/tmp/pydolphinscheduler",
+        "/tmp/test_abc",
+    ],
+)
+def test_init_config_file(teardown_file_env, home: Any):
+    """Test init config file."""
+    if home:
+        os.environ[ENV_PYDS_HOME] = home
+    elif DEV_MODE:
+        pytest.skip(
+            "Avoid delete ~/pydolphinscheduler/config.yaml by accident when test locally."
+        )
+    assert not config_path().exists()
+    configuration.init_config_file()
+    assert config_path().exists()
+
+    assert get_file_content(config_path()) == get_file_content(BUILD_IN_CONFIG_PATH)
+
+
+@pytest.mark.parametrize(
+    "home",
+    [
+        None,
+        "/tmp/pydolphinscheduler",
+        "/tmp/test_abc",
+    ],
+)
+def test_init_config_file_duplicate(teardown_file_env, home: Any):
+    """Test raise error with init config file which already exists."""
+    if home:
+        os.environ[ENV_PYDS_HOME] = home
+    elif DEV_MODE:
+        pytest.skip(
+            "Avoid delete ~/pydolphinscheduler/config.yaml by accident when test locally."
+        )
+    assert not config_path().exists()
+    configuration.init_config_file()
+    assert config_path().exists()
+
+    with pytest.raises(PyDSConfException, match=".*file already exists.*"):
+        configuration.init_config_file()
+
+
+def test_get_configs_build_in():
+    """Test function :func:`get_configs` with build-in config file."""
+    content = get_file_content(BUILD_IN_CONFIG_PATH)
+    assert YamlParser(content).src_parser == configuration.get_configs().src_parser
+    assert YamlParser(content).dict_parser == configuration.get_configs().dict_parser
+
+
+@pytest.mark.parametrize(
+    "key, val, new_val",
+    [
+        ("java_gateway.address", "127.0.0.1", "127.1.1.1"),
+        ("java_gateway.port", 25333, 25555),
+        ("java_gateway.auto_convert", True, False),
+        ("default.user.name", "userPythonGateway", "editUserPythonGateway"),
+        ("default.user.password", "userPythonGateway", "editUserPythonGateway"),
+        (
+            "default.user.email",
+            "userPythonGateway@dolphinscheduler.com",
+            "userPythonGateway@edit.com",
+        ),
+        ("default.user.phone", 11111111111, 22222222222),
+        ("default.user.state", 1, 0),
+        ("default.workflow.project", "project-pydolphin", "eidt-project-pydolphin"),
+        ("default.workflow.tenant", "tenant_pydolphin", "edit_tenant_pydolphin"),
+        ("default.workflow.user", "userPythonGateway", "editUserPythonGateway"),
+        ("default.workflow.queue", "queuePythonGateway", "editQueuePythonGateway"),
+        ("default.workflow.worker_group", "default", "specific"),
+        ("default.workflow.time_zone", "Asia/Shanghai", "Asia/Beijing"),
+    ],
+)
+def test_single_config_get_set(teardown_file_env, key: str, val: Any, new_val: Any):
+    """Test function :func:`get_single_config` and :func:`set_single_config`."""
+    assert val == get_single_config(key)
+    set_single_config(key, new_val)
+    assert new_val == get_single_config(key)
+
+
+def test_single_config_get_set_not_exists_key():
+    """Test function :func:`get_single_config` and :func:`set_single_config` error while key not exists."""
+    not_exists_key = "i_am_not_exists_key"
+    with pytest.raises(PyDSConfException, match=".*do not exists.*"):
+        get_single_config(not_exists_key)
+    with pytest.raises(PyDSConfException, match=".*do not exists.*"):
+        set_single_config(not_exists_key, not_exists_key)
 
 
 @pytest.mark.parametrize(
-    "env, expect",
+    "config_name, expect",
     [
-        (None, "~/pydolphinscheduler"),
+        ("JAVA_GATEWAY_ADDRESS", "127.0.0.1"),
-        ("/tmp/pydolphinscheduler", "/tmp/pydolphinscheduler"),
+        ("JAVA_GATEWAY_PORT", 25333),
-        ("/tmp/test_abc", "/tmp/test_abc"),
+        ("JAVA_GATEWAY_AUTO_CONVERT", True),
+        ("USER_NAME", "userPythonGateway"),
+        ("USER_PASSWORD", "userPythonGateway"),
+        ("USER_EMAIL", "userPythonGateway@dolphinscheduler.com"),
+        ("USER_PHONE", 11111111111),
+        ("USER_STATE", 1),
+        ("WORKFLOW_PROJECT", "project-pydolphin"),
+        ("WORKFLOW_TENANT", "tenant_pydolphin"),
+        ("WORKFLOW_USER", "userPythonGateway"),
+        ("WORKFLOW_QUEUE", "queuePythonGateway"),
+        ("WORKFLOW_WORKER_GROUP", "default"),
+        ("WORKFLOW_TIME_ZONE", "Asia/Shanghai"),
     ],
 )
-def test_get_config_file_path(env, expect):
+def test_get_configuration(config_name: str, expect: Any):
-    """Test get config file path method."""
+    """Test get exists attribute in :mod:`configuration`."""
-    # Avoid env setting by other tests
+    assert expect == getattr(configuration, config_name)
-    os.environ.pop("PYDOLPHINSCHEDULER_HOME", None)
-    if env:
-        os.environ["PYDOLPHINSCHEDULER_HOME"] = env
-    assert (
-        Path(expect).joinpath("config.yaml").expanduser()
-        == configuration.get_config_file_path()
-    )

dolphinscheduler-python/pydolphinscheduler/tests/core/test_default_config_yaml.py

@@ -17,23 +17,23 @@
 
 """Test default config file."""
 
-from typing import Dict
+from ruamel.yaml import YAML
-
+from ruamel.yaml.comments import CommentedMap
-import yaml
 
 from tests.testing.path import path_default_config_yaml
 
 
-def nested_key_check(test_dict: Dict) -> None:
+def nested_key_check(comment_map: CommentedMap) -> None:
     """Test whether default configuration file exists specific character."""
-    for key, val in test_dict.items():
+    for key, val in comment_map.items():
         assert "." not in key, f"There is not allowed special character in key `{key}`."
-        if isinstance(val, dict):
+        if isinstance(val, CommentedMap):
             nested_key_check(val)
 
 
 def test_key_without_dot_delimiter():
     """Test wrapper of whether default configuration file exists specific character."""
+    yaml = YAML()
     with open(path_default_config_yaml, "r") as f:
-        default_config = yaml.safe_load(f)
+        comment_map = yaml.load(f.read())
-        nested_key_check(default_config)
+        nested_key_check(comment_map)

dolphinscheduler-python/pydolphinscheduler/tests/testing/constants.py

@@ -29,6 +29,9 @@ task_without_example = {
     "procedure",
 }
 
+# pydolphinscheduler environment home
+ENV_PYDS_HOME = "PYDOLPHINSCHEDULER_HOME"
+
 # whether in dev mode, if true we will add or remove some tests. Or make be and more detail infos when
 # test failed.
 DEV_MODE = str(

dolphinscheduler-python/pydolphinscheduler/tests/testing/file.py

@@ -0,0 +1,34 @@
+# 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.
+
+"""Testing util about file operating."""
+
+from pathlib import Path
+from typing import Union
+
+
+def get_file_content(path: Union[str, Path]) -> str:
+    """Get file content in given path."""
+    with open(path, mode="r") as f:
+        return f.read()
+
+
+def delete_file(path: Union[str, Path]) -> None:
+    """Delete file in given path."""
+    path = Path(path).expanduser() if isinstance(path, str) else path.expanduser()
+    if path.exists():
+        path.unlink()

dolphinscheduler-python/pydolphinscheduler/tests/utils/test_file.py

@@ -0,0 +1,85 @@
+# 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 file utils."""
+
+import shutil
+from pathlib import Path
+
+import pytest
+
+from pydolphinscheduler.utils import file
+from tests.testing.file import delete_file, get_file_content
+
+content = "test_content"
+file_path = "/tmp/test/file/test_file_write.txt"
+
+
+@pytest.fixture
+def teardown_del_file():
+    """Teardown about delete file."""
+    yield
+    delete_file(file_path)
+
+
+@pytest.fixture
+def setup_crt_first():
+    """Set up and teardown about create file first and then delete it."""
+    file.write(content=content, to_path=file_path)
+    yield
+    delete_file(file_path)
+
+
+def test_write_content(teardown_del_file):
+    """Test function :func:`write` on write behavior with correct content."""
+    assert not Path(file_path).exists()
+    file.write(content=content, to_path=file_path)
+    assert Path(file_path).exists()
+    assert content == get_file_content(file_path)
+
+
+def test_write_not_create_parent(teardown_del_file):
+    """Test function :func:`write` with parent not exists and do not create path."""
+    file_test_dir = Path(file_path).parent
+    if file_test_dir.exists():
+        shutil.rmtree(str(file_test_dir))
+    assert not file_test_dir.exists()
+    with pytest.raises(
+        ValueError,
+        match="Parent directory do not exists and set param `create` to `False`",
+    ):
+        file.write(content=content, to_path=file_path, create=False)
+
+
+def test_write_overwrite(setup_crt_first):
+    """Test success with file exists but set ``True`` to overwrite."""
+    assert Path(file_path).exists()
+
+    new_content = f"new_{content}"
+    file.write(content=new_content, to_path=file_path, overwrite=True)
+    assert new_content == get_file_content(file_path)
+
+
+def test_write_overwrite_error(setup_crt_first):
+    """Test error with file exists but set ``False`` to overwrite."""
+    assert Path(file_path).exists()
+
+    new_content = f"new_{content}"
+    with pytest.raises(
+        FileExistsError, match=".*already exists and you choose not overwrite mode\\."
+    ):
+        file.write(content=new_content, to_path=file_path)

dolphinscheduler-python/pydolphinscheduler/tests/utils/test_path_dict.py

@@ -1,201 +0,0 @@
-# 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."""
-
-import copy
-from typing import Dict, Tuple
-
-import pytest
-
-from pydolphinscheduler.utils.path_dict import PathDict
-
-src_dict_list = [
-    # dict with one single level
-    {"a": 1},
-    # dict with two levels, with same nested keys 'b'
-    {"a": 1, "b": 2, "c": {"d": 3}, "e": {"b": 4}},
-    # dict with three levels, with same nested keys 'b'
-    {"a": 1, "b": 2, "c": {"d": 3}, "e": {"b": {"b": 4}, "f": 5}},
-    # dict with specific key container
-    {
-        "a": 1,
-        "a-b": 2,
-    },
-]
-
-
-@pytest.mark.parametrize("org", src_dict_list)
-def test_val_between_dict_and_path_dict(org: Dict):
-    """Test path dict equal to original dict."""
-    path_dict = PathDict(org)
-    assert org == dict(path_dict)
-
-
-def test_path_dict_basic_attr_access():
-    """Test basic behavior of path dict.
-
-    Including add by attribute, with simple, nested dict, and specific key dict.
-    """
-    expect = copy.deepcopy(src_dict_list[2])
-    path_dict = PathDict(expect)
-
-    # Add node with one level
-    val = 3
-    path_dict.f = val
-    expect.update({"f": val})
-    assert expect == path_dict
-
-    # Add node with multiple level
-    val = {"abc": 123}
-    path_dict.e.g = val
-    expect.update({"e": {"b": {"b": 4}, "f": 5, "g": val}})
-    assert expect == path_dict
-
-    # Specific key
-    expect = copy.deepcopy(src_dict_list[3])
-    path_dict = PathDict(expect)
-    assert 1 == path_dict.a
-    assert 2 == getattr(path_dict, "a-b")
-
-
-@pytest.mark.parametrize(
-    "org, exists, not_exists",
-    [
-        (
-            src_dict_list[0],
-            ("a"),
-            ("b", "a.b"),
-        ),
-        (
-            src_dict_list[1],
-            ("a", "b", "c", "e", "c.d", "e.b"),
-            ("a.b", "c.e", "b.c", "b.e"),
-        ),
-        (
-            src_dict_list[2],
-            ("a", "b", "c", "e", "c.d", "e.b", "e.b.b", "e.b.b", "e.f"),
-            ("a.b", "c.e", "b.c", "b.e", "b.b.f", "b.f"),
-        ),
-    ],
-)
-def test_path_dict_attr(org: Dict, exists: Tuple, not_exists: Tuple):
-    """Test properties' integrity of path dict."""
-    path_dict = PathDict(org)
-    assert all([hasattr(path_dict, path) for path in exists])
-    # assert not any([hasattr(path_dict, path) for path in not_exists])
-
-
-@pytest.mark.parametrize(
-    "org, path_get",
-    [
-        (
-            src_dict_list[0],
-            {"a": 1},
-        ),
-        (
-            src_dict_list[1],
-            {
-                "a": 1,
-                "b": 2,
-                "c": {"d": 3},
-                "c.d": 3,
-                "e": {"b": 4},
-                "e.b": 4,
-            },
-        ),
-        (
-            src_dict_list[2],
-            {
-                "a": 1,
-                "b": 2,
-                "c": {"d": 3},
-                "c.d": 3,
-                "e": {"b": {"b": 4}, "f": 5},
-                "e.b": {"b": 4},
-                "e.b.b": 4,
-                "e.f": 5,
-            },
-        ),
-    ],
-)
-def test_path_dict_get(org: Dict, path_get: Dict):
-    """Test path dict getter function."""
-    path_dict = PathDict(org)
-    assert all([path_get[path] == path_dict.__getattr__(path) for path in path_get])
-
-
-@pytest.mark.parametrize(
-    "org, path_set, expect",
-    [
-        # Add not exists node
-        (
-            src_dict_list[0],
-            {"b": 2},
-            {
-                "a": 1,
-                "b": 2,
-            },
-        ),
-        # Overwrite exists node with different type of value
-        (
-            src_dict_list[0],
-            {"a": "b"},
-            {"a": "b"},
-        ),
-        # Add multiple not exists node with variable types of value
-        (
-            src_dict_list[0],
-            {
-                "b.c.d": 123,
-                "b.c.e": "a",
-                "b.f": {"g": 23, "h": "bc", "i": {"j": "k"}},
-            },
-            {
-                "a": 1,
-                "b": {
-                    "c": {
-                        "d": 123,
-                        "e": "a",
-                    },
-                    "f": {"g": 23, "h": "bc", "i": {"j": "k"}},
-                },
-            },
-        ),
-        # Test complex original data
-        (
-            src_dict_list[2],
-            {
-                "g": 12,
-                "c.h": 34,
-            },
-            {
-                "a": 1,
-                "b": 2,
-                "g": 12,
-                "c": {"d": 3, "h": 34},
-                "e": {"b": {"b": 4}, "f": 5},
-            },
-        ),
-    ],
-)
-def test_path_dict_set(org: Dict, path_set: Dict, expect: Dict):
-    """Test path dict setter function."""
-    path_dict = PathDict(org)
-    for path in path_set:
-        path_dict.__setattr__(path, path_set[path])
-    assert expect == path_dict

dolphinscheduler-python/pydolphinscheduler/tests/utils/test_yaml_parser.py

@@ -0,0 +1,272 @@
+# 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_to_string(src: str, setter: Dict, expect: str):
+    """Test function :func:`YamlParser.to_string`."""
+    yaml_parser = YamlParser(src)
+    for key, val in setter.items():
+        yaml_parser[key] = val
+
+    assert expect == yaml_parser.to_string()
+
+
+@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)