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

[#147] Provide custom environment to ssh connection

Browse source 
Signed-off-by: Dmitry Anurin <d.anurin@yadro.com>
This commit is contained in:
d.anurin 2023-12-14 12:53:51 +03:00 committed by Dmitry Anurin
parent 89522b607c
commit 02f3ef6b40
3 changed files with 7 additions and 3 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
src/frostfs_testlib/hosting/config.py
View file

@ -67,6 +67,7 @@ class HostConfig:
clis: list[CLIConfig] = field(default_factory=list)
attributes: dict[str, str] = field(default_factory=dict)
interfaces: dict[str, str] = field(default_factory=dict)
environment: dict[str, str] = field(default_factory=dict)
def __post_init__(self) -> None:
self.services = [ServiceConfig(**service) for service in self.services or []]

2
src/frostfs_testlib/shell/command_inspectors.py
View file

@ -9,7 +9,7 @@ class SudoInspector(CommandInspector):
def inspect(self, original_command: str, command: str) -> str:
if not command.startswith("sudo"):
return f"sudo -i {command}"
return f"sudo {command}"
return command

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

@ -185,6 +185,7 @@ class SSHShell(Shell):
private_key_passphrase: Optional[str] = None,
port: str = "22",
command_inspectors: Optional[list[CommandInspector]] = None,
custom_environment: Optional[dict] = None
) -> None:
super().__init__()
self.connection_provider = SshConnectionProvider()
@ -196,6 +197,8 @@ class SSHShell(Shell):
self.command_inspectors = command_inspectors or []
self.environment = custom_environment
@property
def _connection(self):
return self.connection_provider.provide(self.host, self.port)
@ -224,7 +227,7 @@ class SSHShell(Shell):
@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, environment=self.environment)
for interactive_input in options.interactive_inputs:
input = interactive_input.input
if not input.endswith("\n"):
@ -251,7 +254,7 @@ class SSHShell(Shell):
@log_command
def _exec_non_interactive(self, command: str, options: CommandOptions) -> CommandResult:
try:
stdin, stdout, stderr = self._connection.exec_command(command, timeout=options.timeout)
stdin, stdout, stderr = self._connection.exec_command(command, timeout=options.timeout, environment=self.environment)
if options.close_stdin:
stdin.close()