from slixmpp.xmlstream.matcher import MatchXPath
from slixmpp.xmlstream.handler import Callback
from slixmpp.exceptions import XMPPError
import unittest
from slixmpp.test import SlixTest
class TestStreamExceptions(SlixTest):
"""
Test handling roster updates.
"""
def tearDown(self):
self.stream_close()
def testExceptionReply(self):
"""Test that raising an exception replies with the original stanza."""
def message(msg):
msg.reply()
msg['body'] = 'Body changed'
raise XMPPError(clear=False)
self.stream_start()
self.xmpp.add_event_handler('message', message)
self.recv("""
This is going to cause an error.
""")
self.send("""
This is going to cause an error.
""")
def testExceptionContinueWorking(self):
"""Test that Slixmpp continues to respond after an XMPPError is raised."""
def message(msg):
msg.reply()
msg['body'] = 'Body changed'
raise XMPPError(clear=False)
self.stream_start()
self.xmpp.add_event_handler('message', message)
self.recv("""
This is going to cause an error.
""")
self.send("""
This is going to cause an error.
""")
self.recv("""
This is going to cause an error.
""")
self.send("""
This is going to cause an error.
""")
def testXMPPErrorException(self):
"""Test raising an XMPPError exception."""
def message(msg):
raise XMPPError(condition='feature-not-implemented',
text="We don't do things that way here.",
etype='cancel',
extension='foo',
extension_ns='foo:error',
extension_args={'test': 'true'})
self.stream_start()
self.xmpp.add_event_handler('message', message)
self.recv("""
This is going to cause an error.
""")
self.send("""
We don't do things that way here.
""", use_values=False)
def testIqErrorException(self):
"""Test using error exceptions with Iq stanzas."""
def handle_iq(iq):
raise XMPPError(condition='feature-not-implemented',
text="We don't do things that way here.",
etype='cancel',
clear=False)
self.stream_start()
self.xmpp.register_handler(
Callback(
'Test Iq',
MatchXPath('{%s}iq/{test}query' % self.xmpp.default_ns),
handle_iq))
self.recv("""
""")
self.send("""
We don't do things that way here.
""", use_values=False)
def testThreadedXMPPErrorException(self):
"""Test raising an XMPPError exception in a threaded handler."""
def message(msg):
raise XMPPError(condition='feature-not-implemented',
text="We don't do things that way here.",
etype='cancel')
self.stream_start()
self.xmpp.add_event_handler('message', message,
threaded=True)
self.recv("""
This is going to cause an error.
""")
self.send("""
We don't do things that way here.
""")
def testUnknownException(self):
"""Test raising an generic exception in a threaded handler."""
raised_errors = []
def message(msg):
raise ValueError("Did something wrong")
def catch_error(*args, **kwargs):
raised_errors.append(True)
self.stream_start()
self.xmpp.exception = catch_error
self.xmpp.add_event_handler('message', message)
self.recv("""
This is going to cause an error.
""")
self.send("""
Slixmpp got into trouble.
""")
self.assertEqual(raised_errors, [True], "Exception was not raised: %s" % raised_errors)
def testUnknownException(self):
"""Test Slixmpp continues to respond after an unknown exception."""
raised_errors = []
def message(msg):
raise ValueError("Did something wrong")
def catch_error(*args, **kwargs):
raised_errors.append(True)
self.stream_start()
self.xmpp.exception = catch_error
self.xmpp.add_event_handler('message', message)
self.recv("""
This is going to cause an error.
""")
self.send("""
Slixmpp got into trouble.
""")
self.recv("""
This is going to cause an error.
""")
self.send("""
Slixmpp got into trouble.
""")
self.assertEqual(raised_errors, [True, True], "Exceptions were not raised: %s" % raised_errors)
suite = unittest.TestLoader().loadTestsFromTestCase(TestStreamExceptions)