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