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

[#357] Synchronize client and CliCommand timeouts

Browse source 
Signed-off-by: Kirill Sosnovskikh <k.sosnovskikh@yadro.com>
This commit is contained in:
k.sosnovskikh 2025-02-25 16:43:34 +03:00 • committed by k.sosnovskikh
parent 97b9b5498a
commit b00d080982
1 changed files with 13 additions and 11 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

24
src/frostfs_testlib/cli/cli_command.py
View file

@ -24,9 +24,7 @@ class CliCommand:
def __init__(self, shell: Shell, cli_exec_path: str, **base_params): def __init__(self, shell: Shell, cli_exec_path: str, **base_params):
self.shell = shell self.shell = shell
self.cli_exec_path = cli_exec_path self.cli_exec_path = cli_exec_path
self.__base_params = " ".join( self.__base_params = " ".join([f"--{param} {value}" for param, value in base_params.items() if value])
[f"--{param} {value}" for param, value in base_params.items() if value]
)
def _format_command(self, command: str, **params) -> str: def _format_command(self, command: str, **params) -> str:
param_str = [] param_str = []
@ -48,9 +46,7 @@ class CliCommand:
val_str = str(value_item).replace("'", "\\'") val_str = str(value_item).replace("'", "\\'")
param_str.append(f"--{param} '{val_str}'") param_str.append(f"--{param} '{val_str}'")
elif isinstance(value, dict): elif isinstance(value, dict):
param_str.append( param_str.append(f'--{param} \'{",".join(f"{key}={val}" for key, val in value.items())}\'')
f'--{param} \'{",".join(f"{key}={val}" for key, val in value.items())}\''
)
else: else:
if "'" in str(value): if "'" in str(value):
value_str = str(value).replace('"', '\\"') value_str = str(value).replace('"', '\\"')
@ -63,12 +59,18 @@ class CliCommand:
return f"{self.cli_exec_path} {self.__base_params} {command or ''} {param_str}" return f"{self.cli_exec_path} {self.__base_params} {command or ''} {param_str}"
def _execute(self, command: Optional[str], **params) -> CommandResult: def _execute(self, command: Optional[str], **params) -> CommandResult:
return self.shell.exec(self._format_command(command, **params)) timeout = int(params["timeout"].rstrip("s")) if params.get("timeout") else None
def _execute_with_password(self, command: Optional[str], password, **params) -> CommandResult:
return self.shell.exec( return self.shell.exec(
self._format_command(command, **params), self._format_command(command, **params),
options=CommandOptions( CommandOptions(timeout=timeout),
interactive_inputs=[InteractiveInput(prompt_pattern="assword", input=password)] )
def _execute_with_password(self, command: Optional[str], password, **params) -> CommandResult:
timeout = int(params["timeout"].rstrip("s")) if params.get("timeout") else None
return self.shell.exec(
self._format_command(command, **params),
CommandOptions(
interactive_inputs=[InteractiveInput(prompt_pattern="assword", input=password)],
timeout=timeout,
), ),
) )