Add whitespace keepalive option.

May be disabled by setting:
    self.whitespace_keepalive = False

The keepalive interval can be adjusted using:
    self.whitespace_keepalive_interval = 300

The default interval is 5min.

.gitignore

@@ -4,3 +4,5 @@ dist/
 MANIFEST
 docs/_build/
 *.swp
+.tox/
+.coverage

MANIFEST.in

@@ -0,0 +1,6 @@
+include README.rst
+include LICENSE
+include testall.py
+recursive-include docs Makefile *.bat *.py *.rst *.css *.ttf *.png
+recursive-include examples *.py
+recursive-include tests *.py

README.rst

@@ -34,7 +34,8 @@ SleekXMPP's design goals and philosphy are:
 
 Get the Code
 ------------
-.. code-block:: sh
+
+Get the latest stable version from PyPI::
 
     pip install sleekxmpp
 
@@ -54,6 +55,15 @@ The latest source code for SleekXMPP may be found on `Github
 **Develop Releases**
     - `Latest Develop Version <http://github.com/fritzy/SleekXMPP/zipball/develop>`_
 
+Installing DNSPython
+---------------------
+If you are using Python3 and wish to use dnspython, you will have to checkout and
+install the ``python3`` branch::
+
+    git clone http://github.com/rthalley/dnspython
+    cd dnspython
+    git checkout python3
+    python3 setup.py install
 
 Discussion
 ----------
@@ -68,7 +78,6 @@ help with SleekXMPP.
 
 Documentation and Testing
 -------------------------
-
 Documentation can be found both inline in the code, and as a Sphinx project in ``/docs``.
 To generate the Sphinx documentation, follow the commands below. The HTML output will
 be in ``docs/_build/html``::
@@ -84,7 +93,6 @@ To run the test suite for SleekXMPP::
 
 The SleekXMPP Boilerplate
 -------------------------
-
 Projects using SleekXMPP tend to follow a basic pattern for setting up client/component
 connections and configuration. Here is the gist of the boilerplate needed for a SleekXMPP
 based project. See the documetation or examples directory for more detailed archetypes for
@@ -136,8 +144,10 @@ SleekXMPP projects::
         xmpp.register_plugin('xep_0199') # XMPP Ping
 
         # If you are working with an OpenFire server, you will need
-        # to useuterborg Larsson version:
-        # xmppissl_version = ssl.PROTOCOL_SSLv3
+        # to use a different SSL version:
+        #
+        # import ssl
+        # xmpp.ssl_version = ssl.PROTOCOL_SSLv3
 
         if xmpp.connect():
             xmpp.process(block=True)

docs/api/clientxmpp.rst

@@ -4,8 +4,6 @@ clientxmpp
 
 .. module:: sleekxmpp.clientxmpp
 
-.. autodata:: SRV_SUPPORT
-
 .. autoclass:: ClientXMPP
     
     .. automethod:: connect

setup.py

@@ -4,16 +4,15 @@
 # Copyright (C) 2007-2011 Nathanael C. Fritz
 # All Rights Reserved
 #
-# This software is licensed as described in the README file,
-# which you should have received as part of this distribution.
-#
+# This software is licensed as described in the README.rst and LICENSE
+# file, which you should have received as part of this distribution.
 
-# from ez_setup import use_setuptools
-from distutils.core import setup
 import sys
+from distutils.core import setup, Command
+# from ez_setup import use_setuptools
 
-import sleekxmpp
-
+from testall import TestCommand
+from sleekxmpp.version import __version__
 # if 'cygwin' in sys.platform.lower():
 #     min_version = '0.6c6'
 # else:
@@ -27,10 +26,10 @@ import sleekxmpp
 #
 # from setuptools import setup, find_packages, Extension, Feature
 
-VERSION          = sleekxmpp.__version__
+VERSION          = __version__
 DESCRIPTION      = 'SleekXMPP is an elegant Python library for XMPP (aka Jabber, Google Talk, etc).'
 with open('README.rst') as readme:
