summaryrefslogtreecommitdiff
path: root/sleekxmpp/test/sleektest.py
diff options
context:
space:
mode:
Diffstat (limited to 'sleekxmpp/test/sleektest.py')
-rw-r--r--sleekxmpp/test/sleektest.py757
1 files changed, 757 insertions, 0 deletions
diff --git a/sleekxmpp/test/sleektest.py b/sleekxmpp/test/sleektest.py
new file mode 100644
index 00000000..dd3df29a
--- /dev/null
+++ b/sleekxmpp/test/sleektest.py
@@ -0,0 +1,757 @@
+"""
+ SleekXMPP: The Sleek XMPP Library
+ Copyright (C) 2010 Nathanael C. Fritz, Lance J.T. Stout
+ This file is part of SleekXMPP.
+
+ See the file LICENSE for copying permission.
+"""
+
+import unittest
+try:
+ import Queue as queue
+except:
+ import queue
+
+import sleekxmpp
+from sleekxmpp import ClientXMPP, ComponentXMPP
+from sleekxmpp.stanza import Message, Iq, Presence
+from sleekxmpp.test import TestSocket, TestLiveSocket
+from sleekxmpp.exceptions import XMPPError, IqTimeout, IqError
+from sleekxmpp.xmlstream import ET, register_stanza_plugin
+from sleekxmpp.xmlstream import ElementBase, StanzaBase
+from sleekxmpp.xmlstream.tostring import tostring
+from sleekxmpp.xmlstream.matcher import StanzaPath, MatcherId
+from sleekxmpp.xmlstream.matcher import MatchXMLMask, MatchXPath
+
+
+class SleekTest(unittest.TestCase):
+
+ """
+ A SleekXMPP specific TestCase class that provides
+ methods for comparing message, iq, and presence stanzas.
+
+ Methods:
+ Message -- Create a Message stanza object.
+ Iq -- Create an Iq stanza object.
+ Presence -- Create a Presence stanza object.
+ check_jid -- Check a JID and its component parts.
+ check -- Compare a stanza against an XML string.
+ stream_start -- Initialize a dummy XMPP client.
+ stream_close -- Disconnect the XMPP client.
+ make_header -- Create a stream header.
+ send_header -- Check that the given header has been sent.
+ send_feature -- Send a raw XML element.
+ send -- Check that the XMPP client sent the given
+ generic stanza.
+ recv -- Queue data for XMPP client to receive, or
+ verify the data that was received from a
+ live connection.
+ recv_header -- Check that a given stream header
+ was received.
+ recv_feature -- Check that a given, raw XML element
+ was recveived.
+ fix_namespaces -- Add top-level namespace to an XML object.
+ compare -- Compare XML objects against each other.
+ """
+
+ def __init__(self, *args, **kwargs):
+ unittest.TestCase.__init__(self, *args, **kwargs)
+ self.xmpp = None
+
+ def parse_xml(self, xml_string):
+ try:
+ xml = ET.fromstring(xml_string)
+ return xml
+ except SyntaxError as e:
+ if 'unbound' in e.msg:
+ known_prefixes = {
+ 'stream': 'http://etherx.jabber.org/streams'}
+
+ prefix = xml_string.split('<')[1].split(':')[0]
+ if prefix in known_prefixes:
+ xml_string = '<fixns xmlns:%s="%s">%s</fixns>' % (
+ prefix,
+ known_prefixes[prefix],
+ xml_string)
+ xml = self.parse_xml(xml_string)
+ xml = xml.getchildren()[0]
+ return xml
+ else:
+ self.fail("XML data was mal-formed:\n%s" % xml_string)
+
+ # ------------------------------------------------------------------
+ # Shortcut methods for creating stanza objects
+
+ def Message(self, *args, **kwargs):
+ """
+ Create a Message stanza.
+
+ Uses same arguments as StanzaBase.__init__
+
+ Arguments:
+ xml -- An XML object to use for the Message's values.
+ """
+ return Message(self.xmpp, *args, **kwargs)
+
+ def Iq(self, *args, **kwargs):
+ """
+ Create an Iq stanza.
+
+ Uses same arguments as StanzaBase.__init__
+
+ Arguments:
+ xml -- An XML object to use for the Iq's values.
+ """
+ return Iq(self.xmpp, *args, **kwargs)
+
+ def Presence(self, *args, **kwargs):
+ """
+ Create a Presence stanza.
+
+ Uses same arguments as StanzaBase.__init__
+
+ Arguments:
+ xml -- An XML object to use for the Iq's values.
+ """
+ return Presence(self.xmpp, *args, **kwargs)
+
+ def check_jid(self, jid, user=None, domain=None, resource=None,
+ bare=None, full=None, string=None):
+ """
+ Verify the components of a JID.
+
+ Arguments:
+ jid -- The JID object to test.
+ user -- Optional. The user name portion of the JID.
+ domain -- Optional. The domain name portion of the JID.
+ resource -- Optional. The resource portion of the JID.
+ bare -- Optional. The bare JID.
+ full -- Optional. The full JID.
+ string -- Optional. The string version of the JID.
+ """
+ if user is not None:
+ self.assertEqual(jid.user, user,
+ "User does not match: %s" % jid.user)
+ if domain is not None:
+ self.assertEqual(jid.domain, domain,
+ "Domain does not match: %s" % jid.domain)
+ if resource is not None:
+ self.assertEqual(jid.resource, resource,
+ "Resource does not match: %s" % jid.resource)
+ if bare is not None:
+ self.assertEqual(jid.bare, bare,
+ "Bare JID does not match: %s" % jid.bare)
+ if full is not None:
+ self.assertEqual(jid.full, full,
+ "Full JID does not match: %s" % jid.full)
+ if string is not None:
+ self.assertEqual(str(jid), string,
+ "String does not match: %s" % str(jid))
+
+ def check_roster(self, owner, jid, name=None, subscription=None,
+ afrom=None, ato=None, pending_out=None, pending_in=None,
+ groups=None):
+ roster = self.xmpp.roster[owner][jid]
+ if name is not None:
+ self.assertEqual(roster['name'], name,
+ "Incorrect name value: %s" % roster['name'])
+ if subscription is not None:
+ self.assertEqual(roster['subscription'], subscription,
+ "Incorrect subscription: %s" % roster['subscription'])
+ if afrom is not None:
+ self.assertEqual(roster['from'], afrom,
+ "Incorrect from state: %s" % roster['from'])
+ if ato is not None:
+ self.assertEqual(roster['to'], ato,
+ "Incorrect to state: %s" % roster['to'])
+ if pending_out is not None:
+ self.assertEqual(roster['pending_out'], pending_out,
+ "Incorrect pending_out state: %s" % roster['pending_out'])
+ if pending_in is not None:
+ self.assertEqual(roster['pending_in'], pending_out,
+ "Incorrect pending_in state: %s" % roster['pending_in'])
+ if groups is not None:
+ self.assertEqual(roster['groups'], groups,
+ "Incorrect groups: %s" % roster['groups'])
+
+ # ------------------------------------------------------------------
+ # Methods for comparing stanza objects to XML strings
+
+ def check(self, stanza, criteria, method='exact',
+ defaults=None, use_values=True):
+ """
+ Create and compare several stanza objects to a correct XML string.
+
+ If use_values is False, tests using stanza.values will not be used.
+
+ Some stanzas provide default values for some interfaces, but
+ these defaults can be problematic for testing since they can easily
+ be forgotten when supplying the XML string. A list of interfaces that
+ use defaults may be provided and the generated stanzas will use the
+ default values for those interfaces if needed.
+
+ However, correcting the supplied XML is not possible for interfaces
+ that add or remove XML elements. Only interfaces that map to XML
+ attributes may be set using the defaults parameter. The supplied XML
+ must take into account any extra elements that are included by default.
+
+ Arguments:
+ stanza -- The stanza object to test.
+ criteria -- An expression the stanza must match against.
+ method -- The type of matching to use; one of:
+ 'exact', 'mask', 'id', 'xpath', and 'stanzapath'.
+ Defaults to the value of self.match_method.
+ defaults -- A list of stanza interfaces that have default
+ values. These interfaces will be set to their
+ defaults for the given and generated stanzas to
+ prevent unexpected test failures.
+ use_values -- Indicates if testing using stanza.values should
+ be used. Defaults to True.
+ """
+ if method is None and hasattr(self, 'match_method'):
+ method = getattr(self, 'match_method')
+
+ if method != 'exact':
+ matchers = {'stanzapath': StanzaPath,
+ 'xpath': MatchXPath,
+ 'mask': MatchXMLMask,
+ 'id': MatcherId}
+ Matcher = matchers.get(method, None)
+ if Matcher is None:
+ raise ValueError("Unknown matching method.")
+ test = Matcher(criteria)
+ self.failUnless(test.match(stanza),
+ "Stanza did not match using %s method:\n" % method + \
+ "Criteria:\n%s\n" % str(criteria) + \
+ "Stanza:\n%s" % str(stanza))
+ else:
+ stanza_class = stanza.__class__
+ if not isinstance(criteria, ElementBase):
+ xml = self.parse_xml(criteria)
+ else:
+ xml = criteria.xml
+
+ # Ensure that top level namespaces are used, even if they
+ # were not provided.
+ self.fix_namespaces(stanza.xml, 'jabber:client')
+ self.fix_namespaces(xml, 'jabber:client')
+
+ stanza2 = stanza_class(xml=xml)
+
+ if use_values:
+ # Using stanza.values will add XML for any interface that
+ # has a default value. We need to set those defaults on
+ # the existing stanzas and XML so that they will compare
+ # correctly.
+ default_stanza = stanza_class()
+ if defaults is None:
+ known_defaults = {
+ Message: ['type'],
+ Presence: ['priority']
+ }
+ defaults = known_defaults.get(stanza_class, [])
+ for interface in defaults:
+ stanza[interface] = stanza[interface]
+ stanza2[interface] = stanza2[interface]
+ # Can really only automatically add defaults for top
+ # level attribute values. Anything else must be accounted
+ # for in the provided XML string.
+ if interface not in xml.attrib:
+ if interface in default_stanza.xml.attrib:
+ value = default_stanza.xml.attrib[interface]
+ xml.attrib[interface] = value
+
+ values = stanza2.values
+ stanza3 = stanza_class()
+ stanza3.values = values
+
+ debug = "Three methods for creating stanzas do not match.\n"
+ debug += "Given XML:\n%s\n" % tostring(xml)
+ debug += "Given stanza:\n%s\n" % tostring(stanza.xml)
+ debug += "Generated stanza:\n%s\n" % tostring(stanza2.xml)
+ debug += "Second generated stanza:\n%s\n" % tostring(stanza3.xml)
+ result = self.compare(xml, stanza.xml, stanza2.xml, stanza3.xml)
+ else:
+ debug = "Two methods for creating stanzas do not match.\n"
+ debug += "Given XML:\n%s\n" % tostring(xml)
+ debug += "Given stanza:\n%s\n" % tostring(stanza.xml)
+ debug += "Generated stanza:\n%s\n" % tostring(stanza2.xml)
+ result = self.compare(xml, stanza.xml, stanza2.xml)
+
+ self.failUnless(result, debug)
+
+ # ------------------------------------------------------------------
+ # Methods for simulating stanza streams.
+
+ def stream_disconnect(self):
+ """
+ Simulate a stream disconnection.
+ """
+ if self.xmpp:
+ self.xmpp.socket.disconnect_error()
+
+ def stream_start(self, mode='client', skip=True, header=None,
+ socket='mock', jid='tester@localhost',
+ password='test', server='localhost',
+ port=5222, sasl_mech=None,
+ plugins=None, plugin_config={}):
+ """
+ Initialize an XMPP client or component using a dummy XML stream.
+
+ Arguments:
+ mode -- Either 'client' or 'component'. Defaults to 'client'.
+ skip -- Indicates if the first item in the sent queue (the
+ stream header) should be removed. Tests that wish
+ to test initializing the stream should set this to
+ False. Otherwise, the default of True should be used.
+ socket -- Either 'mock' or 'live' to indicate if the socket
+ should be a dummy, mock socket or a live, functioning
+ socket. Defaults to 'mock'.
+ jid -- The JID to use for the connection.
+ Defaults to 'tester@localhost'.
+ password -- The password to use for the connection.
+ Defaults to 'test'.
+ server -- The name of the XMPP server. Defaults to 'localhost'.
+ port -- The port to use when connecting to the server.
+ Defaults to 5222.
+ plugins -- List of plugins to register. By default, all plugins
+ are loaded.
+ """
+ if mode == 'client':
+ self.xmpp = ClientXMPP(jid, password,
+ sasl_mech=sasl_mech,
+ plugin_config=plugin_config)
+ elif mode == 'component':
+ self.xmpp = ComponentXMPP(jid, password,
+ server, port,
+ plugin_config=plugin_config)
+ else:
+ raise ValueError("Unknown XMPP connection mode.")
+
+ # We will use this to wait for the session_start event
+ # for live connections.
+ skip_queue = queue.Queue()
+
+ if socket == 'mock':
+ self.xmpp.set_socket(TestSocket())
+
+ # Simulate connecting for mock sockets.
+ self.xmpp.auto_reconnect = False
+ self.xmpp.state._set_state('connected')
+
+ # Must have the stream header ready for xmpp.process() to work.
+ if not header:
+ header = self.xmpp.stream_header
+ self.xmpp.socket.recv_data(header)
+ elif socket == 'live':
+ self.xmpp.socket_class = TestLiveSocket
+
+ def wait_for_session(x):
+ self.xmpp.socket.clear()
+ skip_queue.put('started')
+
+ self.xmpp.add_event_handler('session_start', wait_for_session)
+ if server is not None:
+ self.xmpp.connect((server, port))
+ else:
+ self.xmpp.connect()
+ else:
+ raise ValueError("Unknown socket type.")
+
+ if plugins is None:
+ self.xmpp.register_plugins()
+ else:
+ for plugin in plugins:
+ self.xmpp.register_plugin(plugin)
+ self.xmpp.process(threaded=True)
+ if skip:
+ if socket != 'live':
+ # Mark send queue as usable
+ self.xmpp.session_started_event.set()
+ # Clear startup stanzas
+ self.xmpp.socket.next_sent(timeout=1)
+ if mode == 'component':
+ self.xmpp.socket.next_sent(timeout=1)
+ else:
+ skip_queue.get(block=True, timeout=10)
+
+ def make_header(self, sto='',
+ sfrom='',
+ sid='',
+ stream_ns="http://etherx.jabber.org/streams",
+ default_ns="jabber:client",
+ version="1.0",
+ xml_header=True):
+ """
+ Create a stream header to be received by the test XMPP agent.
+
+ The header must be saved and passed to stream_start.
+
+ Arguments:
+ sto -- The recipient of the stream header.
+ sfrom -- The agent sending the stream header.
+ sid -- The stream's id.
+ stream_ns -- The namespace of the stream's root element.
+ default_ns -- The default stanza namespace.
+ version -- The stream version.
+ xml_header -- Indicates if the XML version header should be
+ appended before the stream header.
+ """
+ header = '<stream:stream %s>'
+ parts = []
+ if xml_header:
+ header = '<?xml version="1.0"?>' + header
+ if sto:
+ parts.append('to="%s"' % sto)
+ if sfrom:
+ parts.append('from="%s"' % sfrom)
+ if sid:
+ parts.append('id="%s"' % sid)
+ parts.append('version="%s"' % version)
+ parts.append('xmlns:stream="%s"' % stream_ns)
+ parts.append('xmlns="%s"' % default_ns)
+ return header % ' '.join(parts)
+
+ def recv(self, data, defaults=[], method='exact',
+ use_values=True, timeout=1):
+ """
+ Pass data to the dummy XMPP client as if it came from an XMPP server.
+
+ If using a live connection, verify what the server has sent.
+
+ Arguments:
+ data -- If a dummy socket is being used, the XML that is to
+ be received next. Otherwise it is the criteria used
+ to match against live data that is received.
+ defaults -- A list of stanza interfaces with default values that
+ may interfere with comparisons.
+ method -- Select the type of comparison to use for
+ verifying the received stanza. Options are 'exact',
+ 'id', 'stanzapath', 'xpath', and 'mask'.
+ Defaults to the value of self.match_method.
+ use_values -- Indicates if stanza comparisons should test using
+ stanza.values. Defaults to True.
+ timeout -- Time to wait in seconds for data to be received by
+ a live connection.
+ """
+ if self.xmpp.socket.is_live:
+ # we are working with a live connection, so we should
+ # verify what has been received instead of simulating
+ # receiving data.
+ recv_data = self.xmpp.socket.next_recv(timeout)
+ if recv_data is None:
+ self.fail("No stanza was received.")
+ xml = self.parse_xml(recv_data)
+ self.fix_namespaces(xml, 'jabber:client')
+ stanza = self.xmpp._build_stanza(xml, 'jabber:client')
+ self.check(stanza, data,
+ method=method,
+ defaults=defaults,
+ use_values=use_values)
+ else:
+ # place the data in the dummy socket receiving queue.
+ data = str(data)
+ self.xmpp.socket.recv_data(data)
+
+ def recv_header(self, sto='',
+ sfrom='',
+ sid='',
+ stream_ns="http://etherx.jabber.org/streams",
+ default_ns="jabber:client",
+ version="1.0",
+ xml_header=False,
+ timeout=1):
+ """
+ Check that a given stream header was received.
+
+ Arguments:
+ sto -- The recipient of the stream header.
+ sfrom -- The agent sending the stream header.
+ sid -- The stream's id. Set to None to ignore.
+ stream_ns -- The namespace of the stream's root element.
+ default_ns -- The default stanza namespace.
+ version -- The stream version.
+ xml_header -- Indicates if the XML version header should be
+ appended before the stream header.
+ timeout -- Length of time to wait in seconds for a
+ response.
+ """
+ header = self.make_header(sto, sfrom, sid,
+ stream_ns=stream_ns,
+ default_ns=default_ns,
+ version=version,
+ xml_header=xml_header)
+ recv_header = self.xmpp.socket.next_recv(timeout)
+ if recv_header is None:
+ raise ValueError("Socket did not return data.")
+
+ # Apply closing elements so that we can construct
+ # XML objects for comparison.
+ header2 = header + '</stream:stream>'
+ recv_header2 = recv_header + '</stream:stream>'
+
+ xml = self.parse_xml(header2)
+ recv_xml = self.parse_xml(recv_header2)
+
+ if sid is None:
+ # Ignore the id sent by the server since
+ # we can't know in advance what it will be.
+ if 'id' in recv_xml.attrib:
+ del recv_xml.attrib['id']
+
+ # Ignore the xml:lang attribute for now.
+ if 'xml:lang' in recv_xml.attrib:
+ del recv_xml.attrib['xml:lang']
+ xml_ns = 'http://www.w3.org/XML/1998/namespace'
+ if '{%s}lang' % xml_ns in recv_xml.attrib:
+ del recv_xml.attrib['{%s}lang' % xml_ns]
+
+ if recv_xml.getchildren:
+ # We received more than just the header
+ for xml in recv_xml.getchildren():
+ self.xmpp.socket.recv_data(tostring(xml))
+
+ attrib = recv_xml.attrib
+ recv_xml.clear()
+ recv_xml.attrib = attrib
+
+ self.failUnless(
+ self.compare(xml, recv_xml),
+ "Stream headers do not match:\nDesired:\n%s\nReceived:\n%s" % (
+ '%s %s' % (xml.tag, xml.attrib),
+ '%s %s' % (recv_xml.tag, recv_xml.attrib)))
+
+ def recv_feature(self, data, method='mask', use_values=True, timeout=1):
+ """
+ """
+ if method is None and hasattr(self, 'match_method'):
+ method = getattr(self, 'match_method')
+
+ if self.xmpp.socket.is_live:
+ # we are working with a live connection, so we should
+ # verify what has been received instead of simulating
+ # receiving data.
+ recv_data = self.xmpp.socket.next_recv(timeout)
+ xml = self.parse_xml(data)
+ recv_xml = self.parse_xml(recv_data)
+ if recv_data is None:
+ self.fail("No stanza was received.")
+ if method == 'exact':
+ self.failUnless(self.compare(xml, recv_xml),
+ "Features do not match.\nDesired:\n%s\nReceived:\n%s" % (
+ tostring(xml), tostring(recv_xml)))
+ elif method == 'mask':
+ matcher = MatchXMLMask(xml)
+ self.failUnless(matcher.match(recv_xml),
+ "Stanza did not match using %s method:\n" % method + \
+ "Criteria:\n%s\n" % tostring(xml) + \
+ "Stanza:\n%s" % tostring(recv_xml))
+ else:
+ raise ValueError("Uknown matching method: %s" % method)
+ else:
+ # place the data in the dummy socket receiving queue.
+ data = str(data)
+ self.xmpp.socket.recv_data(data)
+
+ def send_header(self, sto='',
+ sfrom='',
+ sid='',
+ stream_ns="http://etherx.jabber.org/streams",
+ default_ns="jabber:client",
+ version="1.0",
+ xml_header=False,
+ timeout=1):
+ """
+ Check that a given stream header was sent.
+
+ Arguments:
+ sto -- The recipient of the stream header.
+ sfrom -- The agent sending the stream header.
+ sid -- The stream's id.
+ stream_ns -- The namespace of the stream's root element.
+ default_ns -- The default stanza namespace.
+ version -- The stream version.
+ xml_header -- Indicates if the XML version header should be
+ appended before the stream header.
+ timeout -- Length of time to wait in seconds for a
+ response.
+ """
+ header = self.make_header(sto, sfrom, sid,
+ stream_ns=stream_ns,
+ default_ns=default_ns,
+ version=version,
+ xml_header=xml_header)
+ sent_header = self.xmpp.socket.next_sent(timeout)
+ if sent_header is None:
+ raise ValueError("Socket did not return data.")
+
+ # Apply closing elements so that we can construct
+ # XML objects for comparison.
+ header2 = header + '</stream:stream>'
+ sent_header2 = sent_header + b'</stream:stream>'
+
+ xml = self.parse_xml(header2)
+ sent_xml = self.parse_xml(sent_header2)
+
+ self.failUnless(
+ self.compare(xml, sent_xml),
+ "Stream headers do not match:\nDesired:\n%s\nSent:\n%s" % (
+ header, sent_header))
+
+ def send_feature(self, data, method='mask', use_values=True, timeout=1):
+ """
+ """
+ sent_data = self.xmpp.socket.next_sent(timeout)
+ xml = self.parse_xml(data)
+ sent_xml = self.parse_xml(sent_data)
+ if sent_data is None:
+ self.fail("No stanza was sent.")
+ if method == 'exact':
+ self.failUnless(self.compare(xml, sent_xml),
+ "Features do not match.\nDesired:\n%s\nReceived:\n%s" % (
+ tostring(xml), tostring(sent_xml)))
+ elif method == 'mask':
+ matcher = MatchXMLMask(xml)
+ self.failUnless(matcher.match(sent_xml),
+ "Stanza did not match using %s method:\n" % method + \
+ "Criteria:\n%s\n" % tostring(xml) + \
+ "Stanza:\n%s" % tostring(sent_xml))
+ else:
+ raise ValueError("Uknown matching method: %s" % method)
+
+ def send(self, data, defaults=None, use_values=True,
+ timeout=.5, method='exact'):
+ """
+ Check that the XMPP client sent the given stanza XML.
+
+ Extracts the next sent stanza and compares it with the given
+ XML using check.
+
+ Arguments:
+ stanza_class -- The class of the sent stanza object.
+ data -- The XML string of the expected Message stanza,
+ or an equivalent stanza object.
+ use_values -- Modifies the type of tests used by check_message.
+ defaults -- A list of stanza interfaces that have defaults
+ values which may interfere with comparisons.
+ timeout -- Time in seconds to wait for a stanza before
+ failing the check.
+ method -- Select the type of comparison to use for
+ verifying the sent stanza. Options are 'exact',
+ 'id', 'stanzapath', 'xpath', and 'mask'.
+ Defaults to the value of self.match_method.
+ """
+ sent = self.xmpp.socket.next_sent(timeout)
+ if data is None and sent is None:
+ return
+ if data is None and sent is not None:
+ self.fail("Stanza data was sent: %s" % sent)
+ if sent is None:
+ self.fail("No stanza was sent.")
+
+ xml = self.parse_xml(sent)
+ self.fix_namespaces(xml, 'jabber:client')
+ sent = self.xmpp._build_stanza(xml, 'jabber:client')
+ self.check(sent, data,
+ method=method,
+ defaults=defaults,
+ use_values=use_values)
+
+ def stream_close(self):
+ """
+ Disconnect the dummy XMPP client.
+
+ Can be safely called even if stream_start has not been called.
+
+ Must be placed in the tearDown method of a test class to ensure
+ that the XMPP client is disconnected after an error.
+ """
+ if hasattr(self, 'xmpp') and self.xmpp is not None:
+ self.xmpp.socket.recv_data(self.xmpp.stream_footer)
+ self.xmpp.disconnect()
+
+ # ------------------------------------------------------------------
+ # XML Comparison and Cleanup
+
+ def fix_namespaces(self, xml, ns):
+ """
+ Assign a namespace to an element and any children that
+ don't have a namespace.
+
+ Arguments:
+ xml -- The XML object to fix.
+ ns -- The namespace to add to the XML object.
+ """
+ if xml.tag.startswith('{'):
+ return
+ xml.tag = '{%s}%s' % (ns, xml.tag)
+ for child in xml.getchildren():
+ self.fix_namespaces(child, ns)
+
+ def compare(self, xml, *other):
+ """
+ Compare XML objects.
+
+ Arguments:
+ xml -- The XML object to compare against.
+ *other -- The list of XML objects to compare.
+ """
+ if not other:
+ return False
+
+ # Compare multiple objects
+ if len(other) > 1:
+ for xml2 in other:
+ if not self.compare(xml, xml2):
+ return False
+ return True
+
+ other = other[0]
+
+ # Step 1: Check tags
+ if xml.tag != other.tag:
+ return False
+
+ # Step 2: Check attributes
+ if xml.attrib != other.attrib:
+ return False
+
+ # Step 3: Check text
+ if xml.text is None:
+ xml.text = ""
+ if other.text is None:
+ other.text = ""
+ xml.text = xml.text.strip()
+ other.text = other.text.strip()
+
+ if xml.text != other.text:
+ return False
+
+ # Step 4: Check children count
+ if len(xml.getchildren()) != len(other.getchildren()):
+ return False
+
+ # Step 5: Recursively check children
+ for child in xml:
+ child2s = other.findall("%s" % child.tag)
+ if child2s is None:
+ return False
+ for child2 in child2s:
+ if self.compare(child, child2):
+ break
+ else:
+ return False
+
+ # Step 6: Recursively check children the other way.
+ for child in other:
+ child2s = xml.findall("%s" % child.tag)
+ if child2s is None:
+ return False
+ for child2 in child2s:
+ if self.compare(child, child2):
+ break
+ else:
+ return False
+
+ # Everything matches
+ return True