OpenE2K
/
qemu-e2k
13
4
Fork 0
Code Issues 4 Pull Requests Projects 2 Releases Activity

Acceptance Tests: move useful ssh methods to base class 

Both the virtiofs submounts and the linux ssh mips malta tests
contains useful methods related to ssh that deserve to be made
available to other tests.  Let's move them to an auxiliary, mix-in
class that will be used on the base LinuxTest class.

The method that helps with setting up an ssh connection will now
support both key and password based authentication, defaulting to key
based.

Signed-off-by: Cleber Rosa <crosa@redhat.com>
Reviewed-by: Wainer dos Santos Moschetta <wainersm@redhat.com>
Reviewed-by: Willian Rampazzo <willianr@redhat.com>
Reviewed-by: Eric Auger <eric.auger@redhat.com>
Signed-off-by: Cleber Rosa <crosa@redhat.com>
Message-Id: <20210412044644.55083-5-crosa@redhat.com>
Signed-off-by: John Snow <jsnow@redhat.com>
This commit is contained in:
Cleber Rosa 2021-04-12 00:46:37 -04:00 committed by John Snow
parent 976218cbe7
commit 7edee7ad94
3 changed files with 50 additions and 75 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

48
tests/acceptance/avocado_qemu/__init__.py
View File

@ -20,6 +20,7 @@ import avocado
from avocado.utils import cloudinit
from avocado.utils import datadrainer
from avocado.utils import network
from avocado.utils import ssh
from avocado.utils import vmimage
from avocado.utils.path import find_command
@ -43,6 +44,8 @@ sys.path.append(os.path.join(SOURCE_DIR, 'python'))
from qemu.accel import kvm_available
from qemu.accel import tcg_available
from qemu.machine import QEMUMachine
from qemu.utils import get_info_usernet_hostfwd_port
def is_readable_executable_file(path):
return os.path.isfile(path) and os.access(path, os.R_OK | os.X_OK)
@ -253,7 +256,50 @@ class Test(avocado.Test):
cancel_on_missing=cancel_on_missing)
class LinuxTest(Test):
class LinuxSSHMixIn:
"""Contains utility methods for interacting with a guest via SSH."""
def ssh_connect(self, username, credential, credential_is_key=True):
self.ssh_logger = logging.getLogger('ssh')
res = self.vm.command('human-monitor-command',
command_line='info usernet')
port = get_info_usernet_hostfwd_port(res)
self.assertIsNotNone(port)
self.assertGreater(port, 0)
self.log.debug('sshd listening on port: %d', port)
if credential_is_key:
self.ssh_session = ssh.Session('127.0.0.1', port=port,
user=username, key=credential)
else:
self.ssh_session = ssh.Session('127.0.0.1', port=port,
user=username, password=credential)
for i in range(10):
try:
self.ssh_session.connect()
return
except:
time.sleep(4)
pass
self.fail('ssh connection timeout')
def ssh_command(self, command):
self.ssh_logger.info(command)
result = self.ssh_session.cmd(command)
stdout_lines = [line.rstrip() for line
in result.stdout_text.splitlines()]
for line in stdout_lines:
self.ssh_logger.info(line)
stderr_lines = [line.rstrip() for line
in result.stderr_text.splitlines()]
for line in stderr_lines:
self.ssh_logger.warning(line)
self.assertEqual(result.exit_status, 0,
f'Guest command failed: {command}')
return stdout_lines, stderr_lines
class LinuxTest(Test, LinuxSSHMixIn):
"""Facilitates having a cloud-image Linux based available.
For tests that indend to interact with guests, this is a better choice

40
tests/acceptance/linux_ssh_mips_malta.py
View File

@ -12,16 +12,14 @@ import logging
import time
from avocado import skipUnless
from avocado_qemu import Test
from avocado_qemu import Test, LinuxSSHMixIn
from avocado_qemu import wait_for_console_pattern
from avocado.utils import process
from avocado.utils import archive
from avocado.utils import ssh
from qemu.utils import get_info_usernet_hostfwd_port
class LinuxSSH(Test):
class LinuxSSH(Test, LinuxSSHMixIn):
timeout = 150 # Not for 'configure --enable-debug --enable-debug-tcg'
@ -72,41 +70,9 @@ class LinuxSSH(Test):
def setUp(self):
super(LinuxSSH, self).setUp()
def ssh_connect(self, username, password):
self.ssh_logger = logging.getLogger('ssh')
res = self.vm.command('human-monitor-command',
command_line='info usernet')
port = get_info_usernet_hostfwd_port(res)
if not port:
self.cancel("Failed to retrieve SSH port")
self.log.debug("sshd listening on port:" + port)
self.ssh_session = ssh.Session(self.VM_IP, port=int(port),
user=username, password=password)
for i in range(10):
try:
self.ssh_session.connect()
return
except:
time.sleep(4)
pass
self.fail("ssh connection timeout")
def ssh_disconnect_vm(self):
self.ssh_session.quit()
def ssh_command(self, command, is_root=True):
self.ssh_logger.info(command)
result = self.ssh_session.cmd(command)
stdout_lines = [line.rstrip() for line
in result.stdout_text.splitlines()]
for line in stdout_lines:
self.ssh_logger.info(line)
stderr_lines = [line.rstrip() for line
in result.stderr_text.splitlines()]
for line in stderr_lines:
self.ssh_logger.warning(line)
return stdout_lines, stderr_lines
def boot_debian_wheezy_image_and_ssh_login(self, endianess, kernel_path):
image_url, image_hash = self.get_image_info(endianess)
image_path = self.fetch_asset(image_url, asset_hash=image_hash)
@ -127,7 +93,7 @@ class LinuxSSH(Test):
wait_for_console_pattern(self, console_pattern, 'Oops')
self.log.info('sshd ready')
self.ssh_connect('root', 'root')
self.ssh_connect('root', 'root', False)
def shutdown_via_ssh(self):
self.ssh_command('poweroff')

37
tests/acceptance/virtiofs_submounts.py
View File

@ -9,8 +9,6 @@ from avocado_qemu import LinuxTest, BUILD_DIR
from avocado_qemu import wait_for_console_pattern
from avocado.utils import ssh
from qemu.utils import get_info_usernet_hostfwd_port
def run_cmd(args):
subp = subprocess.Popen(args,
@ -75,41 +73,6 @@ class VirtiofsSubmountsTest(LinuxTest):
:avocado: tags=accel:kvm
"""
def ssh_connect(self, username, keyfile):
self.ssh_logger = logging.getLogger('ssh')
res = self.vm.command('human-monitor-command',
command_line='info usernet')
port = get_info_usernet_hostfwd_port(res)
self.assertIsNotNone(port)
self.assertGreater(port, 0)
self.log.debug('sshd listening on port: %d', port)
self.ssh_session = ssh.Session('127.0.0.1', port=port,
user=username, key=keyfile)
for i in range(10):
try:
self.ssh_session.connect()
return
except:
time.sleep(4)
pass
self.fail('ssh connection timeout')
def ssh_command(self, command):
self.ssh_logger.info(command)
result = self.ssh_session.cmd(command)
stdout_lines = [line.rstrip() for line
in result.stdout_text.splitlines()]
for line in stdout_lines:
self.ssh_logger.info(line)
stderr_lines = [line.rstrip() for line
in result.stderr_text.splitlines()]
for line in stderr_lines:
self.ssh_logger.warning(line)
self.assertEqual(result.exit_status, 0,
f'Guest command failed: {command}')
return stdout_lines, stderr_lines
def run(self, args, ignore_error=False):
stdout, stderr, ret = run_cmd(args)