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 <abokovoy@redhat.com>

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

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)

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):
     """

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)

ipatests/setup.py

@@ -73,6 +73,7 @@ if __name__ == '__main__':
             "pytest_multihost",
             "python-ldap",
             "six",
+            "pexpect",
         ],
         extras_require={
             "integration": ["dbus-python", "pyyaml", "ipaserver"],

pylint_plugins.py

@@ -217,6 +217,7 @@ ipa_class_members = {
             'put_file_contents',
             'get_file_contents',
             'ldap_connect',
+            {'spawn_expect': ['__enter__', '__exit__']},
         ]},
         'replicas',
         'clients',