elisa.core.utils.update

71 72 """ 73 Helper Class for simple look up of updates on the remote elisa server. 74 """ 75 76 # check each day, the value is in seconds 77 check_interval = 86400 78

79 - def __init__(self, install_date, user_id, version, **extra_affiliation_params):

80 super(UpdateChecker, self).__init__() 81 82 # initial data 83 self.install_date = install_date 84 self.user_id = user_id 85 self.version = version 86 self.operating_system = misc.get_os_name() 87 self.extra_affiliation_params = extra_affiliation_params 88 89 self._pending_call = None 90 self._current_dfr = None

91

92 - def parse_result(self, result):

93 """ 94 Parse the given data into a dictionary. The syntax for result has to be 95 a key-value pair per line separated by a colon (':'). Spaces at the 96 beginning or the end are stripped from both the key and the value. 97 For instance:: 98 99 foo: bar 100 test: partial 101 maybe : maybe not 102 103 @param result: the result to parse 104 @type result: C{str} 105 106 @return: a dictionary containing the key-value pairs 107 @rtype: C{dict} 108 """ 109 result_dict = {} 110 for line in result.splitlines(): 111 try: 112 key, value = line.split(':', 1) 113 except ValueError: 114 self.warning("Could not split '%s'" % line) 115 continue 116 117 result_dict[key.strip()] = value.strip() 118 119 return result_dict

120

121 - def _get(self, uri):

122 # This method is meant to be monkey-patched in unit tests. 123 self.info("Retrieving page at %s", uri) 124 return get_page(uri)

125

126 - def request(self):

127 """ 128 Request for update data and parse it 129 """ 130 request_params = {'user_id': quote(self.user_id), 131 'version': quote(self.version), 132 'date': quote(self.install_date), 133 'os': quote(self.operating_system)} 134 request_params.update(self.extra_affiliation_params) 135 136 request_url = UPDATES_URL % request_params 137 138 dfr = self._get(request_url) 139 dfr.addCallback(self.parse_result) 140 return dfr

141

142 - def start(self, callback):

143 """ 144 Sets up an automatic loop of update url calls starting right now. 145 Everytime a result is received the callback is triggered with a 146 dictionary of the parsed result as argument. The next iteration is done 147 every L{check_interval}-seconds. 148 149 @raises AlreadyRunning: if the method was already called before. It is only 150 allowed to call this method once. It is mandatory to call L{stop} before 151 calling start again. 152 """ 153 if self._pending_call is not None or self._current_dfr is not None: 154 raise AlreadyRunning("Don't call me twice") 155 156 self._auto_request(callback)

157

158 - def _got_response(self, result, callback):

159 if not result: 160 self.warning("response of the Server was empty!") 161 return 162 163 dfr = callback(result) 164 return dfr

165

166 - def _response_failed(self, failure):

167 if failure.type != defer.CancelledError: 168 self.warning("Problem when requesting for" \ 169 "update: %s" % failure) 170 171 # in any case, eat the error 172 return None

173

174 - def _update_plugin_cache(self, result):

175 plugin_registry = common.application.plugin_registry 176 177 def get_updates_and_new_recommended_list(cache_file): 178 plugins = \ 179 plugin_registry.get_downloadable_plugins(reload_cache=True) 180 installed = list(plugin_registry.get_plugin_names()) 181 # hack for https://bugs.launchpad.net/elisa/+bug/341172 182 if 'elisa-plugin-amp' not in installed: 183 installed.append('elisa-plugin-amp') 184 available_updates = [] 185 new_recommended = [] 186 187 def get_current_plugin_version(plugin_name): 188 current = plugin_registry.get_plugin_by_name(plugin_name) 189 if current is not None: 190 return LooseVersion(current.version) 191 192 if plugin_name == 'elisa-plugin-amp': 193 # case of #341172: the plugin_registry didn't find amp, 194 # even though it is *always* installed, in that case, we 195 # know we have version 0.1, which didn't have its egg-info. 196 return LooseVersion('0.1')

197 198 def iterate_plugins(plugins, installed, 199 available_updates, new_recommended): 200 for plugin_dict in plugins: 201 plugin = Plugin.from_dict(plugin_dict) 202 if plugin.name not in installed and \ 203 plugin_dict['quality'] == 'recommended' and \ 204 plugin.runs_on_current_platform(): 205 new_recommended.append(plugin_dict) 206 elif plugin.name in installed: 207 current_version = \ 208 get_current_plugin_version(plugin.name) 209 if plugin.version > current_version: 210 available_updates.append(plugin_dict) 211 yield None

Source Code for Module elisa.core.utils.update_checker