From 7363c4b20308e91a8ff01362ece9d5dfb157bca2 Mon Sep 17 00:00:00 2001 From: Sergey Orlov Date: Wed, 20 Jan 2021 18:21:22 +0100 Subject: [PATCH] ipatests: add a tests-oriented wrapper for pexpect module The pexpect module can be used for controlling and testing interactive command-line programs. The wrapper adds testing-oriented features like logging and automatic process termination and default check for process exit status. Reviewed-By: Alexander Bokovoy --- freeipa.spec.in | 2 + ipatests/pytest_ipa/integration/expect.py | 153 +++++++++++++++++++ ipatests/pytest_ipa/integration/host.py | 3 + ipatests/pytest_ipa/integration/transport.py | 9 ++ ipatests/setup.py | 1 + pylint_plugins.py | 1 + 6 files changed, 169 insertions(+) create mode 100644 ipatests/pytest_ipa/integration/expect.py diff --git a/freeipa.spec.in b/freeipa.spec.in index 8f43c261b..7d04f260d 100755 --- a/freeipa.spec.in +++ b/freeipa.spec.in @@ -331,6 +331,7 @@ BuildRequires: python3-lxml BuildRequires: python3-netaddr >= %{python_netaddr_version} BuildRequires: python3-netifaces BuildRequires: python3-paste +BuildRequires: python3-pexpect BuildRequires: python3-pki >= %{pki_version} BuildRequires: python3-polib BuildRequires: python3-pyasn1 @@ -845,6 +846,7 @@ Requires: python3-ipaserver = %{version}-%{release} Requires: iptables Requires: python3-coverage Requires: python3-cryptography >= 1.6 +Requires: python3-pexpect %if 0%{?fedora} # These packages do not exist on RHEL and for ipatests use # they are installed on the controller through other means diff --git a/ipatests/pytest_ipa/integration/expect.py b/ipatests/pytest_ipa/integration/expect.py new file mode 100644 index 000000000..85350a106 --- /dev/null +++ b/ipatests/pytest_ipa/integration/expect.py @@ -0,0 +1,153 @@ +import time +import logging + +import pexpect +from pexpect.exceptions import ExceptionPexpect, TIMEOUT + + +logger = logging.getLogger(__name__) +logger.setLevel(logging.DEBUG) + + +class IpaTestExpect(pexpect.spawn): + """A wrapper class around pexpect.spawn for easier usage in automated tests + + Please see pexpect documentation at + https://pexpect.readthedocs.io/en/stable/api/index.html for general usage + instructions. Note that usage of "+", "*" and '?' at the end of regular + expressions arguments to .expect() is meaningless. + + This wrapper adds ability to use the class as a context manager, which + will take care of verifying process return status and terminating + the process if it did not do it normally. The context manager is the + recommended way of using the class in tests. + Basic usage example: + + ``` + with IpaTestExpect('some_command') as e: + e.expect_exact('yes or no?') + e.sendline('yes') + ``` + + At exit from context manager the following checks are performed by default: + 1. there is nothing in output since last call to .expect() + 2. the process has terminated + 3. return code is 0 + + If any check fails, an exceptio is raised. If you want to override checks + 1 and 3 you can call .expect_exit() explicitly: + + ``` + with IpaTestExpect('some_command') as e: + ... + e.expect_exit(ok_returncode=1, ignore_remaining_output=True) + ``` + + All .expect* methods are strict, meaning that if they do not find the + pattern in the output during given amount of time, the exception is raised. + So they can directly be used to verify output for presence of specific + strings. + + Another addition is .get_last_output() method which can be used get process + output from penultimate up to the last call to .expect(). The result can + be used for more complex checks which can not be expressed as simple + regexes, for example we can check for absence of string in output: + + ``` + with IpaTestExpect('some_command') as e: + ... + e.expect('All done') + output = e.get_last_output() + assert 'WARNING' not in output + ``` + """ + def __init__(self, argv, default_timeout=10, encoding='utf-8'): + if isinstance(argv, str): + command = argv + args = [] + else: + command = argv[0] + args = argv[1:] + super().__init__( + command, args, timeout=default_timeout, encoding=encoding, + echo=False + ) + + def expect_exit(self, timeout=-1, ok_returncode=0, raiseonerr=True, + ignore_remaining_output=False): + if timeout == -1: + timeout = self.timeout + wait_to_exit_until = time.time() + timeout + if not self.eof(): + self.expect(pexpect.EOF, timeout) + errors = [] + if not ignore_remaining_output and self.before.strip(): + errors.append('Unexpected output at program exit: {!r}' + .format(self.before)) + + while time.time() < wait_to_exit_until: + if not self.isalive(): + break + time.sleep(0.1) + else: + errors.append('Program did not exit after waiting for {} seconds' + .format(self.timeout)) + if (not self.isalive() and raiseonerr + and self.exitstatus != ok_returncode): + errors.append('Program exited with unexpected status {}' + .format(self.exitstatus)) + self.exit_checked = True + if errors: + raise ExceptionPexpect( + 'Program exited with an unexpected state:\n' + + '\n'.join(errors)) + + def send(self, s): + """Wrapper to provide logging input string""" + logger.debug('Sending %r', s) + return super().send(s) + + def expect_list(self, pattern_list, *args, **kwargs): + """Wrapper to provide logging output string and expected patterns""" + try: + result = super().expect_list(pattern_list, *args, **kwargs) + finally: + self._log_output(pattern_list) + return result + + def expect_exact(self, pattern_list, *args, **kwargs): + """Wrapper to provide logging output string and expected patterns""" + try: + result = super().expect_exact(pattern_list, *args, **kwargs) + finally: + self._log_output(pattern_list) + return result + + def get_last_output(self): + """Return output consumed by last call to .expect*()""" + output = self.before + if isinstance(self.after, str): + output += self.after + return output + + def _log_output(self, expected): + logger.debug('Output received: %r, expected: "%s", ', + self.get_last_output(), expected) + + def __enter__(self): + self.exit_checked = False + return self + + def __exit__(self, exc_type, exc_val, exc_tb): + exception_occurred = bool(exc_type) + try: + if not self.exit_checked: + self.expect_exit(raiseonerr=not exception_occurred, + ignore_remaining_output=exception_occurred) + except TIMEOUT: + if not exception_occurred: + raise + finally: + if self.isalive(): + logger.error('Command still active, terminating.') + self.terminate(True) diff --git a/ipatests/pytest_ipa/integration/host.py b/ipatests/pytest_ipa/integration/host.py index d1df228b1..bcadcbae2 100644 --- a/ipatests/pytest_ipa/integration/host.py +++ b/ipatests/pytest_ipa/integration/host.py @@ -204,6 +204,9 @@ class Host(pytest_multihost.host.Host): else: return result + def spawn_expect(self, argv, default_timeout=10, encoding='utf-8'): + """Run command on host using IpaTestExpect""" + return self.transport.spawn_expect(argv, default_timeout, encoding) class WinHost(pytest_multihost.host.WinHost): """ diff --git a/ipatests/pytest_ipa/integration/transport.py b/ipatests/pytest_ipa/integration/transport.py index d55e72ddb..f12df1225 100644 --- a/ipatests/pytest_ipa/integration/transport.py +++ b/ipatests/pytest_ipa/integration/transport.py @@ -7,6 +7,8 @@ Provides SSH password login for OpenSSH transport """ import os +from .expect import IpaTestExpect + from pytest_multihost.transport import OpenSSHTransport @@ -46,3 +48,10 @@ class IPAOpenSSHTransport(OpenSSHTransport): self.log.debug("SSH invocation: %s", argv) return argv + + def spawn_expect(self, argv, default_timeout, encoding): + self.log.debug('Starting pexpect ssh session') + if isinstance(argv, str): + argv = [argv] + argv = self._get_ssh_argv() + ['-t', '-q'] + argv + return IpaTestExpect(argv, default_timeout, encoding) diff --git a/ipatests/setup.py b/ipatests/setup.py index 8b89f90ca..46fb09958 100644 --- a/ipatests/setup.py +++ b/ipatests/setup.py @@ -73,6 +73,7 @@ if __name__ == '__main__': "pytest_multihost", "python-ldap", "six", + "pexpect", ], extras_require={ "integration": ["dbus-python", "pyyaml", "ipaserver"], diff --git a/pylint_plugins.py b/pylint_plugins.py index 82096055f..7e1b934e0 100644 --- a/pylint_plugins.py +++ b/pylint_plugins.py @@ -217,6 +217,7 @@ ipa_class_members = { 'put_file_contents', 'get_file_contents', 'ldap_connect', + {'spawn_expect': ['__enter__', '__exit__']}, ]}, 'replicas', 'clients',