# 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_str_repr(src: str, setter: Dict, expect: str): """Test function :func:`YamlParser.to_string`.""" yaml_parser = YamlParser(src) # Equal before change assert f"YamlParser({src})" == repr(yaml_parser) assert src == str(yaml_parser) for key, val in setter.items(): yaml_parser[key] = val # Equal after changed assert expect == str(yaml_parser) assert f"YamlParser({expect})" == repr(yaml_parser)