test: add installation tests

Add test cases for successful/failing installations as well as failing
installation due to no RAUC D-Bus interface available.

Signed-off-by: Bastian Krause <bst@pengutronix.de>

test-requirements.txt

@@ -3,3 +3,4 @@ attrs
 requests
 pydbus
 pygobject
+pexpect

test/conftest.py

@@ -3,11 +3,13 @@
 # SPDX-FileCopyrightText: 2021 Bastian Krause <bst@pengutronix.de>, Pengutronix
 
 import os
+import sys
 from configparser import ConfigParser
 
 import pytest
 
 from hawkbit_mgmt import HawkbitMgmtTestClient, HawkbitError
+from helper import run_pexpect
 
 def pytest_addoption(parser):
     """Register custom argparse-style options."""
@@ -103,3 +105,64 @@ def _adjust_config(options={'client': {}}, remove={}):
         return config
 
     return _adjust_config
+
+@pytest.fixture(scope='session')
+def rauc_bundle(tmp_path_factory):
+    """Creates a temporary 512 KB file to be used as a dummy RAUC bundle."""
+    bundle = tmp_path_factory.mktemp('data') / 'bundle.raucb'
+    bundle.write_bytes(os.urandom(512)*1024)
+    return str(bundle)
+
+@pytest.fixture
+def bundle_assigned(hawkbit, hawkbit_target_added, rauc_bundle):
+    """
+    Creates a softwaremodule containing the file from the rauc_bundle fixture as an artifact.
+    Creates a distributionset from this softwaremodule. Assigns this distributionset to the target
+    created by the hawkbit_target_added fixture. Returns the corresponding action ID of this
+    assignment.
+    """
+    swmodule = hawkbit.add_softwaremodule()
+    artifact = hawkbit.add_artifact(rauc_bundle, swmodule)
+    distributionset = hawkbit.add_distributionset(swmodule)
+    action = hawkbit.assign_target(distributionset)
+
+    yield action
+
+    try:
+        hawkbit.cancel_action(action, hawkbit_target_added, force=True)
+    except HawkbitError:
+        pass
+
+    hawkbit.delete_distributionset(distributionset)
+    hawkbit.delete_artifact(artifact, swmodule)
+    hawkbit.delete_softwaremodule(swmodule)
+
+@pytest.fixture
+def rauc_dbus_install_success(rauc_bundle):
+    """
+    Creates a RAUC D-Bus dummy interface on the SessionBus mimicing a successful installation on
+    Install().
+    """
+    proc = run_pexpect(f'{sys.executable} -m rauc_dbus_dummy {rauc_bundle}',
+                       cwd=os.path.dirname(__file__))
+    proc.expect('Interface published')
+
+    yield
+
+    assert proc.isalive()
+    assert proc.terminate(force=True)
+
+@pytest.fixture
+def rauc_dbus_install_failure(rauc_bundle):
+    """
+    Creates a RAUC D-Bus dummy interface on the SessionBus mimicing a failing installation on
+    Install().
+    """
+    proc = run_pexpect(f'{sys.executable} -m rauc_dbus_dummy {rauc_bundle} --completed-code=1',
+                       cwd=os.path.dirname(__file__), timeout=None)
+    proc.expect('Interface published')
+
+    yield
+
+    assert proc.isalive()
+    assert proc.terminate(force=True)

test/helper.py

@@ -8,6 +8,46 @@
 import logging
 
 
+class PExpectLogger:
+    """
+    pexpect Logger, allows to use Python's logging stdlib. To be passed as pexpect 'logfile".
+    Logs linewise to given logger at given level.
+    """
+    def __init__(self, level=logging.INFO, logger=None):
+        self.level = level
+        self.data = b''
+        self.logger = logger if logger else logging.getLogger()
+
+    def write(self, data):
+        self.data += data
+
+    def flush(self):
+        for line in self.data.splitlines():
+            self.logger.log(self.level, line.decode())
+
+        self.data = b''
+
+
+def run_pexpect(command, *, timeout=30, cwd=None):
+    """
+    Runs given command via pexpect with DBUS_STARTER_BUS_TYPE=session and PATH+=./build. Returns
+    process handle immediately allowing further expect calls. Logs command (with updated env) and
+    its stdout/stderr/exit code.
+    """
+    import pexpect
+
+    logger = logging.getLogger(command.split()[0])
+
+    env = os.environ.copy()
+    env.update({'DBUS_STARTER_BUS_TYPE': 'session'})
+    env['PATH'] += f':{os.path.dirname(os.path.abspath(__file__))}/../build'
+
+    log_env = [ f'{key}={value}' for key, value in set(env.items()) - set(os.environ.items()) ]
+    logger.info('running: %s %s', ' '.join(log_env), command)
+
+    pexpect_log = PExpectLogger(logger=logger)
+    return pexpect.spawn(command, env=env, timeout=timeout, cwd=cwd, logfile=pexpect_log)
+
 def run(command, *, timeout=30):
     """
     Runs given command as subprocess with DBUS_STARTER_BUS_TYPE=session and PATH+=./build. Blocks

test/test_install.py

@@ -0,0 +1,52 @@
+# SPDX-License-Identifier: LGPL-2.1-only
+# SPDX-FileCopyrightText: 2021 Bastian Krause <bst@pengutronix.de>, Pengutronix
+
+from helper import run
+
+def test_install_bundle_no_dbus_iface(hawkbit, bundle_assigned, config):
+    """Assign bundle to target and test installation without RAUC D-Bus interface available."""
+    out, err, exitcode = run(f'rauc-hawkbit-updater -c "{config}" -r')
+
+    err_lines = err.splitlines()
+
+    assert 'New software ready for download' in out
+    assert 'Download complete' in out
+    assert err_lines.pop(0) == \
+            'WARNING: GDBus.Error:org.freedesktop.DBus.Error.ServiceUnknown: The name de.pengutronix.rauc was not provided by any .service files'
+    assert err_lines.pop(0) == 'WARNING: Failed to install software bundle.'
+    assert not err_lines
+    assert exitcode == 1
+
+    status = hawkbit.get_action_status()
+    assert status[0]['type'] == 'error'
+
+def test_install_success(hawkbit, config, bundle_assigned, rauc_dbus_install_success):
+    """
+    Assign bundle to target and test successful download and installation. Make sure installation
+    result is received correctly by hawkBit.
+    """
+    out, err, exitcode = run(f'rauc-hawkbit-updater -c "{config}" -r')
+
+    assert 'New software ready for download' in out
+    assert 'Download complete' in out
+    assert 'Software bundle installed successfully.' in out
+    assert err == ''
+    assert exitcode == 0
+
+    status = hawkbit.get_action_status()
+    assert status[0]['type'] == 'finished'
+
+def test_install_failure(hawkbit, config, bundle_assigned, rauc_dbus_install_failure):
+    """
+    Assign bundle to target and test successful download and failing installation. Make sure
+    installation result is received correctly by hawkBit.
+    """
+    out, err, exitcode = run(f'rauc-hawkbit-updater -c "{config}" -r')
+
+    assert 'New software ready for download' in out
+    assert err.strip() == 'WARNING: Failed to install software bundle.'
+    assert exitcode == 1
+
+    status = hawkbit.get_action_status()
+    assert status[0]['type'] == 'error'
+    assert 'Failed to install software bundle.' in status[0]['messages']