-    LONG_DESCRIPTION = '\n'.join(readme)
+    LONG_DESCRIPTION = ''.join(readme)
 
 CLASSIFIERS      = [ 'Intended Audience :: Developers',
                      'License :: OSI Approved :: MIT License',
@@ -93,5 +92,7 @@ setup(
     license      = 'MIT',
     platforms    = [ 'any' ],
     packages     = packages,
-    requires     = [ 'tlslite', 'pythondns' ],
+    requires     = [ 'dnspython' ],
+    classifiers  = CLASSIFIERS,
+    cmdclass     = {'test': TestCommand}
 )

sleekxmpp/__init__.py

@@ -15,5 +15,4 @@ from sleekxmpp.xmlstream import XMLStream, RestartStream
 from sleekxmpp.xmlstream.matcher import *
 from sleekxmpp.xmlstream.stanzabase import StanzaBase, ET
 
-__version__ = '1.0rc1'
-__version_info__ = (1, 0, 0, 'rc1', 0)
+from sleekxmpp.version import __version__, __version_info__

sleekxmpp/clientxmpp.py

@@ -254,13 +254,6 @@ class ClientXMPP(BaseXMPP):
         self.bindfail = False
         self.features = set()
 
-        def session_timeout():
-            if not self.session_started_event.isSet():
-                log.debug("Session start has taken more than 15 seconds")
-                self.disconnect(reconnect=self.auto_reconnect)
-
-        self.schedule("session timeout checker", 15, session_timeout)
-
     def _handle_stream_features(self, features):
         """
         Process the received stream features.

sleekxmpp/plugins/xep_0050/adhoc.py

@@ -431,8 +431,7 @@ class xep_0050(base_plugin):
         iq = self.xmpp.Iq()
         iq['type'] = 'set'
         iq['to'] = jid
-        if ifrom:
-            iq['from'] = ifrom
+        iq['from'] = ifrom
         iq['command']['node'] = node
         iq['command']['action'] = action
         if sessionid is not None:
@@ -482,9 +481,8 @@ class xep_0050(base_plugin):
         iq = self.xmpp.Iq()
         iq['type'] = 'set'
         iq['to'] = jid
-        if ifrom:
-            iq['from'] = ifrom
-            session['from'] = ifrom
+        iq['from'] = ifrom
+        session['from'] = ifrom
         iq['command']['node'] = node
         iq['command']['action'] = 'execute'
         sessionid = 'client:pending_' + iq['id']

sleekxmpp/plugins/xep_0060/pubsub.py

@@ -1,16 +1,23 @@
-from __future__ import with_statement
-from sleekxmpp.plugins import base
+"""
+    SleekXMPP: The Sleek XMPP Library
+    Copyright (C) 2011  Nathanael C. Fritz
+    This file is part of SleekXMPP.
+
+    See the file LICENSE for copying permission.
+"""
+
 import logging
-#from xml.etree import cElementTree as ET
-from sleekxmpp.xmlstream.stanzabase import registerStanzaPlugin, ElementBase, ET
+
+from sleekxmpp.xmlstream import JID
+from sleekxmpp.plugins.base import base_plugin
 from sleekxmpp.plugins.xep_0060 import stanza
-from sleekxmpp.plugins.xep_0004 import Form
 
 
 log = logging.getLogger(__name__)
 
 
-class xep_0060(base.base_plugin):
+class xep_0060(base_plugin):
+
     """
     XEP-0060 Publish Subscribe
     """
@@ -18,111 +25,404 @@ class xep_0060(base.base_plugin):
     def plugin_init(self):
         self.xep = '0060'
         self.description = 'Publish-Subscribe'
+        self.stanza = stanza
 
-    def create_node(self, jid, node, config=None, ntype=None):
-        iq = IQ(sto=jid, stype='set', sfrom=self.xmpp.jid)
+    def create_node(self, jid, node, config=None, ntype=None, ifrom=None,
+                    block=True, callback=None, timeout=None):
+        """
+        Create and configure a new pubsub node.
+
+        A server MAY use a different name for the node than the one provided,
+        so be sure to check the result stanza for a server assigned name.
+
+        If no configuration form is provided, the node will be created using
+        the server's default configuration. To get the default configuration
+        use get_node_config().
+
+        Arguments:
+            jid      -- The JID of the pubsub service.
+            node     -- Optional name of the node to create. If no name is
+                        provided, the server MAY generate a node ID for you.
+                        The server can also assign a different name than the
+                        one you provide; check the result stanza to see if
+                        the server assigned a name.
+            config   -- Optional XEP-0004 data form of configuration settings.
+            ntype    -- The type of node to create. Servers typically default
+                        to using 'leaf' if no type is provided.
+            ifrom    -- Specify the sender's JID.
+            block    -- Specify if the send call will block until a response
+                        is received, or a timeout occurs. Defaults to True.
+            timeout  -- The length of time (in seconds) to wait for a response
+                        before exiting the send call if blocking is used.
+                        Defaults to sleekxmpp.xmlstream.RESPONSE_TIMEOUT
+            callback -- Optional reference to a stream handler function. Will
+                        be executed when a reply stanza is received.
+        """
+        iq = self.xmpp.Iq(sto=jid, sfrom=ifrom, stype='set')
         iq['pubsub']['create']['node'] = node
-        if ntype is None:
-            ntype = 'leaf'
+
         if config is not None:
-            if 'FORM_TYPE' in submitform.field:
-                config.field['FORM_TYPE'].setValue('http://jabber.org/protocol/pubsub#node_config')
+            form_type = 'http://jabber.org/protocol/pubsub#node_config'
+            if 'FORM_TYPE' in config['fields']:
+                config.field['FORM_TYPE']['value'] = form_type
             else:
-                config.addField('FORM_TYPE', 'hidden', value='http://jabber.org/protocol/pubsub#node_config')
-            if 'pubsub#node_type' in submitform.field:
-                config.field['pubsub#node_type'].setValue(ntype)
-            else:
-                config.addField('pubsub#node_type', value=ntype)
-            iq['pubsub']['configure']['form'] = config
-        return iq.send()
+                config.add_field(var='FORM_TYPE',
+                                 ftype='hidden',
+                                 value=form_type)
+            if ntype:
+                if 'pubsub#node_type' in config['fields']:
+                    config.field['pubsub#node_type']['value'] = ntype
+                else:
+                    config.add_field(var='pubsub#node_type', value=ntype)
+            iq['pubsub']['configure'].append(config)
 
-    def subscribe(self, jid, node, bare=True, subscribee=None):
-        iq = IQ(sto=jid, sfrom=self.xmpp.jid, stype='set')
+        return iq.send(block=block, callback=callback, timeout=timeout)
+
+    def subscribe(self, jid, node, bare=True, subscribee=None, options=None,
+                  ifrom=None, block=True, callback=None, timeout=None):
+        """
+        Subscribe to updates from a pubsub node.
+
+        The rules for determining the JID that is subscribing to the node are:
+            1. If subscribee is given, use that as provided.
+            2. If ifrom was given, use the bare or full version based on bare.
+            3. Otherwise, use self.xmpp.boundjid based on bare.
+
+        Arguments:
+            jid        -- The pubsub service JID.
+            node       -- The node to subscribe to.
+            bare       -- Indicates if the subscribee is a bare or full JID.
+                          Defaults to True for a bare JID.
+            subscribee -- The JID that is subscribing to the node.
+            options    --
+            ifrom      -- Specify the sender's JID.
+            block      -- Specify if the send call will block until a response
+                          is received, or a timeout occurs. Defaults to True.
+            timeout    -- The length of time (in seconds) to wait for a response
+                          before exiting the send call if blocking is used.
+                          Defaults to sleekxmpp.xmlstream.RESPONSE_TIMEOUT
+            callback   -- Optional reference to a stream handler function. Will
+                          be executed when a reply stanza is received.
+        """
+        iq = self.xmpp.Iq(sto=jid, sfrom=ifrom, stype='set')
         iq['pubsub']['subscribe']['node'] = node
-        if subscribee is None:
-            if bare:
-                iq['pubsub']['subscribe']['jid'] = self.xmpp.jid.bare
-            else:
-                iq['pubsub']['subscribe']['jid'] = self.xmpp.jid.full
-        else:
-            iq['pubsub']['subscribe']['jid'] = subscribee
-        return iq.send()
 
-    def unsubscribe(self, jid, node, subid=None, bare=True, subscribee=None):
-        iq = IQ(sto=jid, sfrom=self.xmpp.jid, stype='set')
+        if subscribee is None:
+            if ifrom:
+                if bare:
+                    subscribee = JID(ifrom).bare
+                else:
+                    subscribee = ifrom
+            else:
+                if bare:
+                    subscribee = self.xmpp.boundjid.bare
+                else:
+                    subscribee = self.xmpp.boundjid
+
+        iq['pubsub']['subscribe']['jid'] = subscribee
+        if options is not None:
+            iq['pubsub']['options'].append(options)
+        return iq.send(block=block, callback=callback, timeout=timeout)
+
+    def unsubscribe(self, jid, node, subid=None, bare=True, subscribee=None,
+                    ifrom=None, block=True, callback=None, timeout=None):
+        """
+        Unubscribe from updates from a pubsub node.
+
+        The rules for determining the JID that is unsubscribing
+        from the node are:
+            1. If subscribee is given, use that as provided.
+            2. If ifrom was given, use the bare or full version based on bare.
+            3. Otherwise, use self.xmpp.boundjid based on bare.
+
+        Arguments:
+            jid        -- The pubsub service JID.
+            node       -- The node to subscribe to.
+            subid      -- The specific subscription, if multiple subscriptions
+                          exist for this JID/node combination.
+            bare       -- Indicates if the subscribee is a bare or full JID.
+                          Defaults to True for a bare JID.
+            subscribee -- The JID that is subscribing to the node.
+            ifrom      -- Specify the sender's JID.
+            block      -- Specify if the send call will block until a response
+                          is received, or a timeout occurs. Defaults to True.
+            timeout    -- The length of time (in seconds) to wait for a response
+                          before exiting the send call if blocking is used.
+                          Defaults to sleekxmpp.xmlstream.RESPONSE_TIMEOUT
+            callback   -- Optional reference to a stream handler function. Will
+                          be executed when a reply stanza is received.
+        """
+        iq = self.xmpp.Iq(sto=jid, sfrom=ifrom, stype='set')
         iq['pubsub']['unsubscribe']['node'] = node
-        if subscribee is None:
-            if bare:
-                iq['pubsub']['unsubscribe']['jid'] = self.xmpp.jid.bare
-            else:
-                iq['pubsub']['unsubscribe']['jid'] = self.xmpp.jid.full
-        else:
-            iq['pubsub']['unsubscribe']['jid'] = subscribee
-        if subid is not None:
-            iq['pubsub']['unsubscribe']['subid'] = subid
-        return iq.send()
 
-    def get_node_config(self, jid, node=None): # if no node, then grab default
-        iq = IQ(sto=jid, sfrom=self.xmpp.jid, stype='get')
+        if subscribee is None:
+            if ifrom:
+                if bare:
+                    subscribee = JID(ifrom).bare
+                else:
+                    subscribee = ifrom
+            else:
+                if bare:
+                    subscribee = self.xmpp.boundjid.bare
+                else:
+                    subscribee = self.xmpp.boundjid
+
+        iq['pubsub']['unsubscribe']['jid'] = subscribee
+        iq['pubsub']['unsubscribe']['subid'] = subid
+        return iq.send(block=block, callback=callback, timeout=timeout)
+
+    def get_subscriptions(self, jid, node=None, ifrom=None, block=True,
+                          callback=None, timeout=None):
+        iq = self.xmpp.Iq(sto=jid, sfrom=ifrom, stype='get')
+        iq['pubsub']['subscriptions']['node'] = node
+        return iq.send(block=block, callback=callback, timeout=timeout)
+
+    def get_affiliations(self, jid, node=None, ifrom=None, block=True,
+                         callback=None, timeout=None):
+        iq = self.xmpp.Iq(sto=jid, sfrom=ifrom, stype='get')
+        iq['pubsub']['affiliations']['node'] = node
+        return iq.send(block=block, callback=callback, timeout=timeout)
+
+    def get_subscription_options(self, jid, node=None, user_jid=None, ifrom=None,
+                                 block=True, callback=None, timeout=None):
+        iq = self.xmpp.Iq(sto=jid, sfrom=ifrom, stype='get')
+        if user_jid is None:
+            iq['pubsub']['default']['node'] = node
+        else:
+            iq['pubsub']['options']['node'] = node
+            iq['pubsub']['options']['jid'] = user_jid
+        return iq.send(block=block, callback=callback, timeout=timeout)
+
+    def set_subscription_options(self, jid, node, user_jid, options,
+                                 ifrom=None, block=True, callback=None,
+                                 timeout=None):
+        iq = self.xmpp.Iq(sto=jid, sfrom=ifrom, stype='get')
+        iq['pubsub']['options']['node'] = node
+        iq['pubsub']['options']['jid'] = user_jid
+        iq['pubsub']['options'].append(options)
+        return iq.send(block=block, callback=callback, timeout=timeout)
+
+    def get_node_config(self, jid, node=None, ifrom=None, block=None,
+                        callback=None, timeout=None):
+        """
+        Retrieve the configuration for a node, or the pubsub service's
+        default configuration for new nodes.
+
+        Arguments:
+            jid      -- The JID of the pubsub service.
+            node     -- The node to retrieve the configuration for. If None,
+                        the default configuration for new nodes will be
+                        requested. Defaults to None.
+            ifrom    -- Specify the sender's JID.
+            block    -- Specify if the send call will block until a response
+                        is received, or a timeout occurs. Defaults to True.
+            timeout  -- The length of time (in seconds) to wait for a response
+                        before exiting the send call if blocking is used.
+                        Defaults to sleekxmpp.xmlstream.RESPONSE_TIMEOUT
+            callback -- Optional reference to a stream handler function. Will
+                        be executed when a reply stanza is received.
+        """
+        iq = self.xmpp.Iq(sto=jid, sfrom=ifrom, stype='get')
         if node is None:
             iq['pubsub_owner']['default']
         else:
             iq['pubsub_owner']['configure']['node'] = node
-        return iq.send()
+        return iq.send(block=block, callback=callback, timeout=timeout)
 
-    def get_node_subscriptions(self, jid, node):
-        iq = IQ(sto=jid, sfrom=self.xmpp.jid, stype='get')
+    def get_node_subscriptions(self, jid, node, ifrom=None, block=True,
+                               callback=None, timeout=None):
+        """
+        Retrieve the subscriptions associated with a given node.
+
+        Arguments:
+            jid      -- The JID of the pubsub service.
+            node     -- The node to retrieve subscriptions from.
+            ifrom    -- Specify the sender's JID.
+            block    -- Specify if the send call will block until a response
+                        is received, or a timeout occurs. Defaults to True.
+            timeout  -- The length of time (in seconds) to wait for a response
+                        before exiting the send call if blocking is used.
+                        Defaults to sleekxmpp.xmlstream.RESPONSE_TIMEOUT
+            callback -- Optional reference to a stream handler function. Will
+                        be executed when a reply stanza is received.
+        """
+        iq = self.xmpp.Iq(sto=jid, sfrom=ifrom, stype='get')
         iq['pubsub_owner']['subscriptions']['node'] = node
-        return iq.send()
+        return iq.send(block=block, callback=callback, timeout=timeout)
 
-    def get_node_affiliations(self, jid, node):
-        iq = IQ(sto=jid, sfrom=self.xmpp.jid, stype='get')
+    def get_node_affiliations(self, jid, node, ifrom=None, block=True,
+                              callback=None, timeout=None):
+        """
+        Retrieve the affiliations associated with a given node.
+
+        Arguments:
+            jid      -- The JID of the pubsub service.
+            node     -- The node to retrieve affiliations from.
+            ifrom    -- Specify the sender's JID.
+            block    -- Specify if the send call will block until a response
+                        is received, or a timeout occurs. Defaults to True.
+            timeout  -- The length of time (in seconds) to wait for a response
+                        before exiting the send call if blocking is used.
+                        Defaults to sleekxmpp.xmlstream.RESPONSE_TIMEOUT
+            callback -- Optional reference to a stream handler function. Will
+                        be executed when a reply stanza is received.
+        """
+        iq = self.xmpp.Iq(sto=jid, sfrom=ifrom, stype='get')
         iq['pubsub_owner']['affiliations']['node'] = node
-        return iq.send()
+        return iq.send(block=block, callback=callback, timeout=timeout)
 
-    def delete_node(self, jid, node):
-        iq = IQ(sto=jid, sfrom=self.xmpp.jid, stype='get')
+    def delete_node(self, jid, node, ifrom=None, block=True,
+                    callback=None, timeout=None):
+        """
+        Delete a a pubsub node.
+
+        Arguments:
+            jid      -- The JID of the pubsub service.
+            node     -- The node to delete.
+            ifrom    -- Specify the sender's JID.
+            block    -- Specify if the send call will block until a response
+                        is received, or a timeout occurs. Defaults to True.
+            timeout  -- The length of time (in seconds) to wait for a response
+                        before exiting the send call if blocking is used.
+                        Defaults to sleekxmpp.xmlstream.RESPONSE_TIMEOUT
+            callback -- Optional reference to a stream handler function. Will
+                        be executed when a reply stanza is received.
+        """
+        iq = self.xmpp.Iq(sto=jid, sfrom=ifrom, stype='set')
         iq['pubsub_owner']['delete']['node'] = node
-        return iq.send()
+        return iq.send(block=block, callback=callback, timeout=timeout)
 
-    def set_node_config(self, jid, node, config):
-        iq = IQ(sto=jid, sfrom=self.xmpp.jid, stype='set')
+    def set_node_config(self, jid, node, config, ifrom=None, block=True,
+                        callback=None, timeout=None):
+        iq = self.xmpp.Iq(sto=jid, sfrom=ifrom, stype='set')
         iq['pubsub_owner']['configure']['node'] = node
-        iq['pubsub_owner']['configure']['config'] = config
-        return iq.send()
+        iq['pubsub_owner']['configure']['form'].values = config.values
+        return iq.send(block=block, callback=callback, timeout=timeout)
 
-    def publish(self, jid, node, items=[]):
-        iq = IQ(sto=jid, sfrom=self.xmpp.jid, stype='set')
+    def publish(self, jid, node, id=None, payload=None, options=None,
+                ifrom=None, block=True, callback=None, timeout=None):
+        """
+        Add or edit items in a node.
+
+        You may publish an individual item using the item_id and payload
+        parameters, or you may batch publish by using the items parameter
+        which accepts a list of id/payload tuples.
+        """
+        iq = self.xmpp.Iq(sto=jid, sfrom=ifrom, stype='set')
         iq['pubsub']['publish']['node'] = node
-        for id, payload in items:
-            item = stanza.pubsub.Item()
-            if id is not None:
-                item['id'] = id
-            item['payload'] = payload
-            iq['pubsub']['publish'].append(item)
-        return iq.send()
+        if id is not None:
+            iq['pubsub']['publish']['item']['id'] = id
+        if payload is not None:
+            iq['pubsub']['publish']['item']['payload'] = payload
+        iq['pubsub']['publish_options'] = options
+        return iq.send(block=block, callback=callback, timeout=timeout)
+
+    def retract(self, jid, node, id, notify=None, ifrom=None, block=True,
+                callback=None, timeout=None):
+        """
+        Delete a single item from a node.
+        """
+        iq = self.xmpp.Iq(sto=jid, sfrom=ifrom, stype='set')
 
-    def retract(self, jid, node, item):
-        iq = IQ(sto=jid, sfrom=self.xmpp.jid, stype='set')
         iq['pubsub']['retract']['node'] = node
-        item = stanza.pubsub.Item()
-        item['id'] = item
-        iq['pubsub']['retract'].append(item)
-        return iq.send()
+        iq['pubsub']['retract']['notify'] = notify
+        iq['pubsub']['retract']['item']['id'] = id
+        return iq.send(block=block, callback=callback, timeout=timeout)
 
-    def get_nodes(self, jid):
-        return self.xmpp.plugin['xep_0030'].get_items(jid)
+    def purge(self, jid, node, ifrom=None, block=True, callback=None,
+              timeout=None):
+        """
+        Remove all items from a node.
+        """
+        iq = self.xmpp.Iq(sto=jid, sfrom=ifrom, stype='set')
+        iq['pubsub_owner']['purge']['node'] = node
+        return iq.send(block=block, callback=callback, timeout=timeout)
 
-    def getItems(self, jid, node):
-        return self.xmpp.plugin['xep_0030'].get_items(jid, node)
+    def get_nodes(self, *args, **kwargs):
+        """
+        Discover the nodes provided by a Pubsub service, using disco.
+        """
+        return self.xmpp.plugin['xep_0030'].get_items(*args, **kwargs)
 
-    def modify_affiliation(self, jid, node, affiliation, user_jid=None):
-        iq = IQ(sto=jid, sfrom=self.xmpp.jid, stype='set')
-        iq['pubsub_owner']['affiliations']
-        aff = stanza.pubsub.Affiliation()
-        aff['node'] = node
-        if user_jid is not None:
-            aff['jid'] = user_jid
-        aff['affiliation'] = affiliation
-        iq['pubsub_owner']['affiliations'].append(aff)
-        return iq.send()
+    def get_item(self, jid, node, item_id, ifrom=None, block=True,
+                 callback=None, timeout=None):
+        """
+        Retrieve the content of an individual item.
+        """
+        iq = self.xmpp.Iq(sto=jid, sfrom=ifrom, stype='get')
+        item = self.stanza.Item()
+        item['id'] = item_id
+        iq['pubsub']['items']['node'] = node
+        iq['pubsub']['items'].append(item)
+        return iq.send(block=block, callback=callback, timeout=timeout)
+
+    def get_items(self, jid, node, item_ids=None, max_items=None,
+                  iterator=False, ifrom=None, block=False,
+                  callback=None, timeout=None):
+        """
+        Request the contents of a node's items.
+
+        The desired items can be specified, or a query for the last
+        few published items can be used.
+
+        Pubsub services may use result set management for nodes with
+        many items, so an iterator can be returned if needed.
+        """
+        iq = self.xmpp.Iq(sto=jid, sfrom=ifrom, stype='get')
+        iq['pubsub']['items']['node'] = node
+        iq['pubsub']['items']['max_items'] = max_items
+
+        if item_ids is not None:
+            for item_id in item_ids:
+                item = self.stanza.Item()
+                item['id'] = item_id
+                iq['pubsub']['items'].append(item)
+
+        if iterator:
+            return self.xmpp['xep_0059'].iterate(iq, 'pubsub')
+        else:
+            return iq.send(block=block, callback=callback, timeout=timeout)
+
+    def get_item_ids(self, jid, node, ifrom=None, block=True,
+                     callback=None, timeout=None, iterator=False):
+        """
+        Retrieve the ItemIDs hosted by a given node, using disco.
+        """
+        return self.xmpp.plugin['xep_0030'].get_items(jid, node,
+                                                      ifrom=ifrom,
+                                                      block=block,
+                                                      callback=callback,
+                                                      timeout=timeout,
+                                                      iterator=iterator)
+
+    def modify_affiliations(self, jid, node, affiliations=None, ifrom=None,
+                            block=True, callback=None, timeout=None):
+        iq = self.xmpp.Iq(sto=jid, sfrom=ifrom, stype='set')
+        iq['pubsub_owner']['affiliations']['node'] = node
+
+        if affiliations is None:
+            affiliations = []
+
+        for jid, affiliation in affiliations:
+            aff = self.stanza.OwnerAffiliation()
+            aff['jid'] = jid
+            aff['affiliation'] = affiliation
+            iq['pubsub_owner']['affiliations'].append(aff)
+
+        return iq.send(block=block, callback=callback, timeout=timeout)
+
+    def modify_subscriptions(self, jid, node, subscriptions=None, ifrom=None,
+                             block=True, callback=None, timeout=None):
+        iq = self.xmpp.Iq(sto=jid, sfrom=ifrom, stype='set')
+        iq['pubsub_owner']['subscriptions']['node'] = node
+
+        if subscriptions is None:
+            subscriptions = []
+
+        for jid, subscription in subscriptions:
+            sub = self.stanza.OwnerSubscription()
+            sub['jid'] = jid
+            sub['subscription'] = subscription
+            iq['pubsub_owner']['subscriptions'].append(sub)
+
+        return iq.send(block=block, callback=callback, timeout=timeout)

sleekxmpp/plugins/xep_0060/stanza/__init__.py

@@ -1,3 +1,12 @@
-from sleekxmpp.plugins.xep_0060.stanza.pubsub import Pubsub, Affiliation, Affiliations, Subscription, Subscriptions, SubscribeOptions, Item, Items, Create, Publish, Retract, Unsubscribe, Subscribe, Configure, Options, PubsubState, PubsubStateEvent
-from sleekxmpp.plugins.xep_0060.stanza.pubsub_owner import PubsubOwner, DefaultConfig, OwnerAffiliations, OwnerAffiliation, OwnerConfigure, OwnerDefault, OwnerDelete, OwnerPurge, OwnerRedirect, OwnerSubscriptions, OwnerSubscription
-from sleekxmpp.plugins.xep_0060.stanza.pubsub_event import Event, EventItem, EventRetract, EventItems, EventCollection, EventAssociate, EventDisassociate, EventConfiguration, EventPurge, EventSubscription
+"""
+    SleekXMPP: The Sleek XMPP Library
+    Copyright (C) 2011  Nathanael C. Fritz
+    This file is part of SleekXMPP.
+
+    See the file LICENSE for copying permission.
+"""
+
+from sleekxmpp.plugins.xep_0060.stanza.pubsub import *
+from sleekxmpp.plugins.xep_0060.stanza.pubsub_owner import *
+from sleekxmpp.plugins.xep_0060.stanza.pubsub_event import *
+from sleekxmpp.plugins.xep_0060.stanza.pubsub_errors import *

sleekxmpp/plugins/xep_0060/stanza/base.py

@@ -1,24 +1,29 @@
-from xml.etree import cElementTree as ET
+"""
+    SleekXMPP: The Sleek XMPP Library
+    Copyright (C) 2011  Nathanael C. Fritz
+    This file is part of SleekXMPP.
+
+    See the file LICENSE for copying permission.
+"""
+
+from sleekxmpp.xmlstream import ET
+
 
 class OptionalSetting(object):
-	interfaces = set(('required',))
 
-	def setRequired(self, value):
-		value = bool(value)
-		if value and not self['required']:
-			self.xml.append(ET.Element("{%s}required" % self.namespace))
-		elif not value and self['required']:
-			self.delRequired()
-	
-	def getRequired(self):
-		required = self.xml.find("{%s}required" % self.namespace)
-		if required is not None:
-			return True
-		else:
-			return False
-	
-	def delRequired(self):
-		required = self.xml.find("{%s}required" % self.namespace)
-		if required is not None:
-			self.xml.remove(required)
+    interfaces = set(('required',))
 
+    def set_required(self, value):
+        if value in (True, 'true', 'True', '1'):
+            self.xml.append(ET.Element("{%s}required" % self.namespace))
+        elif self['required']:
+            self.del_required()
+
+    def get_required(self):
+        required = self.xml.find("{%s}required" % self.namespace)
+        return required is not None
+
+    def del_required(self):
+        required = self.xml.find("{%s}required" % self.namespace)
+        if required is not None:
+            self.xml.remove(required)

sleekxmpp/plugins/xep_0060/stanza/pubsub.py

@@ -1,9 +1,13 @@
-from sleekxmpp.xmlstream.stanzabase import registerStanzaPlugin, ElementBase, ET, JID
-from sleekxmpp.stanza.iq import Iq
-from sleekxmpp.stanza.message import Message
-from sleekxmpp.basexmpp import basexmpp
-from sleekxmpp.xmlstream.xmlstream import XMLStream
-import logging
+"""
+    SleekXMPP: The Sleek XMPP Library
+    Copyright (C) 2011  Nathanael C. Fritz
+    This file is part of SleekXMPP.
+
+    See the file LICENSE for copying permission.
+"""
+
+from sleekxmpp import Iq, Message
+from sleekxmpp.xmlstream import register_stanza_plugin, ElementBase, ET, JID
 from sleekxmpp.plugins import xep_0004
 from sleekxmpp.plugins.xep_0060.stanza.base import OptionalSetting
 
@@ -11,12 +15,15 @@ from sleekxmpp.plugins.xep_0060.stanza.base import OptionalSetting
 class Pubsub(ElementBase):
     namespace = 'http://jabber.org/protocol/pubsub'
     name = 'pubsub'
-    plugin_attrib = 'pubsub'
+    plugin_attrib = name
     interfaces = set(tuple())
-    plugin_attrib_map = {}
-    plugin_tag_map = {}
 
-registerStanzaPlugin(Iq, Pubsub)
+
+class Affiliations(ElementBase):
+    namespace = 'http://jabber.org/protocol/pubsub'
+    name = 'affiliations'
+    plugin_attrib = name
+    interfaces = set(('node',))
 
 
 class Affiliation(ElementBase):
@@ -24,25 +31,12 @@ class Affiliation(ElementBase):
     name = 'affiliation'
     plugin_attrib = name
     interfaces = set(('node', 'affiliation', 'jid'))
-    plugin_attrib_map = {}
-    plugin_tag_map = {}
-    
-    def setJid(self, value):
-        self._setAttr('jid', str(value))
 
-    def getJid(self):
-        return JID(self._getAttr('jid'))
+    def set_jid(self, value):
+        self._set_attr('jid', str(value))
 
-class Affiliations(ElementBase):
-    namespace = 'http://jabber.org/protocol/pubsub'
-    name = 'affiliations'
-    plugin_attrib = 'affiliations'
-    interfaces = set(tuple())
-    plugin_attrib_map = {}
-    plugin_tag_map = {}
-    subitem = (Affiliation,)
-
-registerStanzaPlugin(Pubsub, Affiliations)
+    def get_jid(self):
+        return JID(self._get_attr('jid'))
 
 
 class Subscription(ElementBase):
@@ -50,228 +44,257 @@ class Subscription(ElementBase):
     name = 'subscription'
     plugin_attrib = name
     interfaces = set(('jid', 'node', 'subscription', 'subid'))
-    plugin_attrib_map = {}
-    plugin_tag_map = {}
 
-    def setjid(self, value):
-        self._setattr('jid', str(value))
+    def set_jid(self, value):
+        self._set_attr('jid', str(value))
 
-    def getjid(self):
-        return jid(self._getattr('jid'))
+    def get_jid(self):
+        return JID(self._get_attr('jid'))
 
-registerStanzaPlugin(Pubsub, Subscription)
 
 class Subscriptions(ElementBase):
     namespace = 'http://jabber.org/protocol/pubsub'
     name = 'subscriptions'
-    plugin_attrib = 'subscriptions'
-    interfaces = set(tuple())
-    plugin_attrib_map = {}
-    plugin_tag_map = {}
-    subitem = (Subscription,)
-
-registerStanzaPlugin(Pubsub, Subscriptions)
+    plugin_attrib = name
+    interfaces = set(('node',))
 
 
 class SubscribeOptions(ElementBase, OptionalSetting):
     namespace = 'http://jabber.org/protocol/pubsub'
     name = 'subscribe-options'
     plugin_attrib = 'suboptions'
-    plugin_attrib_map = {}
-    plugin_tag_map = {}
     interfaces = set(('required',))
 
-registerStanzaPlugin(Subscription, SubscribeOptions)
 
 class Item(ElementBase):
     namespace = 'http://jabber.org/protocol/pubsub'
     name = 'item'
     plugin_attrib = name
     interfaces = set(('id', 'payload'))
-    plugin_attrib_map = {}
-    plugin_tag_map = {}
 
-    def setPayload(self, value):
-        self.xml.append(value)
+    def set_payload(self, value):
+        del self['payload']
+        self.append(value)
 
-    def getPayload(self):
+    def get_payload(self):
         childs = self.xml.getchildren()
         if len(childs) > 0:
             return childs[0]
 
-    def delPayload(self):
+    def del_payload(self):
         for child in self.xml.getchildren():
             self.xml.remove(child)
 
+
 class Items(ElementBase):
     namespace = 'http://jabber.org/protocol/pubsub'
     name = 'items'
-    plugin_attrib = 'items'
-    interfaces = set(('node',))
-    plugin_attrib_map = {}
-    plugin_tag_map = {}
-    subitem = (Item,)
+    plugin_attrib = name
+    interfaces = set(('node', 'max_items'))
+
+    def set_max_items(self, value):
+        self._set_attr('max_items', str(value))
 
-registerStanzaPlugin(Pubsub, Items)
 
 class Create(ElementBase):
     namespace = 'http://jabber.org/protocol/pubsub'
     name = 'create'
     plugin_attrib = name
     interfaces = set(('node',))
-    plugin_attrib_map = {}
-    plugin_tag_map = {}
 
-registerStanzaPlugin(Pubsub, Create)
 
-#class Default(ElementBase):
-#    namespace = 'http://jabber.org/protocol/pubsub'
-#    name = 'default'
-#    plugin_attrib = name
-#    interfaces = set(('node', 'type'))
-#    plugin_attrib_map = {}
-#    plugin_tag_map = {}
-#
-#    def getType(self):
-#        t = self._getAttr('type')
-#        if not t: t == 'leaf'
-#        return t
-#
-#registerStanzaPlugin(Pubsub, Default)
+class Default(ElementBase):
+    namespace = 'http://jabber.org/protocol/pubsub'
+    name = 'default'
+    plugin_attrib = name
+    interfaces = set(('node', 'type'))
 
-class Publish(Items):
+    def get_type(self):
+        t = self._get_attr('type')
+        if not t:
+            return 'leaf'
+        return t
+
+
+class Publish(ElementBase):
     namespace = 'http://jabber.org/protocol/pubsub'
     name = 'publish'
     plugin_attrib = name
     interfaces = set(('node',))
-    plugin_attrib_map = {}
-    plugin_tag_map = {}
-    subitem = (Item,)
 
-registerStanzaPlugin(Pubsub, Publish)
 
-class Retract(Items):
+class Retract(ElementBase):
     namespace = 'http://jabber.org/protocol/pubsub'
     name = 'retract'
     plugin_attrib = name
     interfaces = set(('node', 'notify'))
-    plugin_attrib_map = {}
-    plugin_tag_map = {}
 
-registerStanzaPlugin(Pubsub, Retract)
+    def get_notify(self):
+        notify = self._get_attr('notify')
+        if notify in ('0', 'false'):
+            return False
+        elif notify in ('1', 'true'):
+            return True
+        return None
+
+    def set_notify(self, value):
+        del self['notify']
+        if value is None:
+            return
+        elif value in (True, '1', 'true', 'True'):
+            self._set_attr('notify', 'true')
+        else:
+            self._set_attr('notify', 'false')
+
 
 class Unsubscribe(ElementBase):
     namespace = 'http://jabber.org/protocol/pubsub'
     name = 'unsubscribe'
     plugin_attrib = name
     interfaces = set(('node', 'jid', 'subid'))
-    plugin_attrib_map = {}
-    plugin_tag_map = {}
 
-    def setJid(self, value):
-        self._setAttr('jid', str(value))
+    def set_jid(self, value):
+        self._set_attr('jid', str(value))
 
-    def getJid(self):
-        return JID(self._getAttr('jid'))
+    def get_jid(self):
+        return JID(self._get_attr('jid'))
 
-registerStanzaPlugin(Pubsub, Unsubscribe)
 
 class Subscribe(ElementBase):
     namespace = 'http://jabber.org/protocol/pubsub'
     name = 'subscribe'
     plugin_attrib = name
     interfaces = set(('node', 'jid'))
-    plugin_attrib_map = {}
-    plugin_tag_map = {}
 
-    def setJid(self, value):
-        self._setAttr('jid', str(value))
+    def set_jid(self, value):
+        self._set_attr('jid', str(value))
 
-    def getJid(self):
-        return JID(self._getAttr('jid'))
+    def get_jid(self):
+        return JID(self._get_attr('jid'))
 
-registerStanzaPlugin(Pubsub, Subscribe)
 
 class Configure(ElementBase):
     namespace = 'http://jabber.org/protocol/pubsub'
     name = 'configure'
     plugin_attrib = name
     interfaces = set(('node', 'type'))
-    plugin_attrib_map = {}
-    plugin_tag_map = {}
 
     def getType(self):
-        t = self._getAttr('type')
-        if not t: t == 'leaf'
+        t = self._get_attr('type')
+        if not t:
+            t == 'leaf'
         return t
 
-registerStanzaPlugin(Pubsub, Configure)
-registerStanzaPlugin(Configure, xep_0004.Form)
 
 class Options(ElementBase):
     namespace = 'http://jabber.org/protocol/pubsub'
     name = 'options'
-    plugin_attrib = 'options'
+    plugin_attrib = name
     interfaces = set(('jid', 'node', 'options'))
-    plugin_attrib_map = {}
-    plugin_tag_map = {}
 
     def __init__(self, *args, **kwargs):
         ElementBase.__init__(self, *args, **kwargs)
 
-    def getOptions(self):
+    def get_options(self):
         config = self.xml.find('{jabber:x:data}x')
-        form = xep_0004.Form()
-        if config is not None:
-            form.fromXML(config)
+        form = xep_0004.Form(xml=config)
         return form
 
-    def setOptions(self, value):
+    def set_options(self, value):
         self.xml.append(value.getXML())
         return self
 
-    def delOptions(self):
+    def del_options(self):
         config = self.xml.find('{jabber:x:data}x')
         self.xml.remove(config)
 
-    def setJid(self, value):
-        self._setAttr('jid', str(value))
+    def set_jid(self, value):
+        self._set_attr('jid', str(value))
 
-    def getJid(self):
-        return JID(self._getAttr('jid'))
+    def get_jid(self):
+        return JID(self._get_attr('jid'))
+
+
+class PublishOptions(ElementBase):
+    namespace = 'http://jabber.org/protocol/pubsub'
+    name = 'publish-options'
+    plugin_attrib = 'publish_options'
+    interfaces = set(('publish_options',))
+    is_extension = True
+
+    def get_publish_options(self):
+        config = self.xml.find('{jabber:x:data}x')
+        if config is None:
+            return None
+        form = xep_0004.Form(xml=config)
+        return form
+
+    def set_publish_options(self, value):
+        if value is None:
+            self.del_publish_options()
+        else:
+            self.xml.append(value.getXML())
+        return self
+
+    def del_publish_options(self):
+        config = self.xml.find('{jabber:x:data}x')
+        if config is not None:
+            self.xml.remove(config)
+        self.parent().xml.remove(self.xml)
 
-registerStanzaPlugin(Pubsub, Options)
-registerStanzaPlugin(Subscribe, Options)
 
 class PubsubState(ElementBase):
+    """This is an experimental pubsub extension."""
     namespace = 'http://jabber.org/protocol/psstate'
     name = 'state'
     plugin_attrib = 'psstate'
     interfaces = set(('node', 'item', 'payload'))
-    plugin_attrib_map = {}
-    plugin_tag_map = {}
 
-    def setPayload(self, value):
+    def set_payload(self, value):
         self.xml.append(value)
 
-    def getPayload(self):
+    def get_payload(self):
         childs = self.xml.getchildren()
         if len(childs) > 0:
             return childs[0]
 
-    def delPayload(self):
+    def del_payload(self):
         for child in self.xml.getchildren():
             self.xml.remove(child)
 
-registerStanzaPlugin(Iq, PubsubState)
 
 class PubsubStateEvent(ElementBase):
+    """This is an experimental pubsub extension."""
     namespace = 'http://jabber.org/protocol/psstate#event'
     name = 'event'
     plugin_attrib = 'psstate_event'
     intefaces = set(tuple())
-    plugin_attrib_map = {}
-    plugin_tag_map = {}
 
-registerStanzaPlugin(Message, PubsubStateEvent)
-registerStanzaPlugin(PubsubStateEvent, PubsubState)
+
+register_stanza_plugin(Iq, PubsubState)
+register_stanza_plugin(Message, PubsubStateEvent)
+register_stanza_plugin(PubsubStateEvent, PubsubState)
+
+
+register_stanza_plugin(Iq, Pubsub)
+register_stanza_plugin(Pubsub, Affiliations)
+register_stanza_plugin(Pubsub, Configure)
+register_stanza_plugin(Pubsub, Create)
+register_stanza_plugin(Pubsub, Default)
+register_stanza_plugin(Pubsub, Items)
+register_stanza_plugin(Pubsub, Options)
+register_stanza_plugin(Pubsub, Publish)
+register_stanza_plugin(Pubsub, PublishOptions)
+register_stanza_plugin(Pubsub, Retract)
+register_stanza_plugin(Pubsub, Subscribe)
+register_stanza_plugin(Pubsub, Subscription)
+register_stanza_plugin(Pubsub, Subscriptions)
+register_stanza_plugin(Pubsub, Unsubscribe)
+register_stanza_plugin(Affiliations, Affiliation, iterable=True)
+register_stanza_plugin(Configure, xep_0004.Form)
+register_stanza_plugin(Items, Item, iterable=True)
+register_stanza_plugin(Publish, Item, iterable=True)
+register_stanza_plugin(Retract, Item)
+register_stanza_plugin(Subscribe, Options)
+register_stanza_plugin(Subscription, SubscribeOptions)
+register_stanza_plugin(Subscriptions, Subscription, iterable=True)

sleekxmpp/plugins/xep_0060/stanza/pubsub_errors.py

@@ -0,0 +1,86 @@
+"""
+    SleekXMPP: The Sleek XMPP Library
+    Copyright (C) 2011  Nathanael C. Fritz
+    This file is part of SleekXMPP.
+
+    See the file LICENSE for copying permission.
+"""
+
+from sleekxmpp.stanza import Error
+from sleekxmpp.xmlstream import ElementBase, ET, register_stanza_plugin
+
+
+class PubsubErrorCondition(ElementBase):
+
+    plugin_attrib = 'pubsub'
+    interfaces = set(('condition', 'unsupported'))
+    plugin_attrib_map = {}
+    plugin_tag_map = {}
+    conditions = set(('closed-node', 'configuration-required', 'invalid-jid',
+                      'invalid-options', 'invalid-payload', 'invalid-subid',
+                      'item-forbidden', 'item-required', 'jid-required',
+                      'max-items-exceeded', 'max-nodes-exceeded',
+                      'nodeid-required', 'not-in-roster-group',
+                      'not-subscribed', 'payload-too-big',
+                      'payload-required' 'pending-subscription',
+                      'presence-subscription-required', 'subid-required',
+                      'too-many-subscriptions', 'unsupported'))
+    condition_ns = 'http://jabber.org/protocol/pubsub#errors'
+
+    def setup(self, xml):
+        """Don't create XML for the plugin."""
+        self.xml = ET.Element('')
+
+    def get_condition(self):
+        """Return the condition element's name."""
+        for child in self.parent().xml.getchildren():
+            if "{%s}" % self.condition_ns in child.tag:
+                cond = child.tag.split('}', 1)[-1]
+                if cond in self.conditions:
+                    return cond
+        return ''
+
+    def set_condition(self, value):
+        """
+        Set the tag name of the condition element.
+
+        Arguments:
+           value -- The tag name of the condition element.
+        """
+        if value in self.conditions:
+            del self['condition']
+            cond = ET.Element("{%s}%s" % (self.condition_ns, value))
+            self.parent().xml.append(cond)
+        return self
+
+    def del_condition(self):
+        """Remove the condition element."""
+        for child in self.parent().xml.getchildren():
+            if "{%s}" % self.condition_ns in child.tag:
+                tag = child.tag.split('}', 1)[-1]
+                if tag in self.conditions:
+                    self.parent().xml.remove(child)
+        return self
+
+    def get_unsupported(self):
+        """Return the name of an unsupported feature"""
+        xml = self.parent().xml.find('{%s}unsupported' % self.condition_ns)
+        if xml is not None:
+            return xml.attrib.get('feature', '')
+        return ''
+
+    def set_unsupported(self, value):
+        """Mark a feature as unsupported"""
+        self.del_unsupported()
+        xml = ET.Element('{%s}unsupported' % self.condition_ns)
+        xml.attrib['feature'] = value
+        self.parent().xml.append(xml)
+
+    def del_unsupported(self):
+        """Delete an unsupported feature condition."""
+        xml = self.parent().xml.find('{%s}unsupported' % self.condition_ns)
+        if xml is not None:
+            self.parent().xml.remove(xml)
+
+
+register_stanza_plugin(Error, PubsubErrorCondition)

sleekxmpp/plugins/xep_0060/stanza/pubsub_event.py

@@ -1,124 +1,112 @@
-from sleekxmpp.xmlstream.stanzabase import registerStanzaPlugin, ElementBase, ET, JID
-from sleekxmpp.stanza.iq import Iq
-from sleekxmpp.stanza.message import Message
-from sleekxmpp.basexmpp import basexmpp
-from sleekxmpp.xmlstream.xmlstream import XMLStream
-import logging
-from sleekxmpp.plugins import xep_0004
+"""
+    SleekXMPP: The Sleek XMPP Library
+    Copyright (C) 2011  Nathanael C. Fritz
+    This file is part of SleekXMPP.
+
+    See the file LICENSE for copying permission.
+"""
+
+from sleekxmpp import Message
+from sleekxmpp.xmlstream import register_stanza_plugin, ElementBase, ET, JID
+from sleekxmpp.plugins.xep_0004 import Form
+
 
 class Event(ElementBase):
-	namespace = 'http://jabber.org/protocol/pubsub#event'
-	name = 'event'
-	plugin_attrib = 'pubsub_event'
-	interfaces = set(('node',))
-	plugin_attrib_map = {}
-	plugin_tag_map = {}
+    namespace = 'http://jabber.org/protocol/pubsub#event'
+    name = 'event'
+    plugin_attrib = 'pubsub_event'
+    interfaces = set(('node',))
 
-registerStanzaPlugin(Message, Event)
 
 class EventItem(ElementBase):
-	namespace = 'http://jabber.org/protocol/pubsub#event'
-	name = 'item'
-	plugin_attrib = 'item'
-	interfaces = set(('id', 'payload'))
-	plugin_attrib_map = {}
-	plugin_tag_map = {}
+    namespace = 'http://jabber.org/protocol/pubsub#event'
+    name = 'item'
+    plugin_attrib = name
+    interfaces = set(('id', 'payload'))
 
-	def setPayload(self, value):
-		self.xml.append(value)
-	
-	def getPayload(self):
-		childs = self.xml.getchildren()
-		if len(childs) > 0:
-			return childs[0]
-	
-	def delPayload(self):
-		for child in self.xml.getchildren():
-			self.xml.remove(child)
+    def set_payload(self, value):
+        self.xml.append(value)
+
+    def get_payload(self):
+        childs = self.xml.getchildren()
+        if len(childs) > 0:
+            return childs[0]
+
+    def del_payload(self):
+        for child in self.xml.getchildren():
+            self.xml.remove(child)
 
 
 class EventRetract(ElementBase):
-	namespace = 'http://jabber.org/protocol/pubsub#event'
-	name = 'retract'
-	plugin_attrib = 'retract'
-	interfaces = set(('id',))
-	plugin_attrib_map = {}
-	plugin_tag_map = {}
+    namespace = 'http://jabber.org/protocol/pubsub#event'
+    name = 'retract'
+    plugin_attrib = name
+    interfaces = set(('id',))
+
 
 class EventItems(ElementBase):
-	namespace = 'http://jabber.org/protocol/pubsub#event'
-	name = 'items'
-	plugin_attrib = 'items'
-	interfaces = set(('node',))
-	plugin_attrib_map = {}
-	plugin_tag_map = {}
-	subitem = (EventItem, EventRetract)
+    namespace = 'http://jabber.org/protocol/pubsub#event'
+    name = 'items'
+    plugin_attrib = name
+    interfaces = set(('node',))
 
-registerStanzaPlugin(Event, EventItems)
 
 class EventCollection(ElementBase):
-	namespace = 'http://jabber.org/protocol/pubsub#event'
-	name = 'collection'
-	plugin_attrib = name
-	interfaces = set(('node',))
-	plugin_attrib_map = {}
-	plugin_tag_map = {}
+    namespace = 'http://jabber.org/protocol/pubsub#event'
+    name = 'collection'
+    plugin_attrib = name
+    interfaces = set(('node',))
 
-registerStanzaPlugin(Event, EventCollection)
 
 class EventAssociate(ElementBase):
-	namespace = 'http://jabber.org/protocol/pubsub#event'
-	name = 'associate'
-	plugin_attrib = name
-	interfaces = set(('node',))
-	plugin_attrib_map = {}
-	plugin_tag_map = {}
+    namespace = 'http://jabber.org/protocol/pubsub#event'
+    name = 'associate'
+    plugin_attrib = name
+    interfaces = set(('node',))
 
-registerStanzaPlugin(EventCollection, EventAssociate)
 
 class EventDisassociate(ElementBase):
-	namespace = 'http://jabber.org/protocol/pubsub#event'
-	name = 'disassociate'
-	plugin_attrib = name
-	interfaces = set(('node',))
-	plugin_attrib_map = {}
-	plugin_tag_map = {}
+    namespace = 'http://jabber.org/protocol/pubsub#event'
+    name = 'disassociate'
+    plugin_attrib = name
+    interfaces = set(('node',))
 
-registerStanzaPlugin(EventCollection, EventDisassociate)
 
 class EventConfiguration(ElementBase):
-	namespace = 'http://jabber.org/protocol/pubsub#event'
-	name = 'configuration'
-	plugin_attrib = name
-	interfaces = set(('node', 'config'))
-	plugin_attrib_map = {}
-	plugin_tag_map = {}
-	
-registerStanzaPlugin(Event, EventConfiguration)
-registerStanzaPlugin(EventConfiguration, xep_0004.Form)
+    namespace = 'http://jabber.org/protocol/pubsub#event'
+    name = 'configuration'
+    plugin_attrib = name
+    interfaces = set(('node', 'config'))
+
 
 class EventPurge(ElementBase):
-	namespace = 'http://jabber.org/protocol/pubsub#event'
-	name = 'purge'
-	plugin_attrib = name
-	interfaces = set(('node',))
-	plugin_attrib_map = {}
-	plugin_tag_map = {}
+    namespace = 'http://jabber.org/protocol/pubsub#event'
+    name = 'purge'
+    plugin_attrib = name
+    interfaces = set(('node',))
 
-registerStanzaPlugin(Event, EventPurge)
 
 class EventSubscription(ElementBase):
-	namespace = 'http://jabber.org/protocol/pubsub#event'
-	name = 'subscription'
-	plugin_attrib = name
-	interfaces = set(('node','expiry', 'jid', 'subid', 'subscription'))
-	plugin_attrib_map = {}
-	plugin_tag_map = {}
-	
-	def setJid(self, value):
-		self._setAttr('jid', str(value))
-	
-	def getJid(self):
-		return JID(self._getAttr('jid'))
+    namespace = 'http://jabber.org/protocol/pubsub#event'
+    name = 'subscription'
+    plugin_attrib = name
+    interfaces = set(('node', 'expiry', 'jid', 'subid', 'subscription'))
 
-registerStanzaPlugin(Event, EventSubscription)
+    def set_jid(self, value):
+        self._set_attr('jid', str(value))
+
+    def get_jid(self):
+        return JID(self._get_attr('jid'))
+
+
+register_stanza_plugin(Message, Event)
+register_stanza_plugin(Event, EventCollection)
+register_stanza_plugin(Event, EventConfiguration)
+register_stanza_plugin(Event, EventItems)
+register_stanza_plugin(Event, EventPurge)
+register_stanza_plugin(Event, EventSubscription)
+register_stanza_plugin(EventCollection, EventAssociate)
+register_stanza_plugin(EventCollection, EventDisassociate)
+register_stanza_plugin(EventConfiguration, Form)
+register_stanza_plugin(EventItems, EventItem, iterable=True)
+register_stanza_plugin(EventItems, EventRetract, iterable=True)

sleekxmpp/plugins/xep_0060/stanza/pubsub_owner.py

@@ -1,155 +1,131 @@
-from sleekxmpp.xmlstream.stanzabase import registerStanzaPlugin, ElementBase, ET, JID
-from sleekxmpp.stanza.iq import Iq
-from sleekxmpp.stanza.message import Message
-from sleekxmpp.basexmpp import basexmpp
-from sleekxmpp.xmlstream.xmlstream import XMLStream
-import logging
-from sleekxmpp.plugins import xep_0004
+"""
+    SleekXMPP: The Sleek XMPP Library
+    Copyright (C) 2011  Nathanael C. Fritz
+    This file is part of SleekXMPP.
+
+    See the file LICENSE for copying permission.
+"""
+
+from sleekxmpp import Iq
+from sleekxmpp.xmlstream import register_stanza_plugin, ElementBase, ET, JID
+from sleekxmpp.plugins.xep_0004 import Form
 from sleekxmpp.plugins.xep_0060.stanza.base import OptionalSetting
-from sleekxmpp.plugins.xep_0060.stanza.pubsub import Affiliations, Affiliation, Configure, Subscriptions
+from sleekxmpp.plugins.xep_0060.stanza.pubsub import Affiliations, Affiliation
+from sleekxmpp.plugins.xep_0060.stanza.pubsub import Configure, Subscriptions
+
 
 class PubsubOwner(ElementBase):
     namespace = 'http://jabber.org/protocol/pubsub#owner'
     name = 'pubsub'
     plugin_attrib = 'pubsub_owner'
     interfaces = set(tuple())
-    plugin_attrib_map = {}
-    plugin_tag_map = {}
 
-registerStanzaPlugin(Iq, PubsubOwner)
 
 class DefaultConfig(ElementBase):
     namespace = 'http://jabber.org/protocol/pubsub#owner'
     name = 'default'
-    plugin_attrib = 'default'
-    interfaces = set(('node', 'type', 'config'))
-    plugin_attrib_map = {}
-    plugin_tag_map = {}
+    plugin_attrib = name
+    interfaces = set(('node', 'config'))
 
     def __init__(self, *args, **kwargs):
         ElementBase.__init__(self, *args, **kwargs)
 
-    def getType(self):
-        t = self._getAttr('type')
-        if not t: t = 'leaf'
-        return t
-
-    def getConfig(self):
+    def get_config(self):
         return self['form']
 
-    def setConfig(self, value):
-        self['form'].setStanzaValues(value.getStanzaValues())
+    def set_config(self, value):
+        self['form'].values = value.values
         return self
 
-registerStanzaPlugin(PubsubOwner, DefaultConfig)
-registerStanzaPlugin(DefaultConfig, xep_0004.Form)
 
 class OwnerAffiliations(Affiliations):
     namespace = 'http://jabber.org/protocol/pubsub#owner'
-    interfaces = set(('node'))
-    plugin_attrib_map = {}
-    plugin_tag_map = {}
+    interfaces = set(('node',))
 
     def append(self, affiliation):
         if not isinstance(affiliation, OwnerAffiliation):
             raise TypeError
         self.xml.append(affiliation.xml)
-        return self.affiliations.append(affiliation)
 
-registerStanzaPlugin(PubsubOwner, OwnerAffiliations)
 
 class OwnerAffiliation(Affiliation):
     namespace = 'http://jabber.org/protocol/pubsub#owner'
     interfaces = set(('affiliation', 'jid'))
-    plugin_attrib_map = {}
-    plugin_tag_map = {}
+
 
 class OwnerConfigure(Configure):
-    name = 'configure'
-    plugin_attrib = 'configure'
     namespace = 'http://jabber.org/protocol/pubsub#owner'
-    interfaces = set(('node', 'config'))
-    plugin_attrib_map = {}
-    plugin_tag_map = {}
-    
-    def getConfig(self):
-        return self['form']
+    name = 'configure'
+    plugin_attrib = name
+    interfaces = set(('node',))
 
-    def setConfig(self, value):
-        self['form'].setStanzaValues(value.getStanzaValues())
-        return self
-
-registerStanzaPlugin(PubsubOwner, OwnerConfigure)
 
 class OwnerDefault(OwnerConfigure):
     namespace = 'http://jabber.org/protocol/pubsub#owner'
-    interfaces = set(('node', 'config'))
-    plugin_attrib_map = {}
-    plugin_tag_map = {}
+    interfaces = set(('node',))
 
 
-registerStanzaPlugin(PubsubOwner, OwnerDefault)
-registerStanzaPlugin(OwnerDefault, xep_0004.Form)
-
 class OwnerDelete(ElementBase, OptionalSetting):
     namespace = 'http://jabber.org/protocol/pubsub#owner'
     name = 'delete'
-    plugin_attrib = 'delete'
-    plugin_attrib_map = {}
-    plugin_tag_map = {}
+    plugin_attrib = name
     interfaces = set(('node',))
 
-registerStanzaPlugin(PubsubOwner, OwnerDelete)
 
 class OwnerPurge(ElementBase, OptionalSetting):
     namespace = 'http://jabber.org/protocol/pubsub#owner'
     name = 'purge'
     plugin_attrib = name
-    plugin_attrib_map = {}
-    plugin_tag_map = {}
+    interfaces = set(('node',))
 
-registerStanzaPlugin(PubsubOwner, OwnerPurge)
 
 class OwnerRedirect(ElementBase):
     namespace = 'http://jabber.org/protocol/pubsub#owner'
     name = 'redirect'
     plugin_attrib = name
     interfaces = set(('node', 'jid'))
-    plugin_attrib_map = {}
-    plugin_tag_map = {}
 
-    def setJid(self, value):
-        self._setAttr('jid', str(value))
+    def set_jid(self, value):
+        self._set_attr('jid', str(value))
 
-    def getJid(self):
-        return JID(self._getAttr('jid'))
+    def get_jid(self):
+        return JID(self._get_attr('jid'))
 
-registerStanzaPlugin(OwnerDelete, OwnerRedirect)
 
 class OwnerSubscriptions(Subscriptions):
     namespace = 'http://jabber.org/protocol/pubsub#owner'
     interfaces = set(('node',))
-    plugin_attrib_map = {}
-    plugin_tag_map = {}
 
     def append(self, subscription):
         if not isinstance(subscription, OwnerSubscription):
             raise TypeError
         self.xml.append(subscription.xml)
-        return self.subscriptions.append(subscription)
 
-registerStanzaPlugin(PubsubOwner, OwnerSubscriptions)
 
 class OwnerSubscription(ElementBase):
     namespace = 'http://jabber.org/protocol/pubsub#owner'
     name = 'subscription'
     plugin_attrib = name
     interfaces = set(('jid', 'subscription'))
-    plugin_attrib_map = {}
-    plugin_tag_map = {}
 
-    def setJid(self, value):
-        self._setAttr('jid', str(value))
+    def set_jid(self, value):
+        self._set_attr('jid', str(value))
 
-    def getJid(self):
-        return JID(self._getAttr('from'))
+    def get_jid(self):
+        return JID(self._get_attr('jid'))
+
+
+register_stanza_plugin(Iq, PubsubOwner)
+register_stanza_plugin(PubsubOwner, DefaultConfig)
+register_stanza_plugin(PubsubOwner, OwnerAffiliations)
+register_stanza_plugin(PubsubOwner, OwnerConfigure)
+register_stanza_plugin(PubsubOwner, OwnerDefault)
+register_stanza_plugin(PubsubOwner, OwnerDelete)
+register_stanza_plugin(PubsubOwner, OwnerPurge)
+register_stanza_plugin(PubsubOwner, OwnerSubscriptions)
+register_stanza_plugin(DefaultConfig, Form)
+register_stanza_plugin(OwnerAffiliations, OwnerAffiliation, iterable=True)
+register_stanza_plugin(OwnerConfigure, Form)
+register_stanza_plugin(OwnerDefault, Form)
+register_stanza_plugin(OwnerDelete, OwnerRedirect)
+register_stanza_plugin(OwnerSubscriptions, OwnerSubscription, iterable=True)

sleekxmpp/plugins/xep_0066/oob.py

@@ -108,8 +108,7 @@ class xep_0066(base_plugin):
         iq = self.xmpp.Iq()
         iq['type'] = 'set'
         iq['to'] = to
-        if ifrom:
-            iq['from'] = ifrom
+        iq['from'] = ifrom
         iq['oob_transfer']['url'] = url
         iq['oob_transfer']['desc'] = desc
         return iq.send(**iqargs)

sleekxmpp/plugins/xep_0092/version.py

@@ -76,8 +76,7 @@ class xep_0092(base_plugin):
         """
         iq = self.xmpp.Iq()
         iq['to'] = jid
-        if ifrom:
-            iq['from'] = ifrom
+        iq['from'] = ifrom
         iq['type'] = 'get'
         iq['query'] = Version.namespace
 

sleekxmpp/plugins/xep_0199/ping.py

@@ -70,6 +70,8 @@ class xep_0199(base_plugin):
             self.xmpp.add_event_handler('session_start',
                                         self._handle_keepalive,
                                         threaded=True)
+            self.xmpp.add_event_handler('session_end',
+                                        self._handle_session_end)
 
     def post_init(self):
         """Handle cross-plugin dependencies."""
@@ -106,6 +108,9 @@ class xep_0199(base_plugin):
                            scheduled_ping,
                            repeat=True)
 
+    def _handle_session_end(self, event):
+        self.xmpp.scheduler.remove('Ping Keep Alive')
+
     def _handle_ping(self, iq):
         """
         Automatically reply to ping requests.
@@ -143,8 +148,7 @@ class xep_0199(base_plugin):
         iq = self.xmpp.Iq()
         iq['type'] = 'get'
         iq['to'] = jid
-        if ifrom:
-            iq['from'] = ifrom
+        iq['from'] = ifrom
         iq.enable('ping')
 
         start_time = time.clock()

sleekxmpp/plugins/xep_0202/time.py

@@ -85,8 +85,7 @@ class xep_0202(base_plugin):
         """
         iq = self.xmpp.Iq()
         iq['type'] = 'get'
-        iq['to'] = 'to'
-        if ifrom:
-            iq['from'] = 'ifrom'
+        iq['to'] = to
+        iq['from'] = ifrom
         iq.enable('entity_time')
         return iq.send(**iqargs)

sleekxmpp/stanza/error.py

@@ -53,6 +53,8 @@ class Error(ElementBase):
     plugin_attrib = 'error'
     interfaces = set(('code', 'condition', 'text', 'type'))
     sub_interfaces = set(('text',))
+    plugin_attrib_map = {}
+    plugin_tag_map = {}
     conditions = set(('bad-request', 'conflict', 'feature-not-implemented',
                       'forbidden', 'gone', 'internal-server-error',
                       'item-not-found', 'jid-malformed', 'not-acceptable',

sleekxmpp/test/livesocket.py

@@ -138,7 +138,7 @@ class TestLiveSocket(object):
         """
         with self.send_queue_lock:
             self.send_queue.put(data)
-        self.socket.send(data)
+        return self.socket.send(data)
 
     # ------------------------------------------------------------------
     # File Socket

sleekxmpp/test/mocksocket.py

@@ -121,6 +121,7 @@ class TestSocket(object):
         if self.disconnected:
             raise socket.error
         self.send_queue.put(data)
+        return len(data)
 
     # ------------------------------------------------------------------
     # File Socket

sleekxmpp/test/sleektest.py

@@ -58,9 +58,6 @@ class SleekTest(unittest.TestCase):
         unittest.TestCase.__init__(self, *args, **kwargs)
         self.xmpp = None
 
-    def runTest(self):
-        pass
-
     def parse_xml(self, xml_string):
         try:
             xml = ET.fromstring(xml_string)
@@ -336,7 +333,6 @@ class SleekTest(unittest.TestCase):
 
             # Simulate connecting for mock sockets.
             self.xmpp.auto_reconnect = False
-            self.xmpp.is_client = True
             self.xmpp.state._set_state('connected')
 
             # Must have the stream header ready for xmpp.process() to work.

sleekxmpp/version.py

@@ -0,0 +1,13 @@
+"""
+    SleekXMPP: The Sleek XMPP Library
+    Copyright (C) 2010  Nathanael C. Fritz
+    This file is part of SleekXMPP.
+
+    See the file LICENSE for copying permission.
+"""
+
+# We don't want to have to import the entire library
+# just to get the version info for setup.py
+
+__version__ = '1.0rc2'
+__version_info__ = (1, 0, 0, 'rc2', 0)

sleekxmpp/xmlstream/scheduler.py

@@ -73,7 +73,8 @@ class Task(object):
         otherwise, execute the callback immediately.
         """
         if self.qpointer is not None:
-            self.qpointer.put(('schedule', self.callback, self.args))
+            self.qpointer.put(('schedule', self.callback,
+                               self.args, self.name))
         else:
             self.callback(*self.args, **self.kwargs)
         self.reset()
@@ -95,31 +96,32 @@ class Scheduler(object):
     http://docs.python.org/library/sched.html#module-sched
 
     Attributes:
-        addq        -- A queue storing added tasks.
-        schedule    -- A list of tasks in order of execution times.
-        thread      -- If threaded, the thread processing the schedule.
-        run         -- Indicates if the scheduler is running.
-        parentqueue -- A parent event queue in control of this scheduler.
-
+        addq     -- A queue storing added tasks.
+        schedule -- A list of tasks in order of execution times.
+        thread   -- If threaded, the thread processing the schedule.
+        run      -- Indicates if the scheduler is running.
+        stop     -- Threading event indicating if the main process
+                    has been stopped.
     Methods:
         add     -- Add a new task to the schedule.
         process -- Process and schedule tasks.
         quit    -- Stop the scheduler.
     """
 
-    def __init__(self, parentqueue=None, parentstop=None):
+    def __init__(self, parentstop=None):
         """
         Create a new scheduler.
 
         Arguments:
-            parentqueue -- A separate event queue controlling this scheduler.
+            parentstop -- A threading event indicating if the main process has
+                          been stopped.
         """
         self.addq = queue.Queue()
         self.schedule = []
         self.thread = None
         self.run = False
-        self.parentqueue = parentqueue
-        self.parentstop = parentstop
+        self.stop = parentstop
+        self.schedule_lock = threading.RLock()
 
     def process(self, threaded=True):
         """
@@ -141,8 +143,7 @@ class Scheduler(object):
         """Process scheduled tasks."""
         self.run = True
         try:
-            while self.run and (self.parentstop is None or \
-                                not self.parentstop.isSet()):
+            while self.run and not self.stop.isSet():
                     wait = 1
                     updated = False
                     if self.schedule:
@@ -156,6 +157,7 @@ class Scheduler(object):
                             newtask = self.addq.get(True, wait)
                     except queue.Empty:
                         cleanup = []
+                        self.schedule_lock.acquire()
                         for task in self.schedule:
                             if time.time() >= task.next:
                                 updated = True
@@ -167,23 +169,18 @@ class Scheduler(object):
                             x = self.schedule.pop(self.schedule.index(task))
                     else:
                         updated = True
+                        self.schedule_lock.acquire()
                         self.schedule.append(newtask)
                     finally:
                         if updated:
                             self.schedule = sorted(self.schedule,
                                                    key=lambda task: task.next)
+                        self.schedule_lock.release()
         except KeyboardInterrupt:
             self.run = False
-            if self.parentstop is not None:
-                log.debug("stopping parent")
-                self.parentstop.set()
         except SystemExit:
             self.run = False
-            if self.parentstop is not None:
-                self.parentstop.set()
         log.debug("Quitting Scheduler thread")
-        if self.parentqueue is not None:
-            self.parentqueue.put(('quit', None, None))
 
     def add(self, name, seconds, callback, args=None,
             kwargs=None, repeat=False, qpointer=None):
@@ -201,8 +198,39 @@ class Scheduler(object):
             qpointer -- A pointer to an event queue for queuing callback
                         execution instead of executing immediately.
         """
-        self.addq.put(Task(name, seconds, callback, args,
-                           kwargs, repeat, qpointer))
+        try:
+            self.schedule_lock.acquire()
+            for task in self.schedule:
+                if task.name == name:
+                    raise ValueError("Key %s already exists" % name)
+
+            self.addq.put(Task(name, seconds, callback, args,
+                               kwargs, repeat, qpointer))
+        except:
+            raise
+        finally:
+            self.schedule_lock.release()
+
+    def remove(self, name):
+        """
+        Remove a scheduled task ahead of schedule, and without
+        executing it.
+
+        Arguments:
+            name -- The name of the task to remove.
+        """
+        try:
+            self.schedule_lock.acquire()
+            the_task = None
+            for task in self.schedule:
+                if task.name == name:
+                    the_task = task
+            if the_task is not None:
+                self.schedule.remove(the_task)
+        except:
+            raise
+        finally:
+            self.schedule_lock.release()
 
     def quit(self):
         """Shutdown the scheduler."""

sleekxmpp/xmlstream/stanzabase.py

@@ -39,15 +39,23 @@ def register_stanza_plugin(stanza, plugin, iterable=False, overrides=False):
                      the parent stanza.
     """
     tag = "{%s}%s" % (plugin.namespace, plugin.name)
+
+    # Prevent weird memory reference gotchas by ensuring
+    # that the parent stanza class has its own set of
+    # plugin info maps and is not using the mappings from
+    # an ancestor class (like ElementBase).
+    plugin_info = ('plugin_attrib_map', 'plugin_tag_map',
+                   'plugin_iterables', 'plugin_overrides')
+    for attr in plugin_info:
+        info = getattr(stanza, attr)
+        setattr(stanza, attr, info.copy())
+
     stanza.plugin_attrib_map[plugin.plugin_attrib] = plugin
     stanza.plugin_tag_map[tag] = plugin
+
     if iterable:
-        # Prevent weird memory reference gotchas.
-        stanza.plugin_iterables = stanza.plugin_iterables.copy()
         stanza.plugin_iterables.add(plugin)
     if overrides:
-        # Prevent weird memory reference gotchas.
-        stanza.plugin_overrides = stanza.plugin_overrides.copy()
         for interface in plugin.overrides:
             stanza.plugin_overrides[interface] = plugin.plugin_attrib
 

sleekxmpp/xmlstream/xmlstream.py

@@ -47,6 +47,10 @@ else:
 # The time in seconds to wait before timing out waiting for response stanzas.
 RESPONSE_TIMEOUT = 30
 
+# The time in seconds to wait for events from the event queue, and also the
+# time between checks for the process stop signal.
+WAIT_TIMEOUT = 1
+
 # The number of threads to use to handle XML stream events. This is not the
 # same as the number of custom event handling threads. HANDLER_THREADS must
 # be at least 1.
@@ -178,6 +182,7 @@ class XMLStream(object):
         self.ssl_version = ssl.PROTOCOL_TLSv1
         self.ca_certs = None
 
+        self.wait_timeout = WAIT_TIMEOUT
         self.response_timeout = RESPONSE_TIMEOUT
         self.reconnect_delay = None
         self.reconnect_max_delay = RECONNECT_MAX_DELAY
@@ -207,15 +212,19 @@ class XMLStream(object):
         self.stream_header = "<stream>"
         self.stream_footer = "</stream>"
 
+        self.whitespace_keepalive = True
+        self.whitespace_keepalive_interval = 300
+
         self.stop = threading.Event()
         self.stream_end_event = threading.Event()
         self.stream_end_event.set()
         self.session_started_event = threading.Event()
+        self.session_timeout = 45
 
         self.event_queue = queue.Queue()
         self.send_queue = queue.Queue()
         self.__failed_send_stanza = None
-        self.scheduler = Scheduler(self.event_queue, self.stop)
+        self.scheduler = Scheduler(self.stop)
 
         self.namespace_map = {StanzaBase.xml_ns: 'xml'}
 
@@ -229,9 +238,12 @@ class XMLStream(object):
         self._id_lock = threading.Lock()
 
         self.auto_reconnect = True
-        self.is_client = False
         self.dns_answers = []
 
+        self.add_event_handler('connected', self._handle_connected)
+        self.add_event_handler('session_start', self._start_keepalive)
+        self.add_event_handler('session_end', self._end_keepalive)
+
     def use_signals(self, signals=None):
         """
         Register signal handlers for SIGHUP and SIGTERM, if possible,
@@ -320,7 +332,6 @@ class XMLStream(object):
         except Socket.error:
             self.default_domain = self.address[0]
 
-        self.is_client = True
         # Respect previous SSL and TLS usage directives.
         if use_ssl is not None:
             self.use_ssl = use_ssl
@@ -337,12 +348,13 @@ class XMLStream(object):
         return connected
 
     def _connect(self):
+        self.scheduler.remove('Session timeout check')
         self.stop.clear()
         if self.default_domain:
             self.address = self.pick_dns_answer(self.default_domain,
                                                 self.address[1])
         self.socket = self.socket_class(Socket.AF_INET, Socket.SOCK_STREAM)
-        self.socket.settimeout(None)
+        self.configure_socket()
 
         if self.reconnect_delay is None:
             delay = 1.0
@@ -446,6 +458,23 @@ class XMLStream(object):
                                        serr.errno, serr.strerror))
             return False
 
