分布式调度框架。
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
 
 
 
 
 
 

91 lines
3.9 KiB

# 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.
"""Utils of command line test."""
import os
from click.testing import CliRunner
class CliTestWrapper:
"""Wrap command click CliRunner.invoke."""
_dev_mode_env_name = "PY_DOLPHINSCHEDULER_DEV_MODE"
_dev_mode_true_val = {"true", "t", "1"}
def __init__(self, *args, **kwargs):
runner = CliRunner()
self.result = runner.invoke(*args, **kwargs)
self.show_result_output()
def _assert_output(self, output: str = None, fuzzy: bool = False):
"""Assert between `CliRunner.invoke.result.output` and parameter `output`.
:param output: The output will check compare to the ``CliRunner.invoke.output``.
:param fuzzy: A flag define whether assert :param:`output` in fuzzy or not.
Check if `CliRunner.invoke.output` contain :param:`output` is set ``True``
and CliRunner.invoke.output equal to :param:`output` if we set it ``False``.
"""
if not output:
return
if fuzzy:
assert output in self.result.output
else:
assert self.result.output.rstrip("\n") == output
def show_result_output(self):
"""Print `CliRunner.invoke.result` output content in debug mode.
It read variable named `PY_DOLPHINSCHEDULER_DEV_MODE` from env, when it set to `true` or `t` or `1`
will print result output when class :class:`CliTestWrapper` is initialization.
"""
dev_mode = str(os.getenv(self._dev_mode_env_name))
if dev_mode.strip().lower() in self._dev_mode_true_val:
print(f"\n{self.result.output}\n")
def assert_success(self, output: str = None, fuzzy: bool = False):
"""Assert test is success.
It would check whether `CliRunner.invoke.exit_code` equals to `0`, with no
exception at the same time. It's also can test the content of `CliRunner.invoke.output`.
:param output: The output will check compare to the ``CliRunner.invoke.output``.
:param fuzzy: A flag define whether assert :param:`output` in fuzzy or not.
Check if `CliRunner.invoke.output` contain :param:`output` is set ``True``
and CliRunner.invoke.output equal to :param:`output` if we set it ``False``.
"""
assert self.result.exit_code == 0
if self.result.exception:
raise self.result.exception
self._assert_output(output, fuzzy)
def assert_fail(self, ret_code: int, output: str = None, fuzzy: bool = False):
"""Assert test is fail.
It would check whether `CliRunner.invoke.exit_code` equals to :param:`ret_code`,
and it will also can test the content of `CliRunner.invoke.output`.
:param ret_code: The returning code of this fail test.
:param output: The output will check compare to the ``CliRunner.invoke.output``.
:param fuzzy: A flag define whether assert :param:`output` in fuzzy or not.
Check if `CliRunner.invoke.output` contain :param:`output` is set ``True``
and CliRunner.invoke.output equal to :param:`output` if we set it ``False``.
"""
assert ret_code == self.result.exit_code
self._assert_output(output, fuzzy)