diff options
Diffstat (limited to 'plugins/eos-updater/tests/eos_updater.py')
| -rw-r--r-- | plugins/eos-updater/tests/eos_updater.py | 414 |
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) |