1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
|
#
# Ubuntu OTA Tests
# Copyright (C) 2015 Canonical
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
#
import dbus
import logging
from autopilot.matchers import Eventually
from dbus.mainloop.glib import DBusGMainLoop
from testtools import matchers
from ubuntu_ota_tests import system
from ubuntu_ota_tests.reactors import ApplyUpdateReactor
from ubuntu_ota_tests.reactors import DownloadReactor
from ubuntu_ota_tests.reactors import CheckForUpdateReactor
logger = logging.getLogger(__name__)
# For some reason the process name is 'system-image-db' on the device.
SYS_IMG_PROC_NAME = 'system-image-db'
# Is there a better mechanism that we can use to ensure that DBusGMainLoop
# isn't called more than once?
_loop_running = False
def ensure_loop_ready():
global _loop_running
if not _loop_running:
DBusGMainLoop(set_as_default=True)
_loop_running = True
def get_system_image_interface():
"""Return a dbus Interface for SystemImage Service.
Note calling this method will cause system-image-dbus to start (if it's not
already started).
"""
ensure_loop_ready()
system_bus = dbus.SystemBus()
service = system_bus.get_object('com.canonical.SystemImage', '/Service')
iface = dbus.Interface(service, 'com.canonical.SystemImage')
return iface
def ensure_system_image_dbus_not_running():
sys_image_iface = get_system_image_interface()
try:
sys_image_iface.Exit()
_ensure_system_image_process_not_running()
except dbus.DBusException:
logger.debug("Attempted to exit system dbus when not running.")
def _ensure_system_image_process_not_running():
"""Raise RuntimeError if process is still running after 10 seconds."""
result = Eventually(
matchers.Equals(False)
).match(_is_system_image_process_running)
if result is not None:
raise RuntimeError(
'system image dbus process still running after 10.0 seconds')
def _is_system_image_process_running():
"""Return True if system-image-dbus process is running False otherwise."""
try:
return system.get_process_by_name(SYS_IMG_PROC_NAME) is not None
except OSError:
return False
def get_current_build_number():
"""Gets the build number of the current image
:returns: current build number as integer
"""
sys_image_iface = get_system_image_interface()
info = sys_image_iface.Information()
return int(info['current_build_number'])
def set_auto_download(on):
"""Sets the value of auto_download in the system image client.
:param on: Whether auto download should be on or off
"""
# The system image client only accepts strings for SetSetting,
# with '2' representing On and '0' Off
auto_download = '2' if on else '0'
sys_image_iface = get_system_image_interface()
sys_image_iface.SetSetting('auto_download', auto_download)
def get_auto_download():
"""Gets the current value of auto_download in the system image client
:returns: True if on, False if off
"""
sys_image_iface = get_system_image_interface()
# We get back either '2' or '0' (strings) so have to convert first to
# Integer then Boolean
return bool(int(sys_image_iface.GetSetting('auto_download')))
def check_for_update():
"""Check for whether updates are available.
:return: namedtuple containing the following attributes:
- is_available: flag indicating whether an update is available
- downloading: flag indicating whether a download is in progress
- available_version: string specifying update target candidate
version, or 'n/a' if no update is available
- update_size: integer providing total size in bytes for the update
- last_update_date: ISO 8601 format UTC date string that the last
update was applied to this device
- error_reason: a string containing the reason the check or download
failed, or the empty string on success
"""
iface = get_system_image_interface()
reactor = CheckForUpdateReactor(iface)
result = reactor.run()
assert len(result) == 1, (
'Unexpected signal count: {}'.format(len(result)))
return result[0]
def apply_update(timeout=180):
"""Calls the ApplyUpdate method on the interface using async reactor
If the ApplyUpdate call is successful the system should reboot, first
the Rebooting signal is sent but this may not be caught depending on the
speed of the device. If something goes wrong or there is no available
update to apply the Rebooting signal is received with the status field
set to False
:param timeout: timeout for the Rebboting signal to be received
:returns: object from the run execution of the related reactor, it has a
status boolean field which should reflect how the apply update
call went
"""
sys_image_iface = get_system_image_interface()
reactor = ApplyUpdateReactor(sys_image_iface)
result = reactor.run(timeout=timeout)
assert len(result) == 1, (
'Unexpected signal count: {}'.format(len(result)))
return result[0]
def download_update(timeout=300):
"""Call DownloadUpdate dbus method and wait for UpdateDownloaded signal.
This will trigger the download to start and wait for completion or failure
signal.
:param timeout: Timeout for the download operation to complete in.
:return: List of received signals.
"""
sys_image_iface = get_system_image_interface()
reactor = DownloadReactor(sys_image_iface)
return reactor.run(timeout=timeout)
|