TrueCloudLab/frostfs-testlib
25
0
Fork 20
Code Issues Pull requests Releases 1 Activity Actions

[#363] Add accounting for timeout inaccuracy between process and cli

Browse source 
Signed-off-by: Kirill Sosnovskikh <k.sosnovskikh@yadro.com>
This commit is contained in:
k.sosnovskikh 2025-03-11 17:22:13 +03:00 • committed by Andrey Berezin
parent c2af1bba5c
commit dfb048fe51
1 changed files with 8 additions and 3 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -1,10 +1,11 @@
from typing import Optional from typing import Optional
from frostfs_testlib.shell import CommandOptions, CommandResult, InteractiveInput, Shell from frostfs_testlib.shell import CommandOptions, CommandResult, InteractiveInput, Shell
from frostfs_testlib.utils.datetime_utils import parse_time
class CliCommand: class CliCommand:
TIMEOUT_INACCURACY = 10
WALLET_SOURCE_ERROR_MSG = "Provide either wallet or wallet_config to specify wallet location" WALLET_SOURCE_ERROR_MSG = "Provide either wallet or wallet_config to specify wallet location"
WALLET_PASSWD_ERROR_MSG = "Provide either wallet_password or wallet_config to specify password" WALLET_PASSWD_ERROR_MSG = "Provide either wallet_password or wallet_config to specify password"
@ -59,14 +60,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:
timeout = int(params["timeout"].rstrip("s")) if params.get("timeout") else None if timeout := params.get("timeout"):
timeout = parse_time(timeout) + self.TIMEOUT_INACCURACY
return self.shell.exec( return self.shell.exec(
self._format_command(command, **params), self._format_command(command, **params),
CommandOptions(timeout=timeout), CommandOptions(timeout=timeout),
) )
def _execute_with_password(self, command: Optional[str], password, **params) -> CommandResult: def _execute_with_password(self, command: Optional[str], password, **params) -> CommandResult:
timeout = int(params["timeout"].rstrip("s")) if params.get("timeout") else None if timeout := params.get("timeout"):
timeout = parse_time(timeout) + self.TIMEOUT_INACCURACY
return self.shell.exec( return self.shell.exec(
self._format_command(command, **params), self._format_command(command, **params),
CommandOptions( CommandOptions(