| Message ID | 20210323221539.3532660-5-crosa@redhat.com (mailing list archive) |
|---|---|
| State | New, archived |
| Headers | show |
| Series | Acceptance Test: introduce base class for Linux based tests | expand |
Hi Cleber, On 3/23/21 11:15 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 nit: strictly speaking they are moved to another class which is inherited by LinuxTest, right? > 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> > --- > tests/acceptance/avocado_qemu/__init__.py | 48 ++++++++++++++++++++++- > tests/acceptance/linux_ssh_mips_malta.py | 38 ++---------------- > tests/acceptance/virtiofs_submounts.py | 37 ----------------- > 3 files changed, 50 insertions(+), 73 deletions(-) > > 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..3f590a081f 100644 > --- a/tests/acceptance/linux_ssh_mips_malta.py > +++ b/tests/acceptance/linux_ssh_mips_malta.py > @@ -12,7 +12,7 @@ > 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 qemu.utils import get_info_usernet_hostfwd_port Can't you remove this now? > > > -class LinuxSSH(Test): > +class LinuxSSH(Test, LinuxSSHMixIn): out of curiosity why can't it be migrated to a LinuxTest? > > timeout = 150 # Not for 'configure --enable-debug --enable-debug-tcg' > > @@ -72,41 +72,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 +95,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) > > Besides, Reviewed-by: Eric Auger <eric.auger@redhat.com> Thanks Eric
On Wed, Mar 24, 2021 at 10:07:31AM +0100, Auger Eric wrote: > Hi Cleber, > > On 3/23/21 11:15 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 > nit: strictly speaking they are moved to another class which is > inherited by LinuxTest, right? > > 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> > > --- > > tests/acceptance/avocado_qemu/__init__.py | 48 ++++++++++++++++++++++- > > tests/acceptance/linux_ssh_mips_malta.py | 38 ++---------------- > > tests/acceptance/virtiofs_submounts.py | 37 ----------------- > > 3 files changed, 50 insertions(+), 73 deletions(-) > > > > 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..3f590a081f 100644 > > --- a/tests/acceptance/linux_ssh_mips_malta.py > > +++ b/tests/acceptance/linux_ssh_mips_malta.py > > @@ -12,7 +12,7 @@ > > 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 qemu.utils import get_info_usernet_hostfwd_port > Can't you remove this now? > > > > Yes, good catch! > > -class LinuxSSH(Test): > > +class LinuxSSH(Test, LinuxSSHMixIn): > out of curiosity why can't it be migrated to a LinuxTest? LinuxTest (currently) assumes that it'll boot via an image obtained with avocado.utils.vmimage, and configured with avocado.utils.cloudinit. Those are not the case for this test. I believe there are still some opportunities for advancing them towards each other, but LinuxTest needs to become more versatile. > > > > timeout = 150 # Not for 'configure --enable-debug --enable-debug-tcg' > > > > @@ -72,41 +72,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 +95,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) > > > > > > Besides, > > Reviewed-by: Eric Auger <eric.auger@redhat.com> > > Thanks > > Eric Thanks for the review! - Cleber.
Hi Cleber, On 3/24/21 11:23 PM, Cleber Rosa wrote: > On Wed, Mar 24, 2021 at 10:07:31AM +0100, Auger Eric wrote: >> Hi Cleber, >> >> On 3/23/21 11:15 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 >> nit: strictly speaking they are moved to another class which is >> inherited by LinuxTest, right? >>> 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> >>> --- >>> tests/acceptance/avocado_qemu/__init__.py | 48 ++++++++++++++++++++++- >>> tests/acceptance/linux_ssh_mips_malta.py | 38 ++---------------- >>> tests/acceptance/virtiofs_submounts.py | 37 ----------------- >>> 3 files changed, 50 insertions(+), 73 deletions(-) >>> >>> 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..3f590a081f 100644 >>> --- a/tests/acceptance/linux_ssh_mips_malta.py >>> +++ b/tests/acceptance/linux_ssh_mips_malta.py >>> @@ -12,7 +12,7 @@ >>> 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 qemu.utils import get_info_usernet_hostfwd_port >> Can't you remove this now? >>> >>> > > Yes, good catch! > >>> -class LinuxSSH(Test): >>> +class LinuxSSH(Test, LinuxSSHMixIn): >> out of curiosity why can't it be migrated to a LinuxTest? > > LinuxTest (currently) assumes that it'll boot via an image obtained > with avocado.utils.vmimage, and configured with > avocado.utils.cloudinit. Those are not the case for this test. I > believe there are still some opportunities for advancing them towards > each other, but LinuxTest needs to become more versatile. OK makes sense. By the way would it be possible to launch other types of cloud-init images? I see that LinuxTest download_boot() only uses a fedora 31 image. I would be interested in being able to run other types of images. Could we make it configurable? I can work on this if it makes sense. Thanks Eric > >>> >>> timeout = 150 # Not for 'configure --enable-debug --enable-debug-tcg' >>> >>> @@ -72,41 +72,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 +95,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) >>> >>> >> >> Besides, >> >> Reviewed-by: Eric Auger <eric.auger@redhat.com> >> >> Thanks >> >> Eric > > Thanks for the review! > > - Cleber. >
Hi, On 3/25/21 9:50 AM, Auger Eric wrote: > Hi Cleber, > > On 3/24/21 11:23 PM, Cleber Rosa wrote: >> On Wed, Mar 24, 2021 at 10:07:31AM +0100, Auger Eric wrote: >>> Hi Cleber, >>> >>> On 3/23/21 11:15 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 >>> nit: strictly speaking they are moved to another class which is >>> inherited by LinuxTest, right? >>>> 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> >>>> --- >>>> tests/acceptance/avocado_qemu/__init__.py | 48 ++++++++++++++++++++++- >>>> tests/acceptance/linux_ssh_mips_malta.py | 38 ++---------------- >>>> tests/acceptance/virtiofs_submounts.py | 37 ----------------- >>>> 3 files changed, 50 insertions(+), 73 deletions(-) >>>> >>>> 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..3f590a081f 100644 >>>> --- a/tests/acceptance/linux_ssh_mips_malta.py >>>> +++ b/tests/acceptance/linux_ssh_mips_malta.py >>>> @@ -12,7 +12,7 @@ >>>> 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 qemu.utils import get_info_usernet_hostfwd_port >>> Can't you remove this now? >>>> >>>> >> Yes, good catch! >> >>>> -class LinuxSSH(Test): >>>> +class LinuxSSH(Test, LinuxSSHMixIn): >>> out of curiosity why can't it be migrated to a LinuxTest? >> LinuxTest (currently) assumes that it'll boot via an image obtained >> with avocado.utils.vmimage, and configured with >> avocado.utils.cloudinit. Those are not the case for this test. I >> believe there are still some opportunities for advancing them towards >> each other, but LinuxTest needs to become more versatile. > OK makes sense. > > By the way would it be possible to launch other types of cloud-init > images? I see that LinuxTest download_boot() only uses a fedora 31 > image. I would be interested in being able to run other types of images. > Could we make it configurable? I can work on this if it makes sense. It is a good idea! Thanks! - Wainer > > Thanks > > Eric >>>> >>>> timeout = 150 # Not for 'configure --enable-debug --enable-debug-tcg' >>>> >>>> @@ -72,41 +72,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 +95,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) >>>> >>>> >>> Besides, >>> >>> Reviewed-by: Eric Auger <eric.auger@redhat.com> >>> >>> Thanks >>> >>> Eric >> Thanks for the review! >> >> - Cleber. >>
On Wed, Mar 24, 2021 at 10:07:31AM +0100, Auger Eric wrote: > Hi Cleber, > > On 3/23/21 11:15 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 > nit: strictly speaking they are moved to another class which is > inherited by LinuxTest, right? I forgot to address this comment previously. Yes, you're right. I'll reword it. Thanks! - Cleber. > > 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> > > --- > > tests/acceptance/avocado_qemu/__init__.py | 48 ++++++++++++++++++++++- > > tests/acceptance/linux_ssh_mips_malta.py | 38 ++---------------- > > tests/acceptance/virtiofs_submounts.py | 37 ----------------- > > 3 files changed, 50 insertions(+), 73 deletions(-) > > > > 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..3f590a081f 100644 > > --- a/tests/acceptance/linux_ssh_mips_malta.py > > +++ b/tests/acceptance/linux_ssh_mips_malta.py > > @@ -12,7 +12,7 @@ > > 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 qemu.utils import get_info_usernet_hostfwd_port > Can't you remove this now? > > > > > > -class LinuxSSH(Test): > > +class LinuxSSH(Test, LinuxSSHMixIn): > out of curiosity why can't it be migrated to a LinuxTest? > > > > timeout = 150 # Not for 'configure --enable-debug --enable-debug-tcg' > > > > @@ -72,41 +72,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 +95,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) > > > > > > Besides, > > Reviewed-by: Eric Auger <eric.auger@redhat.com> > > Thanks > > Eric
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..3f590a081f 100644 --- a/tests/acceptance/linux_ssh_mips_malta.py +++ b/tests/acceptance/linux_ssh_mips_malta.py @@ -12,7 +12,7 @@ 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 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 @@ 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 +95,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)