Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def testConvertData(self):
"""Test that data is converted to base64"""
iq = Iq()
iq['type'] = 'set'
iq['ibb_data']['seq'] = 0
iq['ibb_data']['data'] = 'slixmpp'
self.check(iq, """
def setUp(self):
register_stanza_plugin(Iq, Data)
def plugin_init(self):
register_stanza_plugin(Iq, BlockList)
register_stanza_plugin(Iq, Block)
register_stanza_plugin(Iq, Unblock)
self.xmpp.register_handler(
Callback('Blocked Contact',
StanzaPath('iq@type=set/block'),
self._handle_blocked))
self.xmpp.register_handler(
Callback('Unblocked Contact',
StanzaPath('iq@type=set/unblock'),
self._handle_unblocked))
def plugin_init(self):
register_stanza_plugin(Iq, BlockList)
register_stanza_plugin(Iq, Block)
register_stanza_plugin(Iq, Unblock)
self.xmpp.register_handler(
Callback('Blocked Contact',
StanzaPath('iq@type=set/block'),
self._handle_blocked))
self.xmpp.register_handler(
Callback('Unblocked Contact',
StanzaPath('iq@type=set/unblock'),
self._handle_unblocked))
def plugin_init(self):
register_stanza_plugin(Iq, RPCQuery)
register_stanza_plugin(RPCQuery, MethodCall)
register_stanza_plugin(RPCQuery, MethodResponse)
self.xmpp.register_handler(
Callback('RPC Call', MatchXPath('{%s}iq/{%s}query/{%s}methodCall' % (self.xmpp.default_ns, RPCQuery.namespace, RPCQuery.namespace)),
self._handle_method_call)
)
self.xmpp.register_handler(
Callback('RPC Call', MatchXPath('{%s}iq/{%s}query/{%s}methodResponse' % (self.xmpp.default_ns, RPCQuery.namespace, RPCQuery.namespace)),
self._handle_method_response)
)
self.xmpp.register_handler(
Callback('RPC Call', MatchXPath('{%s}iq/{%s}error' % (self.xmpp.default_ns, self.xmpp.default_ns)),
self._handle_error)
)
self.xmpp.add_event_handler('jabber_rpc_method_call', self._on_jabber_rpc_method_call)
def plugin_init(self):
register_stanza_plugin(Iq, Request)
register_stanza_plugin(Iq, Slot)
register_stanza_plugin(Slot, Put)
register_stanza_plugin(Slot, Get)
register_stanza_plugin(Put, Header, iterable=True)
self.xmpp.register_handler(
Callback('HTTP Upload Request',
StanzaPath('iq@type=get/http_upload_request'),
self._handle_request))
def plugin_init(self):
register_stanza_plugin(Message, SecurityLabel)
register_stanza_plugin(Iq, Catalog)
def _fix_default_info(self, info):
"""
Disco#info results for a JID are required to include at least
one identity and feature. As a default, if no other identity is
provided, Slixmpp will use either the generic component or the
bot client identity. A the standard disco#info feature will also be
added if no features are provided.
Arguments:
info -- The disco#info quest (not the full Iq stanza) to modify.
"""
result = info
if isinstance(info, Iq):
info = info['disco_info']
if not info['node']:
if not info['identities']:
if self.xmpp.is_component:
log.debug("No identity found for this entity. " + \
"Using default component identity.")
info.add_identity('component', 'generic')
else:
log.debug("No identity found for this entity. " + \
"Using default client identity.")
info.add_identity('client', 'bot')
if not info['features']:
log.debug("No features found for this entity. " + \
"Using default disco#info feature.")
info.add_feature(info.namespace)
return result
self.xmpp.register_handler(
Callback(
'HTTP Request',
StanzaPath('iq/http-req'),
self._handle_request
)
)
self.xmpp.register_handler(
Callback(
'HTTP Response',
StanzaPath('iq/http-resp'),
self._handle_response
)
)
register_stanza_plugin(Iq, HTTPRequest, iterable=True)
register_stanza_plugin(Iq, HTTPResponse, iterable=True)
register_stanza_plugin(HTTPRequest, Headers, iterable=True)
register_stanza_plugin(HTTPRequest, HTTPData, iterable=True)
register_stanza_plugin(HTTPResponse, Headers, iterable=True)
register_stanza_plugin(HTTPResponse, HTTPData, iterable=True)
# TODO: Should we register any api's here? self.api.register()
# The session_start event will be triggered when
# the bot establishes its connection with the server
# and the XML streams are ready for use. We want to
# listen for this event so that we we can initialize
# our roster.
self.add_event_handler("session_start", self.start)
self.register_handler(
Callback('Some custom iq',
StanzaPath('iq@type=set/action'),
self._handle_action))
self.add_event_handler('custom_action',
self._handle_action_event)
register_stanza_plugin(Iq, Action)