diff --git a/tests/acceptance/avocado_qemu/__init__.py b/tests/acceptance/avocado_qemu/__init__.py
index 83b1741ec8..67f75f66e5 100644
--- a/tests/acceptance/avocado_qemu/__init__.py
+++ b/tests/acceptance/avocado_qemu/__init__.py
@@ -20,6 +20,7 @@
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 @@
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 @@ def fetch_asset(self, name,
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
diff --git a/tests/acceptance/linux_ssh_mips_malta.py b/tests/acceptance/linux_ssh_mips_malta.py
index 052008f02d..61c9079d04 100644
--- a/tests/acceptance/linux_ssh_mips_malta.py
+++ b/tests/acceptance/linux_ssh_mips_malta.py
@@ -12,16 +12,14 @@
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 @@ def get_kernel_info(self, endianess, wordsize):
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 @@ def boot_debian_wheezy_image_and_ssh_login(self, endianess, kernel_path):
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')
diff --git a/tests/acceptance/virtiofs_submounts.py b/tests/acceptance/virtiofs_submounts.py
index 57a7047342..bed8ce44df 100644
--- a/tests/acceptance/virtiofs_submounts.py
+++ b/tests/acceptance/virtiofs_submounts.py
@@ -9,8 +9,6 @@
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)