Re: [PATCH v3 04/11] Acceptance Tests: move useful ssh methods to base class

["Philippe Mathieu-Daudé" <f4bug@amsat.org>]

On 4/12/21 6:46 AM, Cleber Rosa wrote:
> 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>
> ---
>  tests/acceptance/avocado_qemu/__init__.py | 48 ++++++++++++++++++++++-
>  tests/acceptance/linux_ssh_mips_malta.py  | 40 ++-----------------
>  tests/acceptance/virtiofs_submounts.py    | 37 -----------------
>  3 files changed, 50 insertions(+), 75 deletions(-)
> 
> diff --git a/tests/acceptance/avocado_qemu/__init__.py b/tests/acceptance/avocado_qemu/__init__.py
> index 83b1741ec85..67f75f66e56 100644
> --- a/tests/acceptance/avocado_qemu/__init__.py
> +++ b/tests/acceptance/avocado_qemu/__init__.py

Don't we need to update the copyright in this file?
I ask because I'm never sure how that works.... In my patches
when I move code which is in a copyrighted file, I add the same
copyright lines to the copied file.

> @@ -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 052008f02d4..61c9079d047 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')