mkadilov/frostfs-testlib
1
0
Fork 0
forked from TrueCloudLab/frostfs-testlib
Code Pull requests Activity

Add support test maintenance

Browse source 
Signed-off-by: Dmitriy Zayakin <d.zayakin@yadro.com>
This commit is contained in:
Dmitriy Zayakin 2023-11-20 13:54:47 +03:00 • committed by Dmitriy Zayakin
parent 22647c6d59
commit ed70dada96
8 changed files with 290 additions and 96 deletions
Download patch file Download diff file Expand all files Collapse all files

7
src/frostfs_testlib/shell/local_shell.py
View file

@ -62,7 +62,8 @@ class LocalShell(Shell):
if options.check and result.return_code != 0:
raise RuntimeError(
f"Command: {command}\nreturn code: {result.return_code}\n"
f"Output: {result.stdout}"
f"Output: {result.stdout}\n"
f"Stderr: {result.stderr}\n"
)
return result
@ -94,9 +95,7 @@ class LocalShell(Shell):
return_code=exc.returncode,
)
raise RuntimeError(
f"Command: {command}\nError:\n"
f"return code: {exc.returncode}\n"
f"output: {exc.output}"
f"Command: {command}\nError:\n" f"return code: {exc.returncode}\n" f"output: {exc.output}"
) from exc
except OSError as exc:
raise RuntimeError(f"Command: {command}\nOutput: {exc.strerror}") from exc

33
src/frostfs_testlib/shell/ssh_shell.py
View file

@ -6,27 +6,11 @@ from functools import lru_cache, wraps
from time import sleep
from typing import ClassVar, Optional, Tuple
from paramiko import (
AutoAddPolicy,
Channel,
ECDSAKey,
Ed25519Key,
PKey,
RSAKey,
SSHClient,
SSHException,
ssh_exception,
)
from paramiko import AutoAddPolicy, Channel, ECDSAKey, Ed25519Key, PKey, RSAKey, SSHClient, SSHException, ssh_exception
from paramiko.ssh_exception import AuthenticationException
from frostfs_testlib.reporter import get_reporter
from frostfs_testlib.shell.interfaces import (
CommandInspector,
CommandOptions,
CommandResult,
Shell,
SshCredentials,
)
from frostfs_testlib.shell.interfaces import CommandInspector, CommandOptions, CommandResult, Shell, SshCredentials
logger = logging.getLogger("frostfs.testlib.shell")
reporter = get_reporter()
@ -97,8 +81,7 @@ class SshConnectionProvider:
)
else:
logger.info(
f"Trying to connect to host {host} as {creds.ssh_login} using password "
f"(attempt {attempt})"
f"Trying to connect to host {host} as {creds.ssh_login} using password " f"(attempt {attempt})"
)
connection.connect(
hostname=host,
@ -141,9 +124,7 @@ class HostIsNotAvailable(Exception):
def log_command(func):
@wraps(func)
def wrapper(
shell: "SSHShell", command: str, options: CommandOptions, *args, **kwargs
) -> CommandResult:
def wrapper(shell: "SSHShell", command: str, options: CommandOptions, *args, **kwargs) -> CommandResult:
command_info = command.removeprefix("$ProgressPreference='SilentlyContinue'\n")
with reporter.step(command_info):
logger.info(f'Execute command "{command}" on "{shell.host}"')
@ -238,15 +219,13 @@ class SSHShell(Shell):
if options.check and result.return_code != 0:
raise RuntimeError(
f"Command: {command}\nreturn code: {result.return_code}\nOutput: {result.stdout}"
f"Command: {command}\nreturn code: {result.return_code}\nOutput: {result.stdout}\nStderr: {result.stderr}\n"
)
return result
@log_command
def _exec_interactive(self, command: str, options: CommandOptions) -> CommandResult:
stdin, stdout, stderr = self._connection.exec_command(
command, timeout=options.timeout, get_pty=True
)
stdin, stdout, stderr = self._connection.exec_command(command, timeout=options.timeout, get_pty=True)
for interactive_input in options.interactive_inputs:
input = interactive_input.input
if not input.endswith("\n"):