diff mbox series

[15/22] Acceptance Tests: move useful ssh methods to base class

Message ID 20210203172357.1422425-16-crosa@redhat.com (mailing list archive)
State New, archived
Headers show
Series Acceptance Test: introduce base class for Linux based tests | expand

Commit Message

Cleber Rosa Feb. 3, 2021, 5:23 p.m. UTC 
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 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>
---
 tests/acceptance/avocado_qemu/__init__.py | 49 ++++++++++++++++++++++-
 tests/acceptance/linux_ssh_mips_malta.py  | 38 ++----------------
 tests/acceptance/virtiofs_submounts.py    | 36 -----------------
 3 files changed, 51 insertions(+), 72 deletions(-)

Comments

Wainer dos Santos Moschetta Feb. 9, 2021, 7:56 p.m. UTC | #1 
On 2/3/21 2:23 PM, 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 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>
> ---
>   tests/acceptance/avocado_qemu/__init__.py | 49 ++++++++++++++++++++++-
>   tests/acceptance/linux_ssh_mips_malta.py  | 38 ++----------------
>   tests/acceptance/virtiofs_submounts.py    | 36 -----------------
>   3 files changed, 51 insertions(+), 72 deletions(-)


Reviewed-by: Wainer dos Santos Moschetta <wainersm@redhat.com>


>
> diff --git a/tests/acceptance/avocado_qemu/__init__.py b/tests/acceptance/avocado_qemu/__init__.py
> index b06692a59d..f0649e5011 100644
> --- a/tests/acceptance/avocado_qemu/__init__.py
> +++ b/tests/acceptance/avocado_qemu/__init__.py
> @@ -10,6 +10,7 @@
>   
>   import logging
>   import os
> +import re
>   import sys
>   import uuid
>   import tempfile
> @@ -19,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
>   
> @@ -40,6 +42,8 @@ else:
>   sys.path.append(os.path.join(SOURCE_DIR, 'python'))
>   
>   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)
> @@ -215,7 +219,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
> diff --git a/tests/acceptance/linux_ssh_mips_malta.py b/tests/acceptance/linux_ssh_mips_malta.py
> index 1742235758..3f590a081f 100644
> --- a/tests/acceptance/linux_ssh_mips_malta.py
> +++ b/tests/acceptance/linux_ssh_mips_malta.py
> @@ -12,7 +12,7 @@ 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
> @@ -21,7 +21,7 @@ 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 +72,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: %d", port)
> -        self.ssh_session = ssh.Session(self.VM_IP, port=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 +95,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')
> diff --git a/tests/acceptance/virtiofs_submounts.py b/tests/acceptance/virtiofs_submounts.py
> index 25ea54b6ff..d0fc103f72 100644
> --- a/tests/acceptance/virtiofs_submounts.py
> +++ b/tests/acceptance/virtiofs_submounts.py
> @@ -10,7 +10,6 @@ from avocado_qemu import wait_for_console_pattern
>   from avocado.utils import ssh
>   
>   from qemu.accel import kvm_available
> -from qemu.utils import get_info_usernet_hostfwd_port
>   
>   
>   def run_cmd(args):
> @@ -76,41 +75,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)
>
Willian Rampazzo Feb. 15, 2021, 7:15 p.m. UTC | #2 
On Wed, Feb 3, 2021 at 2:24 PM Cleber Rosa <crosa@redhat.com> 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 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>
> ---
>  tests/acceptance/avocado_qemu/__init__.py | 49 ++++++++++++++++++++++-
>  tests/acceptance/linux_ssh_mips_malta.py  | 38 ++----------------
>  tests/acceptance/virtiofs_submounts.py    | 36 -----------------
>  3 files changed, 51 insertions(+), 72 deletions(-)
>

Here I bring back my comment from patch 14, having a common place for
test specific classes would bring a better organization to the code.
Anyway,

Reviewed-by: Willian Rampazzo <willianr@redhat.com>
diff mbox series

Patch

diff --git a/tests/acceptance/avocado_qemu/__init__.py b/tests/acceptance/avocado_qemu/__init__.py
index b06692a59d..f0649e5011 100644
--- a/tests/acceptance/avocado_qemu/__init__.py
+++ b/tests/acceptance/avocado_qemu/__init__.py
@@ -10,6 +10,7 @@ 
 
 import logging
 import os
+import re
 import sys
 import uuid
 import tempfile
@@ -19,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
 
@@ -40,6 +42,8 @@  else:
 sys.path.append(os.path.join(SOURCE_DIR, 'python'))
 
 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)
@@ -215,7 +219,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
diff --git a/tests/acceptance/linux_ssh_mips_malta.py b/tests/acceptance/linux_ssh_mips_malta.py
index 1742235758..3f590a081f 100644
--- a/tests/acceptance/linux_ssh_mips_malta.py
+++ b/tests/acceptance/linux_ssh_mips_malta.py
@@ -12,7 +12,7 @@  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
@@ -21,7 +21,7 @@  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 +72,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: %d", port)
-        self.ssh_session = ssh.Session(self.VM_IP, port=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 +95,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')
diff --git a/tests/acceptance/virtiofs_submounts.py b/tests/acceptance/virtiofs_submounts.py
index 25ea54b6ff..d0fc103f72 100644
--- a/tests/acceptance/virtiofs_submounts.py
+++ b/tests/acceptance/virtiofs_submounts.py
@@ -10,7 +10,6 @@  from avocado_qemu import wait_for_console_pattern
 from avocado.utils import ssh
 
 from qemu.accel import kvm_available
-from qemu.utils import get_info_usernet_hostfwd_port
 
 
 def run_cmd(args):
@@ -76,41 +75,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)