test: Add test case for user-based confirmation

Signed-off-by: Vyacheslav Yurkov <[email protected]>

test/conftest.py

@@ -224,6 +224,28 @@ def rauc_dbus_install_failure(rauc_bundle):
     assert proc.isalive()
     assert proc.terminate(force=True)
 
+@pytest.fixture
+def install_confirmation_plugin():
+    """
+    Starts a test implementation of confirmation plugin, which is able to confirm or deny
+    confirmation requests
+    """
+    procs = []
+
+    def confirmation_plugin(confirm, code=0, details=''):
+        confirmed = "confirmed" if confirm else "denied"
+        proc = run_pexpect(f'{sys.executable} -m install_confirm --{confirmed} --error-code={code} --details="{details}"',
+                           cwd=os.path.dirname(__file__), timeout=None)
+
+        proc.expect('Confirmation interface published')
+        procs.append(proc)
+
+    yield confirmation_plugin
+
+    for proc in procs:
+        assert proc.isalive()
+        proc.terminate(force=True)
+
 @pytest.fixture(scope='session')
 def nginx_config(tmp_path_factory):
     """

test/test_install.py

@@ -21,6 +21,11 @@ def _install_config(mode):
 
     return _install_config
 
+@pytest.fixture
+def confirm_workflow_hawkbit(hawkbit):
+    hawkbit.set_config('user.confirmation.flow.enabled', True)
+    yield
+    hawkbit.set_config('user.confirmation.flow.enabled', False)
 
 @pytest.mark.parametrize('mode', ('download', 'streaming'))
 def test_install_bundle_no_dbus_iface(hawkbit, install_config, bundle_assigned, mode):
@@ -86,6 +91,83 @@ def test_install_failure(hawkbit, install_config, bundle_assigned, rauc_dbus_ins
     assert status[0]['type'] == 'error'
     assert 'Failed to install software bundle.' in status[0]['messages']
 
+@pytest.mark.parametrize('mode', ('download', 'streaming'))
+@pytest.mark.parametrize("confirm", [True])
+def test_install_confirmation_given(hawkbit, adjust_config, install_config, mode,
+                                    confirm_workflow_hawkbit, bundle_assigned, rauc_dbus_install_success,
+                                    install_confirmation_plugin, confirm):
+    """
+    Enable user confirmation in Hawkbit config. Make sure we can approve requested installation
+    and the result of installation is received correctly by hawkBit.
+    """
+    import re
+
+    if mode == 'streaming':
+        config = adjust_config(
+            {'client': {'stream_bundle': 'true', 'require_confirmation': 'true'}},
+            remove={'client': 'bundle_download_location'})
+    else:
+        config = adjust_config({'client':{'require_confirmation': 'true'}})
+
+    confirmed_regex = re.compile("Action .* confirmed")
+
+    plugin = install_confirmation_plugin(confirm)
+    out, err, exitcode = run(f'rauc-hawkbit-updater -c "{config}" -r')
+
+    assert 'New software ready for download' in out
+    assert confirmed_regex.findall(out)
+    assert 'Software bundle installed successfully.' in out
+
+    status = hawkbit.get_action_status()
+    assert status[0]['type'] == 'finished'
+
+denied_message = "Denied because we are not interested in the update"
+@pytest.mark.parametrize("confirm", [False])
+@pytest.mark.parametrize("error_code", [-120])
+@pytest.mark.parametrize("details", [denied_message])
+def test_install_confirmation_denied(hawkbit, adjust_config, confirm_workflow_hawkbit, bundle_assigned,
+                                     install_confirmation_plugin, confirm, error_code, details):
+    """
+    Enable user confirmation in Hawkbit config. Deny requested installation
+    and check that hawkBit received the feedback.
+    """
+    import re
+
+    config = adjust_config({'client':{'require_confirmation': 'true'}})
+    confirmed_regex = re.compile("Action .* denied")
+
+    plugin = install_confirmation_plugin(confirm, error_code, details)
+    out, err, exitcode = run(f'rauc-hawkbit-updater -c "{config}" -r')
+
+    assert confirmed_regex.findall(out)
+    # make sure the response came from the plugin
+    assert 'GDBus.Error:org.freedesktop.DBus.Error.ServiceUnknown' not in out
+
+    status = hawkbit.get_action_status()
+    # If the confirmation is rejected, the action continues to be in state WAITING_FOR_CONFIRMATION
+    # until it is confirmed at a later point in time or cancelled.
+    assert status[0]['type'] == 'wait_for_confirmation'
+    assert f'Confirmation status code: {error_code}' in status[0]['messages']
+    assert denied_message in status[0]['messages']
+
+def test_install_confirmation_denied_no_plugin(hawkbit, adjust_config, confirm_workflow_hawkbit, bundle_assigned):
+    """
+    Enable user confirmation in Hawkbit config.
+    Don't start confirmation plugin. This is equivalent to the rejection of installation.
+    """
+    import re
+    config = adjust_config({'client':{'require_confirmation': 'true'}})
+    confirmed_regex = re.compile("Action .* denied")
+
+    out, err, exitcode = run(f'rauc-hawkbit-updater -c "{config}" -r')
+
+    status = hawkbit.get_action_status()
+    assert confirmed_regex.findall(out)
+    assert 'GDBus.Error:org.freedesktop.DBus.Error.ServiceUnknown' in err
+    # If the confirmation is rejected, the action continues to be in state WAITING_FOR_CONFIRMATION
+    # until it is confirmed at a later point in time or cancelled.
+    assert status[0]['type'] == 'wait_for_confirmation'
+
 @pytest.mark.parametrize('mode', ('download', 'streaming'))
 def test_install_maintenance_window(hawkbit, install_config, rauc_bundle, assign_bundle,
                                     rauc_dbus_install_success, mode):