forked from TrueCloudLab/frostfs-testlib
Keep only one ssh connection per host
Signed-off-by: Andrey Berezin <a.berezin@yadro.com>
This commit is contained in:
parent
d039bcc221
commit
2c2af7f8ed
4 changed files with 261 additions and 144 deletions
|
|
@ -1,50 +1,68 @@
|
|||
import os
|
||||
from unittest import SkipTest, TestCase
|
||||
|
||||
import pytest
|
||||
|
||||
from frostfs_testlib.shell.interfaces import CommandOptions, InteractiveInput
|
||||
from frostfs_testlib.shell.ssh_shell import SSHShell
|
||||
from frostfs_testlib.shell.ssh_shell import SshConnectionProvider, SSHShell
|
||||
from helpers import format_error_details, get_output_lines
|
||||
|
||||
|
||||
def init_shell() -> SSHShell:
|
||||
host = os.getenv("SSH_SHELL_HOST")
|
||||
def get_shell(host: str):
|
||||
port = os.getenv("SSH_SHELL_PORT", "22")
|
||||
login = os.getenv("SSH_SHELL_LOGIN")
|
||||
private_key_path = os.getenv("SSH_SHELL_PRIVATE_KEY_PATH")
|
||||
private_key_passphrase = os.getenv("SSH_SHELL_PRIVATE_KEY_PASSPHRASE")
|
||||
|
||||
password = os.getenv("SSH_SHELL_PASSWORD", "")
|
||||
private_key_path = os.getenv("SSH_SHELL_PRIVATE_KEY_PATH", "")
|
||||
private_key_passphrase = os.getenv("SSH_SHELL_PRIVATE_KEY_PASSPHRASE", "")
|
||||
|
||||
if not all([host, login, private_key_path, private_key_passphrase]):
|
||||
# TODO: in the future we might use https://pypi.org/project/mock-ssh-server,
|
||||
# at the moment it is not suitable for us because of its issues with stdin
|
||||
raise SkipTest("SSH connection is not configured")
|
||||
pytest.skip("SSH connection is not configured")
|
||||
|
||||
return SSHShell(
|
||||
host=host,
|
||||
port=port,
|
||||
login=login,
|
||||
password=password,
|
||||
private_key_path=private_key_path,
|
||||
private_key_passphrase=private_key_passphrase,
|
||||
)
|
||||
|
||||
|
||||
class TestSSHShellInteractive(TestCase):
|
||||
@classmethod
|
||||
def setUpClass(cls):
|
||||
cls.shell = init_shell()
|
||||
@pytest.fixture(scope="module")
|
||||
def shell() -> SSHShell:
|
||||
return get_shell(host=os.getenv("SSH_SHELL_HOST"))
|
||||
|
||||
def test_command_with_one_prompt(self):
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def shell_same_host() -> SSHShell:
|
||||
return get_shell(host=os.getenv("SSH_SHELL_HOST"))
|
||||
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def shell_another_host() -> SSHShell:
|
||||
return get_shell(host=os.getenv("SSH_SHELL_HOST_2"))
|
||||
|
||||
|
||||
@pytest.fixture(scope="function", autouse=True)
|
||||
def reset_connection():
|
||||
provider = SshConnectionProvider()
|
||||
provider.drop_all()
|
||||
|
||||
|
||||
class TestSSHShellInteractive:
|
||||
def test_command_with_one_prompt(self, shell: SSHShell):
|
||||
script = "password = input('Password: '); print('\\n' + password)"
|
||||
|
||||
inputs = [InteractiveInput(prompt_pattern="Password", input="test")]
|
||||
result = self.shell.exec(
|
||||
f'python3 -c "{script}"', CommandOptions(interactive_inputs=inputs)
|
||||
)
|
||||
result = shell.exec(f'python3 -c "{script}"', CommandOptions(interactive_inputs=inputs))
|
||||
|
||||
self.assertEqual(0, result.return_code)
|
||||
self.assertEqual(["Password: test", "test"], get_output_lines(result))
|
||||
self.assertEqual("", result.stderr)
|
||||
assert result.return_code == 0
|
||||
assert ["Password: test", "test"] == get_output_lines(result)
|
||||
assert not result.stderr
|
||||
|
||||
def test_command_with_several_prompts(self):
|
||||
def test_command_with_several_prompts(self, shell: SSHShell):
|
||||
script = (
|
||||
"input1 = input('Input1: '); print('\\n' + input1); "
|
||||
"input2 = input('Input2: '); print('\\n' + input2)"
|
||||
|
|
@ -54,86 +72,132 @@ class TestSSHShellInteractive(TestCase):
|
|||
InteractiveInput(prompt_pattern="Input2", input="test2"),
|
||||
]
|
||||
|
||||
result = self.shell.exec(
|
||||
f'python3 -c "{script}"', CommandOptions(interactive_inputs=inputs)
|
||||
)
|
||||
result = shell.exec(f'python3 -c "{script}"', CommandOptions(interactive_inputs=inputs))
|
||||
|
||||
self.assertEqual(0, result.return_code)
|
||||
self.assertEqual(
|
||||
["Input1: test1", "test1", "Input2: test2", "test2"], get_output_lines(result)
|
||||
)
|
||||
self.assertEqual("", result.stderr)
|
||||
assert result.return_code == 0
|
||||
assert ["Input1: test1", "test1", "Input2: test2", "test2"] == get_output_lines(result)
|
||||
assert not result.stderr
|
||||
|
||||
def test_invalid_command_with_check(self):
|
||||
def test_invalid_command_with_check(self, shell: SSHShell):
|
||||
script = "invalid script"
|
||||
inputs = [InteractiveInput(prompt_pattern=".*", input="test")]
|
||||
|
||||
with self.assertRaises(RuntimeError) as raised:
|
||||
self.shell.exec(f'python3 -c "{script}"', CommandOptions(interactive_inputs=inputs))
|
||||
with pytest.raises(RuntimeError) as raised:
|
||||
shell.exec(f'python3 -c "{script}"', CommandOptions(interactive_inputs=inputs))
|
||||
|
||||
error = format_error_details(raised.exception)
|
||||
self.assertIn("SyntaxError", error)
|
||||
self.assertIn("return code: 1", error)
|
||||
error = format_error_details(raised.value)
|
||||
assert "SyntaxError" in error
|
||||
assert "return code: 1" in error
|
||||
|
||||
def test_invalid_command_without_check(self):
|
||||
def test_invalid_command_without_check(self, shell: SSHShell):
|
||||
script = "invalid script"
|
||||
inputs = [InteractiveInput(prompt_pattern=".*", input="test")]
|
||||
|
||||
result = self.shell.exec(
|
||||
result = shell.exec(
|
||||
f'python3 -c "{script}"',
|
||||
CommandOptions(interactive_inputs=inputs, check=False),
|
||||
)
|
||||
self.assertIn("SyntaxError", result.stdout)
|
||||
self.assertEqual(1, result.return_code)
|
||||
assert "SyntaxError" in result.stdout
|
||||
assert result.return_code == 1
|
||||
|
||||
def test_non_existing_binary(self):
|
||||
def test_non_existing_binary(self, shell: SSHShell):
|
||||
inputs = [InteractiveInput(prompt_pattern=".*", input="test")]
|
||||
|
||||
with self.assertRaises(RuntimeError) as raised:
|
||||
self.shell.exec("not-a-command", CommandOptions(interactive_inputs=inputs))
|
||||
with pytest.raises(RuntimeError) as raised:
|
||||
shell.exec("not-a-command", CommandOptions(interactive_inputs=inputs))
|
||||
|
||||
error = format_error_details(raised.exception)
|
||||
self.assertIn("return code: 127", error)
|
||||
error = format_error_details(raised.value)
|
||||
assert "return code: 127" in error
|
||||
|
||||
|
||||
class TestSSHShellNonInteractive(TestCase):
|
||||
@classmethod
|
||||
def setUpClass(cls):
|
||||
cls.shell = init_shell()
|
||||
|
||||
def test_correct_command(self):
|
||||
class TestSSHShellNonInteractive:
|
||||
def test_correct_command(self, shell: SSHShell):
|
||||
script = "print('test')"
|
||||
|
||||
result = self.shell.exec(f'python3 -c "{script}"')
|
||||
result = shell.exec(f'python3 -c "{script}"')
|
||||
|
||||
self.assertEqual(0, result.return_code)
|
||||
self.assertEqual("test", result.stdout.strip())
|
||||
self.assertEqual("", result.stderr)
|
||||
assert result.return_code == 0
|
||||
assert result.stdout.strip() == "test"
|
||||
assert not result.stderr
|
||||
|
||||
def test_invalid_command_with_check(self):
|
||||
def test_invalid_command_with_check(self, shell: SSHShell):
|
||||
script = "invalid script"
|
||||
|
||||
with self.assertRaises(RuntimeError) as raised:
|
||||
self.shell.exec(f'python3 -c "{script}"')
|
||||
with pytest.raises(RuntimeError) as raised:
|
||||
shell.exec(f'python3 -c "{script}"')
|
||||
|
||||
error = format_error_details(raised.exception)
|
||||
self.assertIn("Error", error)
|
||||
self.assertIn("return code: 1", error)
|
||||
error = format_error_details(raised.value)
|
||||
assert "Error" in error
|
||||
assert "return code: 1" in error
|
||||
|
||||
def test_invalid_command_without_check(self):
|
||||
def test_invalid_command_without_check(self, shell: SSHShell):
|
||||
script = "invalid script"
|
||||
|
||||
result = self.shell.exec(f'python3 -c "{script}"', CommandOptions(check=False))
|
||||
result = shell.exec(f'python3 -c "{script}"', CommandOptions(check=False))
|
||||
|
||||
self.assertEqual(1, result.return_code)
|
||||
assert result.return_code == 1
|
||||
# TODO: we have inconsistency with local shell here, the local shell captures error info
|
||||
# in stdout while ssh shell captures it in stderr
|
||||
self.assertIn("Error", result.stderr)
|
||||
assert "Error" in result.stderr
|
||||
|
||||
def test_non_existing_binary(self):
|
||||
with self.assertRaises(RuntimeError) as exc:
|
||||
self.shell.exec("not-a-command")
|
||||
def test_non_existing_binary(self, shell: SSHShell):
|
||||
with pytest.raises(RuntimeError) as raised:
|
||||
shell.exec("not-a-command")
|
||||
|
||||
error = format_error_details(exc.exception)
|
||||
self.assertIn("Error", error)
|
||||
self.assertIn("return code: 127", error)
|
||||
error = format_error_details(raised.value)
|
||||
assert "Error" in error
|
||||
assert "return code: 127" in error
|
||||
|
||||
|
||||
class TestSSHShellConnection:
|
||||
def test_connection_provider_is_singleton(self):
|
||||
provider = SshConnectionProvider()
|
||||
provider2 = SshConnectionProvider()
|
||||
assert id(provider) == id(provider2)
|
||||
|
||||
def test_connection_provider_has_creds(self, shell: SSHShell):
|
||||
provider = SshConnectionProvider()
|
||||
assert len(provider.creds) == 1
|
||||
assert len(provider.connections) == 0
|
||||
|
||||
def test_connection_provider_has_only_one_connection(self, shell: SSHShell):
|
||||
provider = SshConnectionProvider()
|
||||
assert len(provider.connections) == 0
|
||||
shell.exec("echo 1")
|
||||
assert len(provider.connections) == 1
|
||||
shell.exec("echo 2")
|
||||
assert len(provider.connections) == 1
|
||||
shell.drop()
|
||||
assert len(provider.connections) == 0
|
||||
|
||||
def test_connection_same_host(self, shell: SSHShell, shell_same_host: SSHShell):
|
||||
provider = SshConnectionProvider()
|
||||
assert len(provider.connections) == 0
|
||||
|
||||
shell.exec("echo 1")
|
||||
assert len(provider.connections) == 1
|
||||
|
||||
shell_same_host.exec("echo 2")
|
||||
assert len(provider.connections) == 1
|
||||
|
||||
shell.drop()
|
||||
assert len(provider.connections) == 0
|
||||
|
||||
shell.exec("echo 3")
|
||||
assert len(provider.connections) == 1
|
||||
|
||||
def test_connection_another_host(self, shell: SSHShell, shell_another_host: SSHShell):
|
||||
provider = SshConnectionProvider()
|
||||
assert len(provider.connections) == 0
|
||||
|
||||
shell.exec("echo 1")
|
||||
assert len(provider.connections) == 1
|
||||
|
||||
shell_another_host.exec("echo 2")
|
||||
assert len(provider.connections) == 2
|
||||
|
||||
shell.drop()
|
||||
assert len(provider.connections) == 1
|
||||
|
||||
shell_another_host.drop()
|
||||
assert len(provider.connections) == 0
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue