[gnome-software/eos-updater-upstream: 5/7] eos-updater: Add mock eos-updater implementation

[Philip Withnall <pwithnall src gnome org>]

commit 08f5ac6729ca854427405e2b432eae82c614d2ed
Author: Philip Withnall <withnall endlessm com>
Date:   Thu Mar 28 15:24:47 2019 +0000

    eos-updater: Add mock eos-updater implementation
    
    This is copied from its canonical upstream source at
    https://github.com/endlessm/eos-updater/blob/master/eos-updater/tests/mock-eos-updater.py.
    
    Please do not make changes to it here without also submitting those
    changes upstream.
    
    It allows the gnome-software eos-updater plugin to be tested by
    providing a mock implementation of the eos-updater service. See the
    documentation comments at the top of the file for instructions on how to
    run it and drive its state machine.
    
    Signed-off-by: Philip Withnall <withnall endlessm com>

 plugins/eos-updater/tests/mock-eos-updater.py | 384 ++++++++++++++++++++++++++
 1 file changed, 384 insertions(+)
---
diff --git a/plugins/eos-updater/tests/mock-eos-updater.py b/plugins/eos-updater/tests/mock-eos-updater.py
new file mode 100644
index 00000000..90ec71ea
--- /dev/null
+++ b/plugins/eos-updater/tests/mock-eos-updater.py
@@ -0,0 +1,384 @@
+'''eos-updater mock template
+
+This creates a mock eos-updater interface (com.endlessm.Updater), with several
+methods on the Mock sidecar interface which allow its internal state flow to be
+controlled.
+
+A typical call chain for this would be:
+ - Test harness calls SetPollAction('update', {}, '', '')
+ - SUT calls Poll()
+ - Test harness calls FinishPoll()
+ - SUT calls Fetch()
+ - Test harness calls FinishFetch()
+ - SUT calls Apply()
+ - Test harness calls FinishApply()
+
+Errors can be simulated by specifying an `early-error` or `late-error` as the
+action in a Set*Action() call. `early-error` will result in the associated
+Poll() call (for example) transitioning to the error state. `late-error` will
+result in a transition to the error state only once (for example) FinishPoll()
+is called.
+
+See the implementation of each Set*Action() method for the set of actions it
+supports.
+
+Usage:
+   python3 -m dbusmock \
+       --template ./plugins/eos-updater/tests/mock-eos-updater.py
+'''
+
+# This program is free software; you can redistribute it and/or modify it under
+# the terms of the GNU Lesser General Public License as published by the Free
+# Software Foundation; either version 2.1 of the License, or (at your option)
+# any later version.  See http://www.gnu.org/copyleft/lgpl.html for the full
+# text of the license.
+#
+# The LGPL 2.1+ has been chosen as that’s the license eos-updater is under.
+
+from enum import IntEnum
+import time
+
+import dbus
+from dbusmock import MOCK_IFACE
+
+
+__author__ = 'Philip Withnall'
+__email__ = 'withnall endlessm com'
+__copyright__ = '© 2019 Endless Mobile Inc.'
+__license__ = 'LGPL 2.1+'
+
+
+class UpdaterState(IntEnum):
+    NONE = 0
+    READY = 1
+    ERROR = 2
+    POLLING = 3
+    UPDATE_AVAILABLE = 4
+    FETCHING = 5
+    UPDATE_READY = 6
+    APPLYING_UPDATE = 7
+    UPDATE_APPLIED = 8
+
+
+BUS_NAME = 'com.endlessm.Updater'
+MAIN_OBJ = '/com/endlessm/Updater'
+MAIN_IFACE = 'com.endlessm.Updater'
+SYSTEM_BUS = True
+
+
+def load(mock, parameters):
+    mock.AddProperties(
+        MAIN_IFACE,
+        dbus.Dictionary({
+            'State': dbus.UInt32(parameters.get('State', 1)),
+            'UpdateID': dbus.String(parameters.get('UpdateID', '')),
+            'UpdateRefspec': dbus.String(parameters.get('UpdateRefspec', '')),
+            'OriginalRefspec':
+                dbus.String(parameters.get('OriginalRefspec', '')),
+            'CurrentID': dbus.String(parameters.get('CurrentID', '')),
+            'UpdateLabel': dbus.String(parameters.get('UpdateLabel', '')),
+            'UpdateMessage': dbus.String(parameters.get('UpdateMessage', '')),
+            'Version': dbus.String(parameters.get('Version', '')),
+            'DownloadSize': dbus.Int64(parameters.get('DownloadSize', 0)),
+            'DownloadedBytes':
+                dbus.Int64(parameters.get('DownloadedBytes', 0)),
+            'UnpackedSize': dbus.Int64(parameters.get('UnpackedSize', 0)),
+            'FullDownloadSize':
+                dbus.Int64(parameters.get('FullDownloadSize', 0)),
+            'FullUnpackedSize':
+                dbus.Int64(parameters.get('FullUnpackedSize', 0)),
+            'ErrorCode': dbus.UInt32(parameters.get('ErrorCode', 0)),
+            'ErrorName': dbus.String(parameters.get('ErrorName', '')),
+            'ErrorMessage': dbus.String(parameters.get('ErrorMessage', '')),
+        }, signature='sv'))
+
+    # Set up initial state
+    mock.__poll_action = 'no-update'
+    mock.__fetch_action = 'success'
+    mock.__apply_action = 'success'
+
+    # Set up private methods
+    mock.__set_properties = __set_properties
+    mock.__change_state = __change_state
+    mock.__set_error = __set_error
+    mock.__check_state = __check_state
+
+
+#
+# Internal utility methods
+#
+
+# Values in @properties must have variant_level≥1
+def __set_properties(self, iface, properties):
+    for key, value in properties.items():
+        self.props[iface][key] = value
+    self.EmitSignal(dbus.PROPERTIES_IFACE, 'PropertiesChanged', 'sa{sv}as', [
+        iface,
+        properties,
+        [],
+    ])
+
+
+def __change_state(self, new_state):
+    props = {
+        'State': dbus.UInt32(new_state, variant_level=1)
+    }
+
+    # Reset error state if necessary.
+    if new_state != UpdaterState.ERROR and \
+       self.props[MAIN_IFACE]['ErrorName'] != '':
+        props['ErrorCode'] = dbus.UInt32(0, variant_level=1)
+        props['ErrorName'] = dbus.String('', variant_level=1)
+        props['ErrorMessage'] = dbus.String('', variant_level=1)
+
+    self.__set_properties(self, MAIN_IFACE, props)
+    self.EmitSignal(MAIN_IFACE, 'StateChanged', 'u', [dbus.UInt32(new_state)])
+
+
+def __set_error(self, error_name, error_message):
+    assert(error_name != '')
+
+    self.__set_properties(self, MAIN_IFACE, {
+        'ErrorName': dbus.String(error_name, variant_level=1),
+        'ErrorMessage': dbus.String(error_message, variant_level=1),
+        'ErrorCode': dbus.UInt32(1, variant_level=1),
+    })
+    self.__change_state(self, UpdaterState.ERROR)
+
+
+def __check_state(self, allowed_states):
+    if self.props[MAIN_IFACE]['State'] not in allowed_states:
+        raise dbus.exceptions.DBusException(
+            'Call not allowed in this state',
+            name='com.endlessm.Updater.Error.WrongState')
+
+
+#
+# Updater methods which are too big for squeezing into AddMethod()
+#
+
+@dbus.service.method(MAIN_IFACE, in_signature='', out_signature='')
+def Poll(self):
+    self.__check_state(self, set([
+        UpdaterState.READY,
+        UpdaterState.UPDATE_AVAILABLE,
+        UpdaterState.UPDATE_READY,
+        UpdaterState.ERROR,
+    ]))
+
+    self.__change_state(self, UpdaterState.POLLING)
+
+    if self.__poll_action == 'early-error':
+        time.sleep(0.5)
+        self.__set_error(self, self.__poll_error_name,
+                         self.__poll_error_message)
+    else:
+        # we now expect the test harness to call FinishPoll() on the mock
+        # interface
+        pass
+
+
+@dbus.service.method(MAIN_IFACE, in_signature='s', out_signature='')
+def PollVolume(self, path):
+    # FIXME: Currently unsupported
+    return self.Poll()
+
+
+@dbus.service.method(MAIN_IFACE, in_signature='', out_signature='')
+def Fetch(self):
+    return self.FetchFull()
+
+
+@dbus.service.method(MAIN_IFACE, in_signature='a{sv}', out_signature='')
+def FetchFull(self, options=None):
+    self.__check_state(self, set([UpdaterState.UPDATE_AVAILABLE]))
+
+    self.__change_state(self, UpdaterState.FETCHING)
+
+    if self.__fetch_action == 'early-error':
+        time.sleep(0.5)
+        self.__set_error(self, self.__fetch_error_name,
+                         self.__fetch_error_message)
+    else:
+        # we now expect the test harness to call FinishFetch() on the mock
+        # interface
+        pass
+
+
+@dbus.service.method(MAIN_IFACE, in_signature='', out_signature='')
+def Apply(self):
+    self.__check_state(self, set([UpdaterState.UPDATE_READY]))
+
+    self.__change_state(self, UpdaterState.APPLYING_UPDATE)
+
+    if self.__apply_action == 'early-error':
+        time.sleep(0.5)
+        self.__set_error(self, self.__apply_error_name,
+                         self.__apply_error_message)
+    else:
+        # we now expect the test harness to call FinishApply() on the mock
+        # interface
+        pass
+
+
+@dbus.service.method(MAIN_IFACE, in_signature='', out_signature='')
+def Cancel(self):
+    self.__check_state(self, set([
+        UpdaterState.POLLING,
+        UpdaterState.FETCHING,
+        UpdaterState.APPLYING_UPDATE,
+    ]))
+
+    time.sleep(1)
+    self.__set_error(self, 'com.endlessm.Updater.Error.Cancelled',
+                     'Update was cancelled')
+
+
+#
+# Convenience methods on the mock
+#
+
+@dbus.service.method(MOCK_IFACE, in_signature='sa{sv}ss', out_signature='')
+def SetPollAction(self, action, update_properties, error_name, error_message):
+    '''Set the action to happen when the SUT calls Poll().
+
+    This sets the action which will happen when Poll() (and subsequently
+    FinishPoll()) are called, including the details of the error which will be
+    returned or the new update which will be advertised.
+    '''
+    # Provide a default update.
+    if not update_properties:
+        update_properties = {
+            'UpdateID': dbus.String('f' * 64, variant_level=1),
+            'UpdateRefspec':
+                dbus.String('remote:new-refspec', variant_level=1),
+            'OriginalRefspec':
+                dbus.String('remote:old-refspec', variant_level=1),
+            'CurrentID': dbus.String('1' * 64, variant_level=1),
+            'UpdateLabel': dbus.String('New OS Update', variant_level=1),
+            'UpdateMessage':
+                dbus.String('Some release notes.', variant_level=1),
+            'Version': dbus.String('3.7.0', variant_level=1),
+            'DownloadSize': dbus.Int64(1000000000, variant_level=1),
+            'UnpackedSize': dbus.Int64(1500000000, variant_level=1),
+            'FullDownloadSize': dbus.Int64(1000000000 * 0.8, variant_level=1),
+            'FullUnpackedSize': dbus.Int64(1500000000 * 0.8, variant_level=1),
+        }
+
+    self.__poll_action = action
+    self.__poll_update_properties = update_properties
+    self.__poll_error_name = error_name
+    self.__poll_error_message = error_message
+
+
+@dbus.service.method(MOCK_IFACE, in_signature='', out_signature='')
+def FinishPoll(self):
+    self.__check_state(self, set([UpdaterState.POLLING]))
+
+    if self.__poll_action == 'no-update':
+        self.__change_state(self, UpdaterState.READY)
+    elif self.__poll_action == 'update':
+        assert(set([
+            'UpdateID',
+            'UpdateRefspec',
+            'OriginalRefspec',
+            'CurrentID',
+            'UpdateLabel',
+            'UpdateMessage',
+            'Version',
+            'FullDownloadSize',
+            'FullUnpackedSize',
+            'DownloadSize',
+            'UnpackedSize',
+        ]) <= set(self.__poll_update_properties.keys()))
+
+        # Set the initial DownloadedBytes based on whether we know the full
+        # download size.
+        props = self.__poll_update_properties
+        if props['DownloadSize'] < 0:
+            props['DownloadedBytes'] = dbus.Int64(-1, variant_level=1)
+        else:
+            props['DownloadedBytes'] = dbus.Int64(0, variant_level=1)
+
+        self.__set_properties(self, MAIN_IFACE, props)
+        self.__change_state(self, UpdaterState.UPDATE_AVAILABLE)
+    elif self.__poll_action == 'early-error':
+        # Handled in Poll() itself.
+        pass
+    elif self.__poll_action == 'late-error':
+        self.__set_error(self, self.__poll_error_name,
+                         self.__poll_error_message)
+    else:
+        assert(False)
+
+
+@dbus.service.method(MOCK_IFACE, in_signature='sss', out_signature='')
+def SetFetchAction(self, action, error_name, error_message):
+    '''Set the action to happen when the SUT calls Fetch().
+
+    This sets the action which will happen when Fetch() (and subsequently
+    FinishFetch()) are called, including the details of the error which will be
+    returned, if applicable.
+    '''
+    self.__fetch_action = action
+    self.__fetch_error_name = error_name
+    self.__fetch_error_message = error_message
+
+
+@dbus.service.method(MOCK_IFACE, in_signature='', out_signature='')
+def FinishFetch(self):
+    self.__check_state(self, set([UpdaterState.FETCHING]))
+
+    if self.__fetch_action == 'success':
+        # Simulate the download.
+        download_size = self.props[MAIN_IFACE]['DownloadSize']
+        for i in range(0, 100):
+            # Allow cancellation.
+            if self.props[MAIN_IFACE]['State'] != UpdaterState.FETCHING:
+                return
+
+            downloaded_bytes = (i / 100.0) * download_size
+            self.__set_properties(self, MAIN_IFACE, {
+                'DownloadedBytes':
+                    dbus.Int64(downloaded_bytes, variant_level=1),
+            })
+            time.sleep(0.1)
+
+        self.__change_state(self, UpdaterState.UPDATE_READY)
+    elif self.__fetch_action == 'early-error':
+        # Handled in Fetch() itself.
+        pass
+    elif self.__fetch_action == 'late-error':
+        self.__set_error(self, self.__fetch_error_name,
+                         self.__fetch_error_message)
+    else:
+        assert(False)
+
+
+@dbus.service.method(MOCK_IFACE, in_signature='sss', out_signature='')
+def SetApplyAction(self, action, error_name, error_message):
+    '''Set the action to happen when the SUT calls Apply().
+
+    This sets the action which will happen when Apply() (and subsequently
+    FinishApply()) are called, including the details of the error which will be
+    returned, if applicable.
+    '''
+    self.__apply_action = action
+    self.__apply_error_name = error_name
+    self.__apply_error_message = error_message
+
+
+@dbus.service.method(MOCK_IFACE, in_signature='', out_signature='')
+def FinishApply(self):
+    self.__check_state(self, set([UpdaterState.APPLYING_UPDATE]))
+
+    if self.__apply_action == 'success':
+        self.__change_state(self, UpdaterState.UPDATE_APPLIED)
+    elif self.__apply_action == 'early-error':
+        # Handled in Apply() itself.
+        pass
+    elif self.__apply_action == 'late-error':
+        self.__set_error(self, self.__apply_error_name,
+                         self.__apply_error_message)
+    else:
+        assert(False)