mirror of
https://github.com/scylladb/scylladb.git
synced 2026-04-29 04:37:00 +00:00
4d8f74f301
so that this helper can be shared across more tests. `test_info.py` will be using it as well. Signed-off-by: Kefu Chai <kefu.chai@scylladb.com>
92 lines
2.5 KiB
Python
92 lines
2.5 KiB
Python
#
|
|
# Copyright 2023-present ScyllaDB
|
|
#
|
|
# SPDX-License-Identifier: AGPL-3.0-or-later
|
|
#
|
|
|
|
import pytest
|
|
import subprocess
|
|
import itertools
|
|
|
|
|
|
def _do_check_nodetool_fails_with(
|
|
nodetool,
|
|
nodetool_args: tuple,
|
|
nodetool_kwargs: dict,
|
|
matcher):
|
|
|
|
with pytest.raises(subprocess.CalledProcessError) as e:
|
|
nodetool(*nodetool_args, **nodetool_kwargs)
|
|
|
|
err_lines = e.value.stderr.rstrip().split('\n')
|
|
out_lines = e.value.stdout.rstrip().split('\n')
|
|
|
|
matcher(out_lines, err_lines)
|
|
|
|
|
|
def check_nodetool_fails_with(
|
|
nodetool,
|
|
nodetool_args: tuple,
|
|
nodetool_kwargs: dict,
|
|
expected_errors: list[str]):
|
|
|
|
def matcher(out_lines, err_lines):
|
|
match = False
|
|
for expected_error in expected_errors:
|
|
if expected_error in err_lines or expected_error in out_lines:
|
|
match = True
|
|
break
|
|
assert match, f'expected error not found in {err_lines=}, {out_lines}'
|
|
|
|
_do_check_nodetool_fails_with(nodetool, nodetool_args, nodetool_kwargs, matcher)
|
|
|
|
|
|
def check_nodetool_fails_with_all(
|
|
nodetool,
|
|
nodetool_args: tuple,
|
|
nodetool_kwargs: dict,
|
|
expected_errors: list[str]):
|
|
|
|
def matcher(out_lines, err_lines):
|
|
match = 0
|
|
for expected_error in expected_errors:
|
|
if expected_error in err_lines or expected_error in out_lines:
|
|
match += 1
|
|
assert match == len(expected_errors), f'not all expected errors are found in {out_lines=} and {err_lines}'
|
|
|
|
_do_check_nodetool_fails_with(nodetool, nodetool_args, nodetool_kwargs, matcher)
|
|
|
|
|
|
def check_nodetool_fails_with_error_contains(
|
|
nodetool,
|
|
nodetool_args: tuple,
|
|
nodetool_kwargs: dict,
|
|
expected_errors: list[str]):
|
|
|
|
def matcher(out_lines, err_lines):
|
|
match = False
|
|
for expected_error in expected_errors:
|
|
for err_line in itertools.chain(out_lines, err_lines):
|
|
if expected_error in err_line:
|
|
match = True
|
|
break
|
|
|
|
assert match, f'none of expected errors is found in {out_lines=} and {err_lines}'
|
|
|
|
_do_check_nodetool_fails_with(nodetool, nodetool_args, nodetool_kwargs, matcher)
|
|
|
|
|
|
def format_size(v):
|
|
units = {
|
|
1 << 40: 'TB',
|
|
1 << 30: 'GB',
|
|
1 << 20: 'MB',
|
|
1 << 10: 'KB',
|
|
}
|
|
|
|
for n, unit in units.items():
|
|
if v > n:
|
|
d = v / n
|
|
return f'{d:.2f} {unit}'
|
|
return f'{v} bytes'
|