+    def _handle_connected(self, event=None):
+        """
+        Add check to ensure that a session is established within
+        a reasonable amount of time.
+        """
+
+        def _handle_session_timeout():
+            if not self.session_started_event.isSet():
+                log.debug("Session start has taken more " + \
+                          "than %d seconds" % self.session_timeout)
+                self.disconnect(reconnect=self.auto_reconnect)
+
+        self.schedule("Session timeout check",
+                self.session_timeout,
+                _handle_session_timeout)
+
+
     def disconnect(self, reconnect=False, wait=False):
         """
         Terminate processing and close the XML streams.
@@ -485,9 +514,9 @@ class XMLStream(object):
         if not self.auto_reconnect:
             self.stop.set()
         try:
+            self.socket.shutdown(Socket.SHUT_RDWR)
             self.socket.close()
             self.filesocket.close()
-            self.socket.shutdown(Socket.SHUT_RDWR)
         except Socket.error as serr:
             self.event('socket_error', serr)
         finally:
@@ -532,6 +561,31 @@ class XMLStream(object):
             if not ignore:
                 self.state._set_state('connected')
 
+    def configure_socket(self):
+        """
+        Set timeout and other options for self.socket.
+
+        Meant to be overridden.
+        """
+        self.socket.settimeout(None)
+
+    def configure_dns(self, resolver, domain=None, port=None):
+        """
+        Configure and set options for a dns.resolver.Resolver
+        instance, and other DNS related tasks. For example, you
+        can also check Socket.getaddrinfo to see if you need to
+        call out to libresolv.so.2 to run res_init().
+
+        Meant to be overridden.
+
+        Arguments:
+            resolver -- A dns.resolver.Resolver instance, or None
+                        if dnspython is not installed.
+            domain   -- The initial domain under consideration.
+            port     -- The initial port under consideration.
+        """
+        pass
+
     def start_tls(self):
         """
         Perform handshakes for TLS.
