summaryrefslogtreecommitdiffstats
path: root/plugins/eos-updater/tests/eos_updater.py
diff options
context:
space:
mode:
Diffstat (limited to '')
-rw-r--r--plugins/eos-updater/tests/eos_updater.py414
1 files changed, 414 insertions, 0 deletions
diff --git a/plugins/eos-updater/tests/eos_updater.py b/plugins/eos-updater/tests/eos_updater.py
new file mode 100644
index 0000000..5e4aa8d
--- /dev/null
+++ b/plugins/eos-updater/tests/eos_updater.py
@@ -0,0 +1,414 @@
+'''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
+from gi.repository import GLib
+import time
+
+import dbus
+import dbus.mainloop.glib
+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
+
+
+dbus.mainloop.glib.DBusGMainLoop(set_as_default=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='',
+ async_callbacks=('success_cb', 'error_cb'))
+def FinishFetch(self, success_cb, error_cb):
+ '''Finish a pending client call to Fetch().
+
+ This is implemented using async_callbacks since if the fetch action is
+ ‘success’ it will block until the simulated download is complete, emitting
+ download progress signals throughout. As it’s implemented asynchronously,
+ this allows any calls to Cancel() to be handled by the mock service
+ part-way through the fetch.
+ '''
+ self.__check_state(self, set([UpdaterState.FETCHING]))
+
+ if self.__fetch_action == 'success':
+ # Simulate the download.
+ i = 0
+ download_size = self.props[MAIN_IFACE]['DownloadSize']
+
+ def _download_progress_cb():
+ nonlocal i
+
+ # Allow cancellation.
+ if self.props[MAIN_IFACE]['State'] != UpdaterState.FETCHING:
+ return False
+
+ downloaded_bytes = (i / 100.0) * download_size
+ self.__set_properties(self, MAIN_IFACE, {
+ 'DownloadedBytes':
+ dbus.Int64(downloaded_bytes, variant_level=1),
+ })
+
+ i += 1
+
+ # Keep looping until the download is complete.
+ if i <= 100:
+ return True
+
+ # When the download is complete, change the service state and
+ # finish the asynchronous FinishFetch() call.
+ self.__change_state(self, UpdaterState.UPDATE_READY)
+ success_cb()
+ return False
+
+ GLib.timeout_add(100, _download_progress_cb)
+ elif self.__fetch_action == 'early-error':
+ # Handled in Fetch() itself.
+ success_cb()
+ elif self.__fetch_action == 'late-error':
+ self.__set_error(self, self.__fetch_error_name,
+ self.__fetch_error_message)
+ success_cb()
+ 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)