comparison src/plugins/plugin_xep_0277.py @ 1913:ee1125fffba8

plugin XEP-0277, test: set keys of data dict as unicode + fix the tests
author souliane <souliane@mailoo.org>
date Fri, 18 Mar 2016 08:58:22 +0100
parents 085f29c20f7e
children d3354c80bd1f
comparison
equal deleted inserted replaced
1912:c38bcc0343b6 1913:ee1125fffba8
179 key = check_conflict(elem.name) 179 key = check_conflict(elem.name)
180 microblog_data[key] = unicode(elem) 180 microblog_data[key] = unicode(elem)
181 181
182 182
183 id_ = item_elt.getAttribute('id', '') # there can be no id for transient nodes 183 id_ = item_elt.getAttribute('id', '') # there can be no id for transient nodes
184 microblog_data['id'] = id_ 184 microblog_data[u'id'] = id_
185 if item_elt.uri not in (pubsub.NS_PUBSUB, NS_PUBSUB_EVENT): 185 if item_elt.uri not in (pubsub.NS_PUBSUB, NS_PUBSUB_EVENT):
186 msg = u"Unsupported namespace {ns} in pubsub item {id_}".format(ns=item_elt.uri, id_=id_) 186 msg = u"Unsupported namespace {ns} in pubsub item {id_}".format(ns=item_elt.uri, id_=id_)
187 log.warning(msg) 187 log.warning(msg)
188 raise failure.Failure(exceptions.DataError(msg)) 188 raise failure.Failure(exceptions.DataError(msg))
189 189
197 try: 197 try:
198 id_elt = entry_elt.elements(NS_ATOM, 'id').next() 198 id_elt = entry_elt.elements(NS_ATOM, 'id').next()
199 except StopIteration: 199 except StopIteration:
200 msg = u'No atom id found in the pubsub item {}, this is not standard !'.format(id_) 200 msg = u'No atom id found in the pubsub item {}, this is not standard !'.format(id_)
201 log.warning(msg) 201 log.warning(msg)
202 microblog_data['atom_id'] = "" 202 microblog_data[u'atom_id'] = ""
203 else: 203 else:
204 microblog_data['atom_id'] = unicode(id_elt) 204 microblog_data[u'atom_id'] = unicode(id_elt)
205 205
206 # title/content(s) 206 # title/content(s)
207 207
208 # FIXME: ATOM and XEP-0277 only allow 1 <title/> element 208 # FIXME: ATOM and XEP-0277 only allow 1 <title/> element
209 # but in the wild we have some blogs with several ones 209 # but in the wild we have some blogs with several ones
231 for key in ('title', 'content'): 231 for key in ('title', 'content'):
232 if key not in microblog_data and ('{}_xhtml'.format(key)) in microblog_data: 232 if key not in microblog_data and ('{}_xhtml'.format(key)) in microblog_data:
233 log.warning(u"item {id_} provide a {key}_xhtml data but not a text one".format(id_=id_, key=key)) 233 log.warning(u"item {id_} provide a {key}_xhtml data but not a text one".format(id_=id_, key=key))
234 # ... and do the conversion if it's not 234 # ... and do the conversion if it's not
235 microblog_data[key] = yield self.host.plugins["TEXT-SYNTAXES"].\ 235 microblog_data[key] = yield self.host.plugins["TEXT-SYNTAXES"].\
236 convert(microblog_data['{}_xhtml'.format(key)], 236 convert(microblog_data[u'{}_xhtml'.format(key)],
237 self.host.plugins["TEXT-SYNTAXES"].SYNTAX_XHTML, 237 self.host.plugins["TEXT-SYNTAXES"].SYNTAX_XHTML,
238 self.host.plugins["TEXT-SYNTAXES"].SYNTAX_TEXT, 238 self.host.plugins["TEXT-SYNTAXES"].SYNTAX_TEXT,
239 False) 239 False)
240 240
241 if 'content' not in microblog_data: 241 if 'content' not in microblog_data:
242 # use the atom title data as the microblog body content 242 # use the atom title data as the microblog body content
243 microblog_data['content'] = microblog_data['title'] 243 microblog_data[u'content'] = microblog_data[u'title']
244 del microblog_data['title'] 244 del microblog_data[u'title']
245 if 'title_xhtml' in microblog_data: 245 if 'title_xhtml' in microblog_data:
246 microblog_data['content_xhtml'] = microblog_data['title_xhtml'] 246 microblog_data[u'content_xhtml'] = microblog_data[u'title_xhtml']
247 del microblog_data['title_xhtml'] 247 del microblog_data[u'title_xhtml']
248 248
249 # published/updated dates 249 # published/updated dates
250 try: 250 try:
251 updated_elt = entry_elt.elements(NS_ATOM, 'updated').next() 251 updated_elt = entry_elt.elements(NS_ATOM, 'updated').next()
252 except StopIteration: 252 except StopIteration:
253 msg = u'No atom updated element found in the pubsub item {}'.format(id_) 253 msg = u'No atom updated element found in the pubsub item {}'.format(id_)
254 raise failure.Failure(exceptions.DataError(msg)) 254 raise failure.Failure(exceptions.DataError(msg))
255 microblog_data['updated'] = unicode(rfc3339.tf_from_timestamp(unicode(updated_elt))) 255 microblog_data[u'updated'] = unicode(rfc3339.tf_from_timestamp(unicode(updated_elt)))
256 try: 256 try:
257 published_elt = entry_elt.elements(NS_ATOM, 'published').next() 257 published_elt = entry_elt.elements(NS_ATOM, 'published').next()
258 except StopIteration: 258 except StopIteration:
259 microblog_data['published'] = microblog_data['updated'] 259 microblog_data[u'published'] = microblog_data[u'updated']
260 else: 260 else:
261 microblog_data['published'] = unicode(rfc3339.tf_from_timestamp(unicode(published_elt))) 261 microblog_data[u'published'] = unicode(rfc3339.tf_from_timestamp(unicode(published_elt)))
262 262
263 # links 263 # links
264 for link_elt in entry_elt.elements(NS_ATOM, 'link'): 264 for link_elt in entry_elt.elements(NS_ATOM, 'link'):
265 if link_elt.getAttribute('rel') == 'replies' and link_elt.getAttribute('title') == 'comments': 265 if link_elt.getAttribute('rel') == 'replies' and link_elt.getAttribute('title') == 'comments':
266 key = check_conflict('comments', True) 266 key = check_conflict('comments', True)
269 service, node = self.parseCommentUrl(microblog_data[key]) 269 service, node = self.parseCommentUrl(microblog_data[key])
270 except: 270 except:
271 log.warning(u"Can't parse url {}".format(microblog_data[key])) 271 log.warning(u"Can't parse url {}".format(microblog_data[key]))
272 del microblog_data[key] 272 del microblog_data[key]
273 else: 273 else:
274 microblog_data['{}_service'.format(key)] = service.full() 274 microblog_data[u'{}_service'.format(key)] = service.full()
275 microblog_data['{}_node'.format(key)] = node 275 microblog_data[u'{}_node'.format(key)] = node
276 else: 276 else:
277 rel = link_elt.getAttribute('rel','') 277 rel = link_elt.getAttribute('rel','')
278 title = link_elt.getAttribute('title','') 278 title = link_elt.getAttribute('title','')
279 href = link_elt.getAttribute('href','') 279 href = link_elt.getAttribute('href','')
280 log.warning(u"Unmanaged link element: rel={rel} title={title} href={href}".format(rel=rel, title=title, href=href)) 280 log.warning(u"Unmanaged link element: rel={rel} title={title} href={href}".format(rel=rel, title=title, href=href))
290 try: 290 try:
291 name_elt = author_elt.elements(NS_ATOM, 'name').next() 291 name_elt = author_elt.elements(NS_ATOM, 'name').next()
292 except StopIteration: 292 except StopIteration:
293 log.warning(u"No name element found in author element of item {}".format(id_)) 293 log.warning(u"No name element found in author element of item {}".format(id_))
294 else: 294 else:
295 microblog_data['author'] = unicode(name_elt) 295 microblog_data[u'author'] = unicode(name_elt)
296 # uri 296 # uri
297 try: 297 try:
298 uri_elt = author_elt.elements(NS_ATOM, 'uri').next() 298 uri_elt = author_elt.elements(NS_ATOM, 'uri').next()
299 except StopIteration: 299 except StopIteration:
300 log.debug(u"No uri element found in author element of item {}".format(id_)) 300 log.debug(u"No uri element found in author element of item {}".format(id_))
301 if publisher: 301 if publisher:
302 microblog_data['author_jid'] = publisher 302 microblog_data[u'author_jid'] = publisher
303 else: 303 else:
304 uri = unicode(uri_elt) 304 uri = unicode(uri_elt)
305 if uri.startswith("xmpp:"): 305 if uri.startswith("xmpp:"):
306 uri = uri[5:] 306 uri = uri[5:]
307 microblog_data['author_jid'] = uri 307 microblog_data[u'author_jid'] = uri
308 else: 308 else:
309 microblog_data['author_jid'] = item_elt.getAttribute("publisher") or "" 309 microblog_data[u'author_jid'] = item_elt.getAttribute(u"publisher") or ""
310 310
311 if not publisher: 311 if not publisher:
312 log.debug(u"No publisher attribute, we can't verify author jid") 312 log.debug(u"No publisher attribute, we can't verify author jid")
313 microblog_data['author_jid_verified'] = C.BOOL_FALSE 313 microblog_data[u'author_jid_verified'] = C.BOOL_FALSE
314 elif jid.JID(publisher).userhostJID() == jid.JID(uri).userhostJID(): 314 elif jid.JID(publisher).userhostJID() == jid.JID(uri).userhostJID():
315 microblog_data['author_jid_verified'] = C.BOOL_TRUE 315 microblog_data[u'author_jid_verified'] = C.BOOL_TRUE
316 else: 316 else:
317 log.warning(u"item atom:uri differ from publisher attribute, spoofing attempt ? atom:uri = {} publisher = {}".format(uri, item_elt.getAttribute("publisher"))) 317 log.warning(u"item atom:uri differ from publisher attribute, spoofing attempt ? atom:uri = {} publisher = {}".format(uri, item_elt.getAttribute("publisher")))
318 microblog_data['author_jid_verified'] = C.BOOL_FALSE 318 microblog_data[u'author_jid_verified'] = C.BOOL_FALSE
319 # email 319 # email
320 try: 320 try:
321 email_elt = author_elt.elements(NS_ATOM, 'email').next() 321 email_elt = author_elt.elements(NS_ATOM, 'email').next()
322 except StopIteration: 322 except StopIteration:
323 pass 323 pass
324 else: 324 else:
325 microblog_data['author_email'] = unicode(email_elt) 325 microblog_data[u'author_email'] = unicode(email_elt)
326 326
327 # categories 327 # categories
328 categories = (category_elt.getAttribute('term','') for category_elt in entry_elt.elements(NS_ATOM, 'category')) 328 categories = (category_elt.getAttribute('term','') for category_elt in entry_elt.elements(NS_ATOM, 'category'))
329 common.iter2dict('tag', categories, microblog_data) 329 common.iter2dict('tag', categories, microblog_data)
330 330