@@ -566,6 +620,30 @@ class XMLStream(object):
             log.warning("Tried to enable TLS, but ssl module not found.")
             return False
 
+    def _start_keepalive(self, event):
+        """
+        Begin sending whitespace periodically to keep the connection alive.
+
+        May be disabled by setting:
+            self.whitespace_keepalive = False
+
+        The keepalive interval can be set using:
+            self.whitespace_keepalive_interval = 300
+        """
+
+        def send_keepalive():
+            if self.send_queue.empty():
+                self.send_raw(' ')
+
+        self.schedule('Whitespace Keepalive',
+                      self.whitespace_keepalive_interval,
+                      send_keepalive,
+                      repeat=True)
+
+    def _end_keepalive(self, event):
+        """Stop sending whitespace keepalives"""
+        self.scheduler.remove('Whitespace Keepalive')
+
     def start_stream_handler(self, xml):
         """
         Perform any initialization actions, such as handshakes, once the
@@ -669,18 +747,24 @@ class XMLStream(object):
         if port is None:
             port = self.default_port
         if DNSPYTHON:
+            resolver = dns.resolver.get_default_resolver()
+            self.configure_dns(resolver, domain=domain, port=port)
+
             try:
-                answers = dns.resolver.query(domain, dns.rdatatype.A)
+                answers = resolver.query(domain, dns.rdatatype.A)
             except (dns.resolver.NXDOMAIN, dns.resolver.NoAnswer):
                 log.warning("No A records for %s" % domain)
+                return [((domain, port), 0, 0)]
             except dns.exception.Timeout:
                 log.warning("DNS resolution timed out " + \
                             "for A record of %s" % domain)
-            answers = [((answer.address, port), 0, 0) for answer in answers]
-            return answers
+                return [((domain, port), 0, 0)]
+            else:
+                return [((ans.address, port), 0, 0) for ans in answers]
         else:
             log.warning("dnspython is not installed -- " + \
                         "relying on OS A record resolution")
+            self.configure_dns(None, domain=domain, port=port)
             return [((domain, port), 0, 0)]
 
     def pick_dns_answer(self, domain, port=None):
@@ -905,7 +989,15 @@ class XMLStream(object):
         if now:
             log.debug("SEND (IMMED): %s" % data)
             try:
-                self.socket.send(data.encode('utf-8'))
+                data = data.encode('utf-8')
+                total = len(data)
+                sent = 0
+                count = 0
+                while sent < total and not self.stop.is_set():
+                    sent += self.socket.send(data[sent:])
+                    count += 1
+                if count > 1:
+                    log.debug('SENT: %d chunks' % count)
             except Socket.error as serr:
                 self.event('socket_error', serr)
                 log.warning("Failed to send %s" % data)
@@ -973,44 +1065,48 @@ class XMLStream(object):
         Processing will continue after any recoverable errors
         if reconnections are allowed.
         """
-        firstrun = True
 
         # The body of this loop will only execute once per connection.
         # Additional passes will be made only if an error occurs and
         # reconnecting is permitted.
-        while firstrun or (self.auto_reconnect and not self.stop.isSet()):
-            firstrun = False
+        while True:
             try:
-                if self.is_client:
-                    self.send_raw(self.stream_header, now=True)
                 # The call to self.__read_xml will block and prevent
                 # the body of the loop from running until a disconnect
                 # occurs. After any reconnection, the stream header will
                 # be resent and processing will resume.
-                while not self.stop.isSet() and self.__read_xml():
+                while not self.stop.is_set():
+                    # Only process the stream while connected to the server
+                    if not self.state.ensure('connected', wait=0.1,
+                                             block_on_transition=True):
+                        continue
                     # Ensure the stream header is sent for any
                     # new connections.
-                    if self.is_client:
+                    if not self.session_started_event.is_set():
                         self.send_raw(self.stream_header, now=True)
+                    self.__read_xml()
             except KeyboardInterrupt:
                 log.debug("Keyboard Escape Detected in _process")
                 self.stop.set()
             except SystemExit:
                 log.debug("SystemExit in _process")
                 self.stop.set()
+                self.scheduler.quit()
             except Socket.error as serr:
                 self.event('socket_error', serr)
                 log.exception('Socket Error')
             except:
-                if not self.stop.isSet():
+                if not self.stop.is_set():
                     log.exception('Connection error.')
-            if not self.stop.isSet() and self.auto_reconnect:
-                self.reconnect()
+
+            if not self.stop.is_set():
+                if self.auto_reconnect:
+                    self.reconnect()
+                else:
+                    continue
             else:
-                self.event('killed', direct=True)
                 self.disconnect()
-                self.event_queue.put(('quit', None, None))
-        self.scheduler.run = False
+                break
 
     def __read_xml(self):
         """
@@ -1150,7 +1246,8 @@ class XMLStream(object):
         try:
             while not self.stop.isSet():
                 try:
-                    event = self.event_queue.get(True, timeout=5)
+                    wait = self.wait_timeout
+                    event = self.event_queue.get(True, timeout=wait)
                 except queue.Empty:
                     event = None
                 if event is None:
@@ -1168,8 +1265,9 @@ class XMLStream(object):
                         log.exception(error_msg % handler.name)
                         orig.exception(e)
                 elif etype == 'schedule':
+                    name = args[1]
                     try:
-                        log.debug('Scheduled event: %s' % args)
+                        log.debug('Scheduled event: %s: %s' % (name, args[0]))
                         handler(*args[0])
                     except Exception as e:
                         log.exception('Error processing scheduled task')
@@ -1211,7 +1309,7 @@ class XMLStream(object):
         Extract stanzas from the send queue and send them on the stream.
         """
         try:
-            while not self.stop.isSet():
+            while not self.stop.is_set():
                 self.session_started_event.wait()
                 if self.__failed_send_stanza is not None:
                     data = self.__failed_send_stanza
@@ -1223,22 +1321,26 @@ class XMLStream(object):
                         continue
                 log.debug("SEND: %s" % data)
                 try:
-                    self.socket.send(data.encode('utf-8'))
+                    enc_data = data.encode('utf-8')
+                    total = len(enc_data)
+                    sent = 0
+                    count = 0
+                    while sent < total and not self.stop.is_set():
+                        sent += self.socket.send(enc_data[sent:])
+                        count += 1
+                    if count > 1:
+                        log.debug('SENT: %d chunks' % count)
                     self.send_queue.task_done()
                 except Socket.error as serr:
                     self.event('socket_error', serr)
                     log.warning("Failed to send %s" % data)
                     self.__failed_send_stanza = data
                     self.disconnect(self.auto_reconnect)
-        except KeyboardInterrupt:
-            log.debug("Keyboard Escape Detected in _send_thread")
-            self.event('killed', direct=True)
-            self.disconnect()
-            return
-        except SystemExit:
-            self.disconnect()
-            self.event_queue.put(('quit', None, None))
-            return
+        except Exception as ex:
+            log.exception('Unexpected error in send thread: %s' % ex)
+            self.exception(ex)
+            if not self.stop.is_set():
+                self.disconnect(self.auto_reconnect)
 
     def exception(self, exception):
         """

testall.py

@@ -1,70 +1,63 @@
 #!/usr/bin/env python
-import unittest
-import logging
-import sys
+
 import os
+import sys
+import logging
+import unittest
+import distutils.core
 
-class testoverall(unittest.TestCase):
+from glob import glob
+from os.path import splitext, basename, join as pjoin
 
-	def testModules(self):
-		"""Testing all modules by compiling them"""
-		import compileall
-		import re
-		if sys.version_info < (3,0):
-			self.failUnless(compileall.compile_dir('.' + os.sep + 'sleekxmpp', rx=re.compile('/[.]svn'), quiet=True))
-		else:
-			self.failUnless(compileall.compile_dir('.' + os.sep + 'sleekxmpp', rx=re.compile('/[.]svn|.*26.*'), quiet=True))
 
-	def	testTabNanny(self):
-		"""Invoking the tabnanny"""
-		import tabnanny
-		self.failIf(tabnanny.check("." + os.sep + 'sleekxmpp'))
-		#raise "Help!"
+def run_tests():
+    """
+    Find and run all tests in the tests/ directory.
+
+    Excludes live tests (tests/live_*).
+    """
+    testfiles = ['tests.test_overall']
+    exclude = ['__init__.py', 'test_overall.py']
+    for t in glob(pjoin('tests', '*.py')):
+        if True not in [t.endswith(ex) for ex in exclude]:
+            if basename(t).startswith('test_'):
+                testfiles.append('tests.%s' % splitext(basename(t))[0])
+
+    suites = []
+    for file in testfiles:
+        __import__(file)
+        suites.append(sys.modules[file].suite)
+
+    tests = unittest.TestSuite(suites)
+    runner = unittest.TextTestRunner(verbosity=2)
+
+    # Disable logging output
+    logging.basicConfig(level=100)
+    logging.disable(100)
+
+    result = runner.run(tests)
+    return result
+
+
+# Add a 'test' command for setup.py
+
+class TestCommand(distutils.core.Command):
+
+    user_options = [ ]
+
+    def initialize_options(self):
+        self._dir = os.getcwd()
+
+    def finalize_options(self):
+        pass
+
+    def run(self):
+        run_tests()
 
-	def disabled_testMethodLength(self):
-		"""Testing for excessive method lengths"""
-		import re
-		dirs = os.walk(sys.path[0] + os.sep + 'sleekxmpp')
-		offenders = []
-		for d in dirs:
-			if not '.svn' in d[0]:
-				for filename in d[2]:
-					if filename.endswith('.py') and d[0].find("template%stemplates" % os.sep) == -1:
-						with open("%s%s%s" % (d[0],os.sep,filename), "r") as fp:
-							cur = None
-							methodline = lineno = methodlen = methodindent = 0
-							for line in fp:
-								indentlevel = re.compile("^[\t ]*").search(line).end()
-								line = line.expandtabs()
-								lineno += 1
-								if line.strip().startswith("def ") or line.strip().startswith("except") or (line.strip() and methodindent > indentlevel) or (line.strip() and methodindent == indentlevel): #new method found or old one ended
-									if cur: #existing method needs final evaluation
-										if methodlen > 50 and not cur.strip().startswith("def setupUi"):
-											offenders.append("Method '%s' on line %s of %s/%s is longer than 50 lines (%s)" % (cur.strip(),methodline,d[0][len(rootp):],filename,methodlen))
-										methodlen = 0
-									cur = line
-									methodindent = indentlevel
-									methodline = lineno
-								if line and cur and not line.strip().startswith("#") and not (cur.strip().startswith("try:") and methodindent == 0): #if we weren't all whitespace and weren't a comment
-									methodlen += 1
-		self.failIf(offenders,"\n".join(offenders))
-			
 
 if __name__ == '__main__':
-	logging.basicConfig(level=100)
-	logging.disable(100)
-	#this doesn't need to be very clean
-	alltests = [unittest.TestLoader().loadTestsFromTestCase(testoverall)]
-	rootp = sys.path[0] + os.sep + 'tests'
-	dirs = os.walk(rootp)
-	for d in dirs:
-		if not '.svn' in d[0]:
-			for filename in d[2]:
-				if filename.startswith('test_') and filename.endswith('.py'):
-					modname = ('tests' + "." + filename)[:-3].replace(os.sep,'.')
-					__import__(modname)
-					#sys.modules[modname].config = moduleconfig
-					alltests.append(sys.modules[modname].suite)
-	alltests_suite = unittest.TestSuite(alltests)
-	result = unittest.TextTestRunner(verbosity=2).run(alltests_suite)
-	print("""<tests xmlns='http://andyet.net/protocol/tests' ran='%s' errors='%s' fails='%s' success='%s' />""" % (result.testsRun, len(result.errors), len(result.failures), result.wasSuccessful()))
+    result = run_tests()
+    print("<tests %s ran='%s' errors='%s' fails='%s' success='%s' />" % (
+        "xmlns='http//andyet.net/protocol/tests'",
+        result.testsRun, len(result.errors),
+        len(result.failures), result.wasSuccessful()))

tests/test_overall.py

@@ -0,0 +1,29 @@
+import os
+import re
+import sys
+import unittest
+import tabnanny
+import compileall
+
+class TestOverall(unittest.TestCase):
+
+    """
+    Test overall package health by compiling and checking
+    code style.
+    """
+
+    def testModules(self):
+        """Testing all modules by compiling them"""
+        src = '.%ssleekxmpp' % os.sep
+        if sys.version_info < (3, 0):
+            rx = re.compile('/[.]svn')
+        else:
+            rx = re.compile('/[.]svn|.*26.*')
+        self.failUnless(compileall.compile_dir(src, rx=rx, quiet=True))
+
+    def testTabNanny(self):
+        """Testing that indentation is consistent"""
+        self.failIf(tabnanny.check('..%ssleekxmpp' % os.sep))
+
+
+suite = unittest.TestLoader().loadTestsFromTestCase(TestOverall)

tests/test_stanza_xep_0060.py

@@ -148,14 +148,13 @@ class TestPubsubStanzas(SleekTest):
         iq = self.Iq()
         iq['pubsub_owner']['default']
         iq['pubsub_owner']['default']['node'] = 'mynode'
-        iq['pubsub_owner']['default']['type'] = 'leaf'
         iq['pubsub_owner']['default']['form'].addField('pubsub#title',
                                                        ftype='text-single',
                                                        value='This thing is awesome')
         self.check(iq, """
 	      <iq id="0">
             <pubsub xmlns="http://jabber.org/protocol/pubsub#owner">
-              <default node="mynode" type="leaf">
+              <default node="mynode">
                 <x xmlns="jabber:x:data" type="form">
                   <field var="pubsub#title" type="text-single">
                     <value>This thing is awesome</value>
@@ -213,6 +212,9 @@ class TestPubsubStanzas(SleekTest):
         item2['payload'] = payload2
         iq['pubsub']['publish'].append(item)
         iq['pubsub']['publish'].append(item2)
+        form = xep_0004.Form()
+        form.addField('pubsub#description', ftype='text-single', value='this thing is awesome')
+        iq['pubsub']['publish_options'] = form
 
         self.check(iq, """
           <iq id="0">
@@ -231,6 +233,13 @@ class TestPubsubStanzas(SleekTest):
                   </thinger2>
                 </item>
               </publish>
+              <publish-options>
+                <x xmlns="jabber:x:data" type="submit">
+                  <field var="pubsub#description">
+                    <value>this thing is awesome</value>
+                  </field>
+                </x>
+              </publish-options>
             </pubsub>
           </iq>""")
 
@@ -508,4 +517,59 @@ class TestPubsubStanzas(SleekTest):
             </event>
           </message>""")
 
+    def testPubsubError(self):
+        """Test getting a pubsub specific condition from an error stanza"""
+        iq = self.Iq()
+        iq['error']['type'] = 'cancel'
+        iq['error']['code'] = '501'
+        iq['error']['condition'] = 'feature-not-implemented'
+        iq['error']['pubsub']['condition'] = 'subid-required'
+        self.check(iq, """
+          <iq type="error">
+            <error type="cancel" code="501">
+              <feature-not-implemented xmlns="urn:ietf:params:xml:ns:xmpp-stanzas" />
+              <subid-required xmlns="http://jabber.org/protocol/pubsub#errors" />
+            </error>
+          </iq>
+        """, use_values=False)
+
+        del iq['error']['pubsub']['condition']
+        self.check(iq, """
+          <iq type="error">
+            <error type="cancel" code="501">
+              <feature-not-implemented xmlns="urn:ietf:params:xml:ns:xmpp-stanzas" />
+            </error>
+          </iq>
+        """, use_values=False)
+
+    def testPubsubUnsupportedError(self):
+        """Test getting the feature from an unsupported error"""
+        iq = self.Iq()
+        iq['error']['type'] = 'cancel'
+        iq['error']['code'] = '501'
+        iq['error']['condition'] = 'feature-not-implemented'
+        iq['error']['pubsub']['condition'] = 'unsupported'
+        iq['error']['pubsub']['unsupported'] = 'instant-node'
+        self.check(iq, """
+          <iq type="error">
+            <error type="cancel" code="501">
+              <feature-not-implemented xmlns="urn:ietf:params:xml:ns:xmpp-stanzas" />
+              <unsupported xmlns="http://jabber.org/protocol/pubsub#errors" feature="instant-node" />
+            </error>
+          </iq>
+        """, use_values=False)
+
+        self.assertEqual(iq['error']['pubsub']['condition'], 'unsupported')
+        self.assertEqual(iq['error']['pubsub']['unsupported'], 'instant-node')
+
+        del iq['error']['pubsub']['unsupported']
+        self.check(iq, """
+          <iq type="error">
+            <error type="cancel" code="501">
+              <feature-not-implemented xmlns="urn:ietf:params:xml:ns:xmpp-stanzas" />
+            </error>
+          </iq>
+        """, use_values=False)
+
+
 suite = unittest.TestLoader().loadTestsFromTestCase(TestPubsubStanzas)

tests/test_stream_xep_0060.py

@@ -0,0 +1,794 @@
+import sys
+import time
+import threading
+
+from sleekxmpp.test import *
+from sleekxmpp.stanza.atom import AtomEntry
+from sleekxmpp.xmlstream import register_stanza_plugin
+
+
+class TestStreamPubsub(SleekTest):
+
+    """
+    Test using the XEP-0030 plugin.
+    """
+
+    def setUp(self):
+        self.stream_start()
+
+    def tearDown(self):
+        self.stream_close()
+
+    def testCreateInstantNode(self):
+        """Test creating an instant node"""
+        t = threading.Thread(name='create_node',
+                             target=self.xmpp['xep_0060'].create_node,
+                             args=('pubsub.example.com', None))
+        t.start()
+
+        self.send("""
+          <iq type="set" id="1" to="pubsub.example.com">
+            <pubsub xmlns="http://jabber.org/protocol/pubsub">
+              <create />
+            </pubsub>
+          </iq>
+        """)
+
+        self.recv("""
+          <iq type="result" id="1"
+              to="tester@localhost" from="pubsub.example.com">
+            <pubsub xmlns="http://jabber.org/protocol/pubsub">
+              <create node="25e3d37dabbab9541f7523321421edc5bfeb2dae" />
+            </pubsub>
+          </iq>
+        """)
+
+        t.join()
+
+    def testCreateNodeNoConfig(self):
+        """Test creating a node without a config"""
+        self.xmpp['xep_0060'].create_node(
+            'pubsub.example.com',
+            'princely_musings',
+            block=False)
+        self.send("""
+          <iq type="set" id="1" to="pubsub.example.com">
+            <pubsub xmlns="http://jabber.org/protocol/pubsub">
+              <create node="princely_musings" />
+            </pubsub>
+          </iq>
+        """)
+
+    def testCreateNodeConfig(self):
+        """Test creating a node with a config"""
+        form = self.xmpp['xep_0004'].stanza.Form()
+        form['type'] = 'submit'
+        form.add_field(var='pubsub#access_model', value='whitelist')
+
+        self.xmpp['xep_0060'].create_node(
+                'pubsub.example.com',
+                'princely_musings',
+                config=form, block=False)
+
+        self.send("""
+          <iq type="set" id="1" to="pubsub.example.com">
+            <pubsub xmlns="http://jabber.org/protocol/pubsub">
+              <create node="princely_musings" />
+              <configure>
+                <x xmlns="jabber:x:data" type="submit">
+                  <field var="pubsub#access_model">
+                    <value>whitelist</value>
+                  </field>
+                  <field var="FORM_TYPE">
+                    <value>http://jabber.org/protocol/pubsub#node_config</value>
+                  </field>
+                </x>
+              </configure>
+            </pubsub>
+          </iq>
+        """)
+
+    def testDeleteNode(self):
+        """Test deleting a node"""
+        self.xmpp['xep_0060'].delete_node(
+            'pubsub.example.com',
+            'some_node',
+            block=False)
+        self.send("""
+          <iq type="set" to="pubsub.example.com" id="1">
+            <pubsub xmlns="http://jabber.org/protocol/pubsub#owner">
+              <delete node="some_node" />
+            </pubsub>
+          </iq>
+        """)
+
+    def testSubscribeCase1(self):
+        """
+        Test subscribing to a node: Case 1:
+        No subscribee, default 'from' JID, bare JID
+        """
+        self.xmpp['xep_0060'].subscribe(
+            'pubsub.example.com',
+            'somenode',
+            block=False)
+        self.send("""
+          <iq type="set" id="1" to="pubsub.example.com">
+            <pubsub xmlns="http://jabber.org/protocol/pubsub">
+              <subscribe node="somenode" jid="tester@localhost" />
+            </pubsub>
+          </iq>
+        """)
+
+    def testSubscribeCase2(self):
+        """
+        Test subscribing to a node: Case 2:
+        No subscribee, given 'from' JID, bare JID
+        """
+        self.xmpp['xep_0060'].subscribe(
+            'pubsub.example.com',
+            'somenode',
+            ifrom='foo@comp.example.com/bar',
+            block=False)
+        self.send("""
+          <iq type="set" id="1"
+              to="pubsub.example.com" from="foo@comp.example.com/bar">
+            <pubsub xmlns="http://jabber.org/protocol/pubsub">
+              <subscribe node="somenode" jid="foo@comp.example.com" />
+            </pubsub>
+          </iq>
+        """)
+
+    def testSubscribeCase3(self):
+        """
+        Test subscribing to a node: Case 3:
+        No subscribee, given 'from' JID, full JID
+        """
+        self.xmpp['xep_0060'].subscribe(
+            'pubsub.example.com',
+            'somenode',
+            ifrom='foo@comp.example.com/bar',
+            bare=False,
+            block=False)
+        self.send("""
+          <iq type="set" id="1"
+              to="pubsub.example.com" from="foo@comp.example.com/bar">
+            <pubsub xmlns="http://jabber.org/protocol/pubsub">
+              <subscribe node="somenode" jid="foo@comp.example.com/bar" />
+            </pubsub>
+          </iq>
+        """)
+
+    def testSubscribeCase4(self):
+        """
+        Test subscribing to a node: Case 4:
+        No subscribee, no 'from' JID, full JID
+        """
+        self.stream_close()
+        self.stream_start(jid='tester@localhost/full')
+
+        self.xmpp['xep_0060'].subscribe(
+            'pubsub.example.com',
+            'somenode',
+            bare=False,
+            block=False)
+        self.send("""
+          <iq type="set" id="1"
+              to="pubsub.example.com">
+            <pubsub xmlns="http://jabber.org/protocol/pubsub">
+              <subscribe node="somenode" jid="tester@localhost/full" />
+            </pubsub>
+          </iq>
+        """)
+
+    def testSubscribeCase5(self):
+        """
+        Test subscribing to a node: Case 5:
+        Subscribee given
+        """
+        self.xmpp['xep_0060'].subscribe(
+            'pubsub.example.com',
+            'somenode',
+            subscribee='user@example.com/foo',
+            ifrom='foo@comp.example.com/bar',
+            block=False)
+        self.send("""
+          <iq type="set" id="1"
+              to="pubsub.example.com" from="foo@comp.example.com/bar">
+            <pubsub xmlns="http://jabber.org/protocol/pubsub">
+              <subscribe node="somenode" jid="user@example.com/foo" />
+            </pubsub>
+          </iq>
+        """)
+
+    def testSubscribeWithOptions(self):
+        """Test subscribing to a node, with options."""
+        opts = self.xmpp['xep_0004'].make_form()
+        opts.add_field(
+                var='FORM_TYPE',
+                value='http://jabber.org/protocol/pubsub#subscribe_options',
+                ftype='hidden')
+        opts.add_field(
+                var='pubsub#digest',
+                value=False,
+                ftype='boolean')
+        opts['type'] = 'submit'
+
+        self.xmpp['xep_0060'].subscribe(
+            'pubsub.example.com',
+            'somenode',
+            options=opts,
+            block=False)
+        self.send("""
+          <iq type="set" id="1" to="pubsub.example.com">
+            <pubsub xmlns="http://jabber.org/protocol/pubsub">
+              <subscribe node="somenode" jid="tester@localhost" />
+              <options>
+                <x xmlns="jabber:x:data" type="submit">
+                  <field var="FORM_TYPE">
+                    <value>http://jabber.org/protocol/pubsub#subscribe_options</value>
+                  </field>
+                  <field var="pubsub#digest">
+                    <value>0</value>
+                  </field>
+                </x>
+              </options>
+            </pubsub>
+          </iq>
+        """)
+
+    def testUnsubscribeCase1(self):
+        """
+        Test unsubscribing from a node: Case 1:
+        No subscribee, default 'from' JID, bare JID
+        """
+        self.xmpp['xep_0060'].unsubscribe(
+            'pubsub.example.com',
+            'somenode',
+            block=False)
+        self.send("""
+          <iq type="set" id="1" to="pubsub.example.com">
+            <pubsub xmlns="http://jabber.org/protocol/pubsub">
+              <unsubscribe node="somenode" jid="tester@localhost" />
+            </pubsub>
+          </iq>
+        """)
+
+    def testUnsubscribeCase2(self):
+        """
+        Test unsubscribing from a node: Case 2:
+        No subscribee, given 'from' JID, bare JID
+        """
+        self.xmpp['xep_0060'].unsubscribe(
+            'pubsub.example.com',
+            'somenode',
+            ifrom='foo@comp.example.com/bar',
+            block=False)
+        self.send("""
+          <iq type="set" id="1"
+              to="pubsub.example.com" from="foo@comp.example.com/bar">
+            <pubsub xmlns="http://jabber.org/protocol/pubsub">
+              <unsubscribe node="somenode" jid="foo@comp.example.com" />
+            </pubsub>
+          </iq>
+        """)
+
+    def testUnsubscribeCase3(self):
+        """
+        Test unsubscribing from a node: Case 3:
+        No subscribee, given 'from' JID, full JID
+        """
+        self.xmpp['xep_0060'].unsubscribe(
+            'pubsub.example.com',
+            'somenode',
+            ifrom='foo@comp.example.com/bar',
+            bare=False,
+            block=False)
+        self.send("""
+          <iq type="set" id="1"
+              to="pubsub.example.com" from="foo@comp.example.com/bar">
+            <pubsub xmlns="http://jabber.org/protocol/pubsub">
+              <unsubscribe node="somenode" jid="foo@comp.example.com/bar" />
+            </pubsub>
+          </iq>
+        """)
+
+    def testUnsubscribeCase4(self):
+        """
+        Test unsubscribing from a node: Case 4:
+        No subscribee, no 'from' JID, full JID
+        """
+        self.stream_close()
+        self.stream_start(jid='tester@localhost/full')
+
+        self.xmpp['xep_0060'].unsubscribe(
+            'pubsub.example.com',
+            'somenode',
+            bare=False,
+            block=False)
+        self.send("""
+          <iq type="set" id="1"
+              to="pubsub.example.com">
+            <pubsub xmlns="http://jabber.org/protocol/pubsub">
+              <unsubscribe node="somenode" jid="tester@localhost/full" />
+            </pubsub>
+          </iq>
+        """)
+
+    def testUnsubscribeCase5(self):
+        """
+        Test unsubscribing from a node: Case 5:
+        Subscribee given
+        """
+        self.xmpp['xep_0060'].unsubscribe(
+            'pubsub.example.com',
+            'somenode',
+            subscribee='user@example.com/foo',
+            ifrom='foo@comp.example.com/bar',
+            block=False)
+        self.send("""
+          <iq type="set" id="1"
+              to="pubsub.example.com" from="foo@comp.example.com/bar">
+            <pubsub xmlns="http://jabber.org/protocol/pubsub">
+              <unsubscribe node="somenode" jid="user@example.com/foo" />
+            </pubsub>
+          </iq>
+        """)
+
+    def testGetDefaultNodeConfig(self):
+        """Test retrieving the default node config for a pubsub service."""
+        self.xmpp['xep_0060'].get_node_config(
+                'pubsub.example.com',
+                block=False)
+        self.send("""
+          <iq type="get" id="1" to="pubsub.example.com">
+            <pubsub xmlns="http://jabber.org/protocol/pubsub#owner">
+              <default />
+            </pubsub>
+          </iq>
+        """, use_values=False)
+
+    def testGetNodeConfig(self):
+        """Test getting the config for a given node."""
+        self.xmpp['xep_0060'].get_node_config(
+            'pubsub.example.com',
+            'somenode',
+            block=False)
+        self.send("""
+          <iq type="get" id="1" to="pubsub.example.com">
+            <pubsub xmlns="http://jabber.org/protocol/pubsub#owner">
+              <configure node="somenode" />
+            </pubsub>
+          </iq>
+        """, use_values=False)
+
+    def testSetNodeConfig(self):
+        """Test setting the configuration for a node."""
+        form = self.xmpp['xep_0004'].make_form()
+        form.add_field(var='FORM_TYPE', ftype='hidden',
+                       value='http://jabber.org/protocol/pubsub#node_config')
+        form.add_field(var='pubsub#title', ftype='text-single',
+                       value='This is awesome!')
+        form['type'] = 'submit'
+
+        self.xmpp['xep_0060'].set_node_config(
+            'pubsub.example.com',
+            'somenode',
+            form,
+            block=False)
+        self.send("""
+          <iq type="set" id="1" to="pubsub.example.com">
+            <pubsub xmlns="http://jabber.org/protocol/pubsub#owner">
+              <configure node="somenode">
+                <x xmlns="jabber:x:data" type="submit">
+                  <field var="FORM_TYPE">
+                    <value>http://jabber.org/protocol/pubsub#node_config</value>
+                  </field>
+                  <field var="pubsub#title">
+                    <value>This is awesome!</value>
+                  </field>
+                </x>
+              </configure>
+            </pubsub>
+          </iq>
+        """)
+
+    def testPublishNoItems(self):
+        """Test publishing no items (in order to generate events)"""
+        self.xmpp['xep_0060'].publish(
+            'pubsub.example.com',
+            'somenode',
+            block=False)
+        self.send("""
+          <iq type="set" id="1" to="pubsub.example.com">
+            <pubsub xmlns="http://jabber.org/protocol/pubsub">
+              <publish node="somenode" />
+            </pubsub>
+          </iq>
+        """)
+
+    def testPublishSingle(self):
+        """Test publishing a single item."""
+        payload = AtomEntry()
+        payload['title'] = 'Test'
+
+        register_stanza_plugin(self.xmpp['xep_0060'].stanza.Item, AtomEntry)
+
+        self.xmpp['xep_0060'].publish(
+            'pubsub.example.com',
+            'somenode',
+            id='id42',
+            payload=payload,
+            block=False)
+        self.send("""
+          <iq type="set" id="1" to="pubsub.example.com">
+            <pubsub xmlns="http://jabber.org/protocol/pubsub">
+              <publish node="somenode">
+                <item id="id42">
+                  <entry xmlns="http://www.w3.org/2005/Atom">
+                    <title>Test</title>
+                  </entry>
+                </item>
+              </publish>
+            </pubsub>
+          </iq>
+        """)
+
+    def testPublishSingleOptions(self):
+        """Test publishing a single item, with options."""
+        payload = AtomEntry()
+        payload['title'] = 'Test'
+
+        register_stanza_plugin(self.xmpp['xep_0060'].stanza.Item, AtomEntry)
+
+        options = self.xmpp['xep_0004'].make_form()
+        options.add_field(var='FORM_TYPE', ftype='hidden',
+              value='http://jabber.org/protocol/pubsub#publish-options')
+        options.add_field(var='pubsub#access_model', ftype='text-single',
+              value='presence')
+        options['type'] = 'submit'
+
+        self.xmpp['xep_0060'].publish(
+            'pubsub.example.com',
+            'somenode',
+            id='ID42',
+            payload=payload,
+            options=options,
+            block=False)
+        self.send("""
+          <iq type="set" id="1" to="pubsub.example.com">
+            <pubsub xmlns="http://jabber.org/protocol/pubsub">
+              <publish node="somenode">
+                <item id="ID42">
+                  <entry xmlns="http://www.w3.org/2005/Atom">
+                    <title>Test</title>
+                  </entry>
+                </item>
+              </publish>
+              <publish-options>
+                <x xmlns="jabber:x:data" type="submit">
+                  <field var="FORM_TYPE">
+                    <value>http://jabber.org/protocol/pubsub#publish-options</value>
+                  </field>
+                  <field var="pubsub#access_model">
+                    <value>presence</value>
+                  </field>
+                </x>
+              </publish-options>
+            </pubsub>
+          </iq>
+        """, use_values=False)
+
+    def testRetract(self):
+        """Test deleting an item."""
+        self.xmpp['xep_0060'].retract(
+            'pubsub.example.com',
+            'somenode',
+            'ID1',
+            notify=True,
+            block=False)
+        self.send("""
+          <iq type="set" id="1" to="pubsub.example.com">
+            <pubsub xmlns="http://jabber.org/protocol/pubsub">
+              <retract node="somenode" notify="true">
+                <item id="ID1" />
+              </retract>
+            </pubsub>
+          </iq>
+        """)
+
+    def testRetract(self):
+        """Test deleting an item."""
+        self.xmpp['xep_0060'].retract(
+            'pubsub.example.com',
+            'somenode',
+            'ID1',
+            block=False)
+        self.send("""
+          <iq type="set" id="1" to="pubsub.example.com">
+            <pubsub xmlns="http://jabber.org/protocol/pubsub">
+              <retract node="somenode">
+                <item id="ID1" />
+              </retract>
+            </pubsub>
+          </iq>
+        """)
+
+    def testPurge(self):
+        """Test removing all items from a node."""
+        self.xmpp['xep_0060'].purge(
+                'pubsub.example.com',
+                'somenode',
+                block=False)
+        self.send("""
+          <iq type="set" id="1" to="pubsub.example.com">
+            <pubsub xmlns="http://jabber.org/protocol/pubsub#owner">
+              <purge node="somenode" />
+            </pubsub>
+          </iq>
+        """)
+
+    def testGetItem(self):
+        """Test retrieving a single item."""
+        self.xmpp['xep_0060'].get_item(
+            'pubsub.example.com',
+            'somenode',
+            'id42',
+            block=False)
+        self.send("""
+          <iq type="get" id="1" to="pubsub.example.com">
+            <pubsub xmlns="http://jabber.org/protocol/pubsub">
+              <items node="somenode">
+                <item id="id42" />
+              </items>
+            </pubsub>
+          </iq>
+        """)
+
+    def testGetLatestItems(self):
+        """Test retrieving the most recent N items."""
+        self.xmpp['xep_0060'].get_items(
+            'pubsub.example.com',
+            'somenode',
+            max_items=3,
+            block=False)
+        self.send("""
+          <iq type="get" id="1" to="pubsub.example.com">
+            <pubsub xmlns="http://jabber.org/protocol/pubsub">
+              <items node="somenode" max_items="3" />
+            </pubsub>
+          </iq>
+        """)
+
+    def testGetAllItems(self):
+        """Test retrieving all items."""
+        self.xmpp['xep_0060'].get_items(
+            'pubsub.example.com',
+            'somenode',
+            block=False)
+        self.send("""
+          <iq type="get" id="1" to="pubsub.example.com">
+            <pubsub xmlns="http://jabber.org/protocol/pubsub">
+              <items node="somenode" />
+            </pubsub>
+          </iq>
+        """)
+
+    def testGetSpecificItems(self):
+        """Test retrieving a specific set of items."""
+        self.xmpp['xep_0060'].get_items(
+            'pubsub.example.com',
+            'somenode',
+            item_ids=['A', 'B', 'C'],
+            block=False)
+        self.send("""
+          <iq type="get" id="1" to="pubsub.example.com">
+            <pubsub xmlns="http://jabber.org/protocol/pubsub">
+              <items node="somenode">
+                <item id="A" />
+                <item id="B" />
+                <item id="C" />
+              </items>
+            </pubsub>
+          </iq>
+        """)
+
+    def testGetSubscriptionGlobalDefaultOptions(self):
+        """Test getting the subscription options for a node/JID."""
+        self.xmpp['xep_0060'].get_subscription_options(
+            'pubsub.example.com',
+            block=False)
+        self.send("""
+          <iq type="get" id="1" to="pubsub.example.com">
+            <pubsub xmlns="http://jabber.org/protocol/pubsub">
+              <default />
+            </pubsub>
+          </iq>
+        """, use_values=False)
+
+    def testGetSubscriptionNodeDefaultOptions(self):
+        """Test getting the subscription options for a node/JID."""
+        self.xmpp['xep_0060'].get_subscription_options(
+            'pubsub.example.com',
+            node='somenode',
+            block=False)
+        self.send("""
+          <iq type="get" id="1" to="pubsub.example.com">
+            <pubsub xmlns="http://jabber.org/protocol/pubsub">
+              <default node="somenode" />
+            </pubsub>
+          </iq>
+        """, use_values=False)
+
+    def testGetSubscriptionOptions(self):
+        """Test getting the subscription options for a node/JID."""
+        self.xmpp['xep_0060'].get_subscription_options(
+            'pubsub.example.com',
+            'somenode',
+            'tester@localhost',
+            block=False)
+        self.send("""
+          <iq type="get" id="1" to="pubsub.example.com">
+            <pubsub xmlns="http://jabber.org/protocol/pubsub">
+              <options node="somenode" jid="tester@localhost" />
+            </pubsub>
+          </iq>
+        """, use_values=False)
+
+    def testSetSubscriptionOptions(self):
+        """Test setting the subscription options for a node/JID."""
+        opts = self.xmpp['xep_0004'].make_form()
+        opts.add_field(
+                var='FORM_TYPE',
+                value='http://jabber.org/protocol/pubsub#subscribe_options',
+                ftype='hidden')
+        opts.add_field(
+                var='pubsub#digest',
+                value=False,
+                ftype='boolean')
+        opts['type'] = 'submit'
+
+        self.xmpp['xep_0060'].set_subscription_options(
+            'pubsub.example.com',
+            'somenode',
+            'tester@localhost',
+            opts,
+            block=False)
+        self.send("""
+          <iq type="get" id="1" to="pubsub.example.com">
+            <pubsub xmlns="http://jabber.org/protocol/pubsub">
+              <options node="somenode" jid="tester@localhost">
+                <x xmlns="jabber:x:data" type="submit">
+                  <field var="FORM_TYPE">
+                    <value>http://jabber.org/protocol/pubsub#subscribe_options</value>
+                  </field>
+                  <field var="pubsub#digest">
+                    <value>0</value>
+                  </field>
+                </x>
+              </options>
+            </pubsub>
+          </iq>
+        """)
+
+    def testGetNodeSubscriptions(self):
+        """Test retrieving all subscriptions for a node."""
+        self.xmpp['xep_0060'].get_node_subscriptions(
+            'pubsub.example.com',
+            'somenode',
+            block=False)
+        self.send("""
+          <iq type="get" id="1" to="pubsub.example.com">
+            <pubsub xmlns="http://jabber.org/protocol/pubsub#owner">
+              <subscriptions node="somenode" />
+            </pubsub>
+          </iq>
+        """)
+
+    def testGetSubscriptions(self):
+        """Test retrieving a users's subscriptions."""
+        self.xmpp['xep_0060'].get_subscriptions(
+            'pubsub.example.com',
+            block=False)
+        self.send("""
+          <iq type="get" id="1" to="pubsub.example.com">
+            <pubsub xmlns="http://jabber.org/protocol/pubsub">
+              <subscriptions />
+            </pubsub>
+          </iq>
+        """)
+
+    def testGetSubscriptionsForNode(self):
+        """Test retrieving a users's subscriptions for a given node."""
+        self.xmpp['xep_0060'].get_subscriptions(
+            'pubsub.example.com',
+            node='somenode',
+            block=False)
+        self.send("""
+          <iq type="get" id="1" to="pubsub.example.com">
+            <pubsub xmlns="http://jabber.org/protocol/pubsub">
+              <subscriptions node="somenode" />
+            </pubsub>
+          </iq>
+        """)
+
+    def testGetAffiliations(self):
+        """Test retrieving a users's affiliations."""
+        self.xmpp['xep_0060'].get_affiliations(
+            'pubsub.example.com',
+            block=False)
+        self.send("""
+          <iq type="get" id="1" to="pubsub.example.com">
+            <pubsub xmlns="http://jabber.org/protocol/pubsub">
+              <affiliations />
+            </pubsub>
+          </iq>
+        """)
+
+    def testGetAffiliatinssForNode(self):
+        """Test retrieving a users's affiliations for a given node."""
+        self.xmpp['xep_0060'].get_affiliations(
+            'pubsub.example.com',
+            node='somenode',
+            block=False)
+        self.send("""
+          <iq type="get" id="1" to="pubsub.example.com">
+            <pubsub xmlns="http://jabber.org/protocol/pubsub">
+              <affiliations node="somenode" />
+            </pubsub>
+          </iq>
+        """)
+
+    def testGetNodeAffiliations(self):
+        """Test getting the affiliations for a node."""
+        self.xmpp['xep_0060'].get_node_affiliations(
+            'pubsub.example.com',
+            'somenode',
+            block=False)
+        self.send("""
+          <iq type="get" id="1" to="pubsub.example.com">
+            <pubsub xmlns="http://jabber.org/protocol/pubsub#owner">
+              <affiliations node="somenode" />
+            </pubsub>
+          </iq>
+        """)
+
+    def testModifySubscriptions(self):
+        """Test owner modifying node subscriptions."""
+        self.xmpp['xep_0060'].modify_subscriptions(
+            'pubsub.example.com',
+            'somenode',
+            subscriptions=[('user@example.com', 'subscribed'),
+                           ('foo@example.net', 'none')],
+            block=False)
+        self.send("""
+          <iq type="set" id="1" to="pubsub.example.com">
+            <pubsub xmlns="http://jabber.org/protocol/pubsub#owner">
+              <subscriptions node="somenode">
+                <subscription jid="user@example.com" subscription="subscribed" />
+                <subscription jid="foo@example.net" subscription="none" />
+              </subscriptions>
+            </pubsub>
+          </iq>
+        """)
+
+    def testModifyAffiliations(self):
+        """Test owner modifying node affiliations."""
+        self.xmpp['xep_0060'].modify_affiliations(
+            'pubsub.example.com',
+            'somenode',
+            affiliations=[('user@example.com', 'publisher'),
+                          ('foo@example.net', 'none')],
+            block=False)
+        self.send("""
+          <iq type="set" id="1" to="pubsub.example.com">
+            <pubsub xmlns="http://jabber.org/protocol/pubsub#owner">
+              <affiliations node="somenode">
+                <affiliation jid="user@example.com" affiliation="publisher" />
+                <affiliation jid="foo@example.net" affiliation="none" />
+              </affiliations>
+            </pubsub>
+          </iq>
+        """)
+
+
+suite = unittest.TestLoader().loadTestsFromTestCase(TestStreamPubsub)

tox.ini

@@ -0,0 +1,5 @@
+[tox]
+envlist = py26,py27,py31,py32
+[testenv]
+deps = nose
+commands = nosetests --where=tests --exclude=live -i sleektest.py