<s> """<STR_LIT>""" <EOL> from twisted . internet import defer <EOL> from twisted . trial import unittest <EOL> from twisted . words . protocols . jabber import error <EOL> from twisted . words . protocols . jabber . jid import JID <EOL> from twisted . words . protocols . jabber . xmlstream import toResponse <EOL> from twisted . words . xish import domish , utility <EOL> from wokkel import xmppim <EOL> from wokkel . generic import ErrorStanza , parseXml <EOL> from wokkel . test . helpers import TestableRequestHandlerMixin , XmlStreamStub <EOL> NS_XML = '<STR_LIT>' <EOL> NS_ROSTER = '<STR_LIT>' <EOL> class PresenceClientProtocolTest ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> self . output = [ ] <EOL> self . protocol = xmppim . PresenceClientProtocol ( ) <EOL> self . protocol . parent = self <EOL> def send ( self , obj ) : <EOL> self . output . append ( obj ) <EOL> def test_unavailableDirected ( self ) : <EOL> """<STR_LIT>""" <EOL> self . protocol . unavailable ( JID ( '<STR_LIT>' ) ) <EOL> presence = self . output [ - <NUM_LIT:1> ] <EOL> self . assertEquals ( "<STR_LIT>" , presence . name ) <EOL> self . assertEquals ( None , presence . uri ) <EOL> self . assertEquals ( "<STR_LIT>" , presence . getAttribute ( '<STR_LIT:to>' ) ) <EOL> self . assertEquals ( "<STR_LIT>" , presence . getAttribute ( '<STR_LIT:type>' ) ) <EOL> def test_unavailableWithStatus ( self ) : <EOL> """<STR_LIT>""" <EOL> self . protocol . unavailable ( JID ( '<STR_LIT>' ) , <EOL> { None : '<STR_LIT>' } ) <EOL> presence = self . output [ - <NUM_LIT:1> ] <EOL> self . assertEquals ( "<STR_LIT>" , presence . name ) <EOL> self . assertEquals ( None , presence . uri ) <EOL> self . assertEquals ( "<STR_LIT>" , presence . getAttribute ( '<STR_LIT:to>' ) ) <EOL> self . assertEquals ( "<STR_LIT>" , presence . getAttribute ( '<STR_LIT:type>' ) ) <EOL> self . assertEquals ( "<STR_LIT>" , unicode ( presence . status ) ) <EOL> def test_unavailableBroadcast ( self ) : <EOL> """<STR_LIT>""" <EOL> self . protocol . unavailable ( None ) <EOL> presence = self . output [ - <NUM_LIT:1> ] <EOL> self . assertEquals ( "<STR_LIT>" , presence . name ) <EOL> self . assertEquals ( None , presence . uri ) <EOL> self . assertEquals ( None , presence . getAttribute ( '<STR_LIT:to>' ) ) <EOL> self . assertEquals ( "<STR_LIT>" , presence . getAttribute ( '<STR_LIT:type>' ) ) <EOL> def test_unavailableBroadcastNoEntityParameter ( self ) : <EOL> """<STR_LIT>""" <EOL> self . protocol . unavailable ( ) <EOL> presence = self . output [ - <NUM_LIT:1> ] <EOL> self . assertEquals ( "<STR_LIT>" , presence . name ) <EOL> self . assertEquals ( None , presence . uri ) <EOL> self . assertEquals ( None , presence . getAttribute ( '<STR_LIT:to>' ) ) <EOL> self . assertEquals ( "<STR_LIT>" , presence . getAttribute ( '<STR_LIT:type>' ) ) <EOL> class AvailabilityPresenceTest ( unittest . TestCase ) : <EOL> def test_fromElement ( self ) : <EOL> xml = """<STR_LIT>""" <EOL> presence = xmppim . AvailabilityPresence . fromElement ( parseXml ( xml ) ) <EOL> self . assertEquals ( JID ( '<STR_LIT>' ) , presence . sender ) <EOL> self . assertEquals ( JID ( '<STR_LIT>' ) , presence . recipient ) <EOL> self . assertTrue ( presence . available ) <EOL> self . assertEquals ( '<STR_LIT>' , presence . show ) <EOL> self . assertEquals ( { None : "<STR_LIT>" } , presence . statuses ) <EOL> self . assertEquals ( <NUM_LIT:50> , presence . priority ) <EOL> class PresenceProtocolTest ( unittest . TestCase ) : <EOL> """<STR_LIT>""" <EOL> def setUp ( self ) : <EOL> self . output = [ ] <EOL> self . protocol = xmppim . PresenceProtocol ( ) <EOL> self . protocol . parent = self <EOL> self . protocol . xmlstream = utility . EventDispatcher ( ) <EOL> self . protocol . connectionInitialized ( ) <EOL> def send ( self , obj ) : <EOL> self . output . append ( obj ) <EOL> def test_errorReceived ( self ) : <EOL> """<STR_LIT>""" <EOL> xml = """<STR_LIT>""" <EOL> def errorReceived ( error ) : <EOL> xmppim . PresenceProtocol . errorReceived ( self . protocol , error ) <EOL> try : <EOL> self . assertIsInstance ( error , ErrorStanza ) <EOL> except : <EOL> d . errback ( ) <EOL> else : <EOL> d . callback ( None ) <EOL> d = defer . Deferred ( ) <EOL> self . protocol . errorReceived = errorReceived <EOL> self . protocol . xmlstream . dispatch ( parseXml ( xml ) ) <EOL> return d <EOL> def test_availableReceived ( self ) : <EOL> """<STR_LIT>""" <EOL> xml = """<STR_LIT>""" <EOL> def availableReceived ( presence ) : <EOL> xmppim . PresenceProtocol . availableReceived ( self . protocol , presence ) <EOL> try : <EOL> self . assertIsInstance ( presence , xmppim . AvailabilityPresence ) <EOL> except : <EOL> d . errback ( ) <EOL> else : <EOL> d . callback ( None ) <EOL> d = defer . Deferred ( ) <EOL> self . protocol . availableReceived = availableReceived <EOL> self . protocol . xmlstream . dispatch ( parseXml ( xml ) ) <EOL> return d <EOL> def test_unavailableReceived ( self ) : <EOL> """<STR_LIT>""" <EOL> xml = """<STR_LIT>""" <EOL> def unavailableReceived ( presence ) : <EOL> xmppim . PresenceProtocol . unavailableReceived ( self . protocol , presence ) <EOL> try : <EOL> self . assertIsInstance ( presence , xmppim . AvailabilityPresence ) <EOL> except : <EOL> d . errback ( ) <EOL> else : <EOL> d . callback ( None ) <EOL> d = defer . Deferred ( ) <EOL> self . protocol . unavailableReceived = unavailableReceived <EOL> self . protocol . xmlstream . dispatch ( parseXml ( xml ) ) <EOL> return d <EOL> def test_subscribeReceived ( self ) : <EOL> """<STR_LIT>""" <EOL> xml = """<STR_LIT>""" <EOL> def subscribeReceived ( presence ) : <EOL> xmppim . PresenceProtocol . subscribeReceived ( self . protocol , presence ) <EOL> try : <EOL> self . assertIsInstance ( presence , xmppim . SubscriptionPresence ) <EOL> except : <EOL> d . errback ( ) <EOL> else : <EOL> d . callback ( None ) <EOL> d = defer . Deferred ( ) <EOL> self . protocol . subscribeReceived = subscribeReceived <EOL> self . protocol . xmlstream . dispatch ( parseXml ( xml ) ) <EOL> return d <EOL> def test_unsubscribeReceived ( self ) : <EOL> """<STR_LIT>""" <EOL> xml = """<STR_LIT>""" <EOL> def unsubscribeReceived ( presence ) : <EOL> xmppim . PresenceProtocol . unsubscribeReceived ( self . protocol , presence ) <EOL> try : <EOL> self . assertIsInstance ( presence , xmppim . SubscriptionPresence ) <EOL> except : <EOL> d . errback ( ) <EOL> else : <EOL> d . callback ( None ) <EOL> d = defer . Deferred ( ) <EOL> self . protocol . unsubscribeReceived = unsubscribeReceived <EOL> self . protocol . xmlstream . dispatch ( parseXml ( xml ) ) <EOL> return d <EOL> def test_subscribedReceived ( self ) : <EOL> """<STR_LIT>""" <EOL> xml = """<STR_LIT>""" <EOL> def subscribedReceived ( presence ) : <EOL> xmppim . PresenceProtocol . subscribedReceived ( self . protocol , presence ) <EOL> try : <EOL> self . assertIsInstance ( presence , xmppim . SubscriptionPresence ) <EOL> except : <EOL> d . errback ( ) <EOL> else : <EOL> d . callback ( None ) <EOL> d = defer . Deferred ( ) <EOL> self . protocol . subscribedReceived = subscribedReceived <EOL> self . protocol . xmlstream . dispatch ( parseXml ( xml ) ) <EOL> return d <EOL> def test_unsubscribedReceived ( self ) : <EOL> """<STR_LIT>""" <EOL> xml = """<STR_LIT>""" <EOL> def unsubscribedReceived ( presence ) : <EOL> xmppim . PresenceProtocol . unsubscribedReceived ( self . protocol , <EOL> presence ) <EOL> try : <EOL> self . assertIsInstance ( presence , xmppim . SubscriptionPresence ) <EOL> except : <EOL> d . errback ( ) <EOL> else : <EOL> d . callback ( None ) <EOL> d = defer . Deferred ( ) <EOL> self . protocol . unsubscribedReceived = unsubscribedReceived <EOL> self . protocol . xmlstream . dispatch ( parseXml ( xml ) ) <EOL> return d <EOL> def test_probeReceived ( self ) : <EOL> """<STR_LIT>""" <EOL> xml = """<STR_LIT>""" <EOL> def probeReceived ( presence ) : <EOL> xmppim . PresenceProtocol . probeReceived ( self . protocol , presence ) <EOL> try : <EOL> self . assertIsInstance ( presence , xmppim . ProbePresence ) <EOL> except : <EOL> d . errback ( ) <EOL> else : <EOL> d . callback ( None ) <EOL> d = defer . Deferred ( ) <EOL> self . protocol . probeReceived = probeReceived <EOL> self . protocol . xmlstream . dispatch ( parseXml ( xml ) ) <EOL> return d <EOL> def test_available ( self ) : <EOL> """<STR_LIT>""" <EOL> self . protocol . available ( JID ( '<STR_LIT>' ) , <EOL> show = u'<STR_LIT>' , <EOL> status = u'<STR_LIT>' , <EOL> priority = <NUM_LIT:50> ) <EOL> element = self . output [ - <NUM_LIT:1> ] <EOL> self . assertEquals ( "<STR_LIT>" , element . getAttribute ( '<STR_LIT:to>' ) ) <EOL> self . assertIdentical ( None , element . getAttribute ( '<STR_LIT:type>' ) ) <EOL> self . assertEquals ( u'<STR_LIT>' , unicode ( element . show ) ) <EOL> self . assertEquals ( u'<STR_LIT>' , unicode ( element . status ) ) <EOL> self . assertEquals ( u'<STR_LIT>' , unicode ( element . priority ) ) <EOL> def test_availableLanguages ( self ) : <EOL> """<STR_LIT>""" <EOL> self . protocol . available ( JID ( '<STR_LIT>' ) , <EOL> show = u'<STR_LIT>' , <EOL> statuses = { None : u'<STR_LIT>' , <EOL> '<STR_LIT>' : u'<STR_LIT>' } , <EOL> priority = <NUM_LIT:50> ) <EOL> element = self . output [ - <NUM_LIT:1> ] <EOL> self . assertEquals ( "<STR_LIT>" , element . getAttribute ( '<STR_LIT:to>' ) ) <EOL> self . assertIdentical ( None , element . getAttribute ( '<STR_LIT:type>' ) ) <EOL> self . assertEquals ( u'<STR_LIT>' , unicode ( element . show ) ) <EOL> statuses = { } <EOL> for status in element . elements ( ) : <EOL> if status . name == '<STR_LIT:status>' : <EOL> lang = status . getAttribute ( ( NS_XML , '<STR_LIT>' ) ) <EOL> statuses [ lang ] = unicode ( status ) <EOL> self . assertIn ( None , statuses ) <EOL> self . assertEquals ( u'<STR_LIT>' , statuses [ None ] ) <EOL> self . assertIn ( '<STR_LIT>' , statuses ) <EOL> self . assertEquals ( u'<STR_LIT>' , statuses [ '<STR_LIT>' ] ) <EOL> self . assertEquals ( u'<STR_LIT>' , unicode ( element . priority ) ) <EOL> def test_availableSender ( self ) : <EOL> """<STR_LIT>""" <EOL> self . protocol . available ( JID ( '<STR_LIT>' ) , <EOL> sender = JID ( '<STR_LIT>' ) ) <EOL> element = self . output [ - <NUM_LIT:1> ] <EOL> self . assertEquals ( "<STR_LIT>" , element . getAttribute ( '<STR_LIT>' ) ) <EOL> def test_unavailableDirected ( self ) : <EOL> """<STR_LIT>""" <EOL> self . protocol . unavailable ( JID ( '<STR_LIT>' ) ) <EOL> element = self . output [ - <NUM_LIT:1> ] <EOL> self . assertEquals ( "<STR_LIT>" , element . name ) <EOL> self . assertEquals ( None , element . uri ) <EOL> self . assertEquals ( "<STR_LIT>" , element . getAttribute ( '<STR_LIT:to>' ) ) <EOL> self . assertEquals ( "<STR_LIT>" , element . getAttribute ( '<STR_LIT:type>' ) ) <EOL> def test_unavailableWithStatus ( self ) : <EOL> """<STR_LIT>""" <EOL> self . protocol . unavailable ( JID ( '<STR_LIT>' ) , <EOL> { None : '<STR_LIT>' } ) <EOL> element = self . output [ - <NUM_LIT:1> ] <EOL> self . assertEquals ( "<STR_LIT>" , element . name ) <EOL> self . assertEquals ( None , element . uri ) <EOL> self . assertEquals ( "<STR_LIT>" , element . getAttribute ( '<STR_LIT:to>' ) ) <EOL> self . assertEquals ( "<STR_LIT>" , element . getAttribute ( '<STR_LIT:type>' ) ) <EOL> self . assertEquals ( "<STR_LIT>" , unicode ( element . status ) ) <EOL> def test_unavailableBroadcast ( self ) : <EOL> """<STR_LIT>""" <EOL> self . protocol . unavailable ( None ) <EOL> element = self . output [ - <NUM_LIT:1> ] <EOL> self . assertEquals ( "<STR_LIT>" , element . name ) <EOL> self . assertEquals ( None , element . uri ) <EOL> self . assertEquals ( None , element . getAttribute ( '<STR_LIT:to>' ) ) <EOL> self . assertEquals ( "<STR_LIT>" , element . getAttribute ( '<STR_LIT:type>' ) ) <EOL> def test_unavailableBroadcastNoRecipientParameter ( self ) : <EOL> """<STR_LIT>""" <EOL> self . protocol . unavailable ( ) <EOL> element = self . output [ - <NUM_LIT:1> ] <EOL> self . assertEquals ( "<STR_LIT>" , element . name ) <EOL> self . assertEquals ( None , element . uri ) <EOL> self . assertEquals ( None , element . getAttribute ( '<STR_LIT:to>' ) ) <EOL> self . assertEquals ( "<STR_LIT>" , element . getAttribute ( '<STR_LIT:type>' ) ) <EOL> def test_unavailableSender ( self ) : <EOL> """<STR_LIT>""" <EOL> self . protocol . unavailable ( JID ( '<STR_LIT>' ) , <EOL> sender = JID ( '<STR_LIT>' ) ) <EOL> element = self . output [ - <NUM_LIT:1> ] <EOL> self . assertEquals ( "<STR_LIT>" , element . getAttribute ( '<STR_LIT>' ) ) <EOL> def test_subscribeSender ( self ) : <EOL> """<STR_LIT>""" <EOL> self . protocol . subscribe ( JID ( '<STR_LIT>' ) , <EOL> sender = JID ( '<STR_LIT>' ) ) <EOL> element = self . output [ - <NUM_LIT:1> ] <EOL> self . assertEquals ( "<STR_LIT>" , element . getAttribute ( '<STR_LIT>' ) ) <EOL> def test_unsubscribeSender ( self ) : <EOL> """<STR_LIT>""" <EOL> self . protocol . unsubscribe ( JID ( '<STR_LIT>' ) , <EOL> sender = JID ( '<STR_LIT>' ) ) <EOL> element = self . output [ - <NUM_LIT:1> ] <EOL> self . assertEquals ( "<STR_LIT>" , element . getAttribute ( '<STR_LIT>' ) ) <EOL> def test_subscribedSender ( self ) : <EOL> """<STR_LIT>""" <EOL> self . protocol . subscribed ( JID ( '<STR_LIT>' ) , <EOL> sender = JID ( '<STR_LIT>' ) ) <EOL> element = self . output [ - <NUM_LIT:1> ] <EOL> self . assertEquals ( "<STR_LIT>" , element . getAttribute ( '<STR_LIT>' ) ) <EOL> def test_unsubscribedSender ( self ) : <EOL> """<STR_LIT>""" <EOL> self . protocol . unsubscribed ( JID ( '<STR_LIT>' ) , <EOL> sender = JID ( '<STR_LIT>' ) ) <EOL> element = self . output [ - <NUM_LIT:1> ] <EOL> self . assertEquals ( "<STR_LIT>" , element . getAttribute ( '<STR_LIT>' ) ) <EOL> def test_probeSender ( self ) : <EOL> """<STR_LIT>""" <EOL> self . protocol . probe ( JID ( '<STR_LIT>' ) , <EOL> sender = JID ( '<STR_LIT>' ) ) <EOL> element = self . output [ - <NUM_LIT:1> ] <EOL> self . assertEquals ( "<STR_LIT>" , element . getAttribute ( '<STR_LIT>' ) ) <EOL> class RosterItemTest ( unittest . TestCase ) : <EOL> """<STR_LIT>""" <EOL> def test_toElement ( self ) : <EOL> """<STR_LIT>""" <EOL> item = xmppim . RosterItem ( JID ( '<STR_LIT>' ) ) <EOL> element = item . toElement ( ) <EOL> self . assertEqual ( '<STR_LIT>' , element . name ) <EOL> self . assertEqual ( NS_ROSTER , element . uri ) <EOL> self . assertFalse ( element . hasAttribute ( '<STR_LIT>' ) ) <EOL> self . assertFalse ( element . hasAttribute ( '<STR_LIT>' ) ) <EOL> self . assertEqual ( u"<STR_LIT>" , element . getAttribute ( '<STR_LIT:name>' , u"<STR_LIT>" ) ) <EOL> self . assertFalse ( element . hasAttribute ( '<STR_LIT>' ) ) <EOL> self . assertEquals ( <NUM_LIT:0> , len ( list ( element . elements ( ) ) ) ) <EOL> def test_toElementMinimal ( self ) : <EOL> """<STR_LIT>""" <EOL> item = xmppim . RosterItem ( JID ( '<STR_LIT>' ) ) <EOL> element = item . toElement ( ) <EOL> self . assertEqual ( u'<STR_LIT>' , element . getAttribute ( '<STR_LIT>' ) ) <EOL> def test_toElementSubscriptionNone ( self ) : <EOL> """<STR_LIT>""" <EOL> item = xmppim . RosterItem ( JID ( '<STR_LIT>' ) , <EOL> subscriptionTo = False , <EOL> subscriptionFrom = False ) <EOL> element = item . toElement ( ) <EOL> self . assertIdentical ( None , element . getAttribute ( '<STR_LIT>' ) ) <EOL> def test_toElementSubscriptionTo ( self ) : <EOL> """<STR_LIT>""" <EOL> item = xmppim . RosterItem ( JID ( '<STR_LIT>' ) , <EOL> subscriptionTo = True , <EOL> subscriptionFrom = False ) <EOL> element = item . toElement ( ) <EOL> self . assertEqual ( '<STR_LIT:to>' , element . getAttribute ( '<STR_LIT>' ) ) <EOL> def test_toElementSubscriptionFrom ( self ) : <EOL> """<STR_LIT>""" <EOL> item = xmppim . RosterItem ( JID ( '<STR_LIT>' ) , <EOL> subscriptionTo = False , <EOL> subscriptionFrom = True ) <EOL> element = item . toElement ( ) <EOL> self . assertEqual ( '<STR_LIT>' , element . getAttribute ( '<STR_LIT>' ) ) <EOL> def test_toElementSubscriptionBoth ( self ) : <EOL> """<STR_LIT>""" <EOL> item = xmppim . RosterItem ( JID ( '<STR_LIT>' ) , <EOL> subscriptionTo = True , <EOL> subscriptionFrom = True ) <EOL> element = item . toElement ( ) <EOL> self . assertEqual ( '<STR_LIT>' , element . getAttribute ( '<STR_LIT>' ) ) <EOL> def test_toElementSubscriptionRemove ( self ) : <EOL> """<STR_LIT>""" <EOL> item = xmppim . RosterItem ( JID ( '<STR_LIT>' ) ) <EOL> item . remove = True <EOL> element = item . toElement ( ) <EOL> self . assertEqual ( '<STR_LIT>' , element . getAttribute ( '<STR_LIT>' ) ) <EOL> def test_toElementAsk ( self ) : <EOL> """<STR_LIT>""" <EOL> item = xmppim . RosterItem ( JID ( '<STR_LIT>' ) ) <EOL> item . pendingOut = True <EOL> element = item . toElement ( ) <EOL> self . assertEqual ( '<STR_LIT>' , element . getAttribute ( '<STR_LIT>' ) ) <EOL> def test_toElementName ( self ) : <EOL> """<STR_LIT>""" <EOL> item = xmppim . RosterItem ( JID ( '<STR_LIT>' ) , <EOL> name = '<STR_LIT>' ) <EOL> element = item . toElement ( ) <EOL> self . assertEqual ( u'<STR_LIT>' , element . getAttribute ( '<STR_LIT:name>' ) ) <EOL> def test_toElementGroups ( self ) : <EOL> """<STR_LIT>""" <EOL> groups = set ( [ '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> item = xmppim . RosterItem ( JID ( '<STR_LIT>' ) , <EOL> groups = groups ) <EOL> element = item . toElement ( ) <EOL> foundGroups = set ( ) <EOL> for child in element . elements ( ) : <EOL> if child . uri == NS_ROSTER and child . name == '<STR_LIT>' : <EOL> foundGroups . add ( unicode ( child ) ) <EOL> self . assertEqual ( groups , foundGroups ) <EOL> def test_toElementApproved ( self ) : <EOL> """<STR_LIT>""" <EOL> item = xmppim . RosterItem ( JID ( '<STR_LIT>' ) ) <EOL> item . approved = True <EOL> element = item . toElement ( ) <EOL> self . assertEqual ( u'<STR_LIT:true>' , element . getAttribute ( '<STR_LIT>' ) ) <EOL> def test_fromElementMinimal ( self ) : <EOL> """<STR_LIT>""" <EOL> xml = """<STR_LIT>""" <EOL> item = xmppim . RosterItem . fromElement ( parseXml ( xml ) ) <EOL> self . assertEqual ( JID ( u"<STR_LIT>" ) , item . entity ) <EOL> self . assertEqual ( u"<STR_LIT>" , item . name ) <EOL> self . assertFalse ( item . subscriptionTo ) <EOL> self . assertFalse ( item . subscriptionFrom ) <EOL> self . assertFalse ( item . pendingOut ) <EOL> self . assertFalse ( item . approved ) <EOL> self . assertEqual ( set ( ) , item . groups ) <EOL> def test_fromElementName ( self ) : <EOL> """<STR_LIT>""" <EOL> xml = """<STR_LIT>""" <EOL> item = xmppim . RosterItem . fromElement ( parseXml ( xml ) ) <EOL> self . assertEqual ( u"<STR_LIT>" , item . name ) <EOL> def test_fromElementGroups ( self ) : <EOL> """<STR_LIT>""" <EOL> xml = """<STR_LIT>""" <EOL> item = xmppim . RosterItem . fromElement ( parseXml ( xml ) ) <EOL> self . assertIn ( u"<STR_LIT>" , item . groups ) <EOL> self . assertIn ( u"<STR_LIT>" , item . groups ) <EOL> def test_fromElementSubscriptionNone ( self ) : <EOL> """<STR_LIT>""" <EOL> xml = """<STR_LIT>""" <EOL> item = xmppim . RosterItem . fromElement ( parseXml ( xml ) ) <EOL> self . assertFalse ( item . remove ) <EOL> self . assertFalse ( item . subscriptionTo ) <EOL> self . assertFalse ( item . subscriptionFrom ) <EOL> def test_fromElementSubscriptionTo ( self ) : <EOL> """<STR_LIT>""" <EOL> xml = """<STR_LIT>""" <EOL> item = xmppim . RosterItem . fromElement ( parseXml ( xml ) ) <EOL> self . assertFalse ( item . remove ) <EOL> self . assertTrue ( item . subscriptionTo ) <EOL> self . assertFalse ( item . subscriptionFrom ) <EOL> def test_fromElementSubscriptionFrom ( self ) : <EOL> """<STR_LIT>""" <EOL> xml = """<STR_LIT>""" <EOL> item = xmppim . RosterItem . fromElement ( parseXml ( xml ) ) <EOL> self . assertFalse ( item . remove ) <EOL> self . assertFalse ( item . subscriptionTo ) <EOL> self . assertTrue ( item . subscriptionFrom ) <EOL> def test_fromElementSubscriptionBoth ( self ) : <EOL> """<STR_LIT>""" <EOL> xml = """<STR_LIT>""" <EOL> item = xmppim . RosterItem . fromElement ( parseXml ( xml ) ) <EOL> self . assertFalse ( item . remove ) <EOL> self . assertTrue ( item . subscriptionTo ) <EOL> self . assertTrue ( item . subscriptionFrom ) <EOL> def test_fromElementSubscriptionRemove ( self ) : <EOL> """<STR_LIT>""" <EOL> xml = """<STR_LIT>""" <EOL> item = xmppim . RosterItem . fromElement ( parseXml ( xml ) ) <EOL> self . assertTrue ( item . remove ) <EOL> def test_fromElementPendingOut ( self ) : <EOL> """<STR_LIT>""" <EOL> xml = """<STR_LIT>""" <EOL> item = xmppim . RosterItem . fromElement ( parseXml ( xml ) ) <EOL> self . assertTrue ( item . pendingOut ) <EOL> def test_fromElementApprovedTrue ( self ) : <EOL> """<STR_LIT>""" <EOL> xml = """<STR_LIT>""" <EOL> item = xmppim . RosterItem . fromElement ( parseXml ( xml ) ) <EOL> self . assertTrue ( item . approved ) <EOL> def test_fromElementApproved1 ( self ) : <EOL> """<STR_LIT>""" <EOL> xml = """<STR_LIT>""" <EOL> item = xmppim . RosterItem . fromElement ( parseXml ( xml ) ) <EOL> self . assertTrue ( item . approved ) <EOL> def test_jidDeprecationGet ( self ) : <EOL> """<STR_LIT>""" <EOL> item = xmppim . RosterItem ( JID ( '<STR_LIT>' ) ) <EOL> entity = self . assertWarns ( DeprecationWarning , <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> xmppim . __file__ , <EOL> getattr , item , '<STR_LIT>' ) <EOL> self . assertIdentical ( entity , item . entity ) <EOL> def test_jidDeprecationSet ( self ) : <EOL> """<STR_LIT>""" <EOL> item = xmppim . RosterItem ( JID ( '<STR_LIT>' ) ) <EOL> self . assertWarns ( DeprecationWarning , <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> xmppim . __file__ , <EOL> setattr , item , '<STR_LIT>' , <EOL> JID ( '<STR_LIT>' ) ) <EOL> self . assertEqual ( JID ( '<STR_LIT>' ) , item . entity ) <EOL> def test_askDeprecationGet ( self ) : <EOL> """<STR_LIT>""" <EOL> item = xmppim . RosterItem ( JID ( '<STR_LIT>' ) ) <EOL> item . pendingOut = True <EOL> ask = self . assertWarns ( DeprecationWarning , <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> xmppim . __file__ , <EOL> getattr , item , '<STR_LIT>' ) <EOL> self . assertTrue ( ask ) <EOL> def test_askDeprecationSet ( self ) : <EOL> """<STR_LIT>""" <EOL> item = xmppim . RosterItem ( JID ( '<STR_LIT>' ) ) <EOL> self . assertWarns ( DeprecationWarning , <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> xmppim . __file__ , <EOL> setattr , item , '<STR_LIT>' , <EOL> True ) <EOL> self . assertTrue ( item . pendingOut ) <EOL> class RosterRequestTest ( unittest . TestCase ) : <EOL> """<STR_LIT>""" <EOL> def test_fromElement ( self ) : <EOL> """<STR_LIT>""" <EOL> xml = """<STR_LIT>""" <EOL> request = xmppim . RosterRequest . fromElement ( parseXml ( xml ) ) <EOL> self . assertEqual ( '<STR_LIT>' , request . stanzaType ) <EOL> self . assertEqual ( JID ( '<STR_LIT>' ) , request . recipient ) <EOL> self . assertEqual ( JID ( '<STR_LIT>' ) , request . sender ) <EOL> self . assertEqual ( None , request . item ) <EOL> self . assertEqual ( None , request . version ) <EOL> def test_fromElementItem ( self ) : <EOL> """<STR_LIT>""" <EOL> xml = """<STR_LIT>""" <EOL> request = xmppim . RosterRequest . fromElement ( parseXml ( xml ) ) <EOL> self . assertNotIdentical ( None , request . item ) <EOL> self . assertEqual ( JID ( '<STR_LIT>' ) , request . item . entity ) <EOL> def test_fromElementVersion ( self ) : <EOL> """<STR_LIT>""" <EOL> xml = """<STR_LIT>""" <EOL> request = xmppim . RosterRequest . fromElement ( parseXml ( xml ) ) <EOL> self . assertEqual ( '<STR_LIT>' , request . version ) <EOL> def test_fromElementVersionEmpty ( self ) : <EOL> """<STR_LIT>""" <EOL> xml = """<STR_LIT>""" <EOL> request = xmppim . RosterRequest . fromElement ( parseXml ( xml ) ) <EOL> self . assertEqual ( '<STR_LIT>' , request . version ) <EOL> def test_toElement ( self ) : <EOL> """<STR_LIT>""" <EOL> request = xmppim . RosterRequest ( ) <EOL> element = request . toElement ( ) <EOL> children = element . elements ( ) <EOL> child = children . next ( ) <EOL> self . assertEqual ( NS_ROSTER , child . uri ) <EOL> self . assertEqual ( '<STR_LIT>' , child . name ) <EOL> def test_toElementItem ( self ) : <EOL> """<STR_LIT>""" <EOL> request = xmppim . RosterRequest ( ) <EOL> request . item = xmppim . RosterItem ( JID ( '<STR_LIT>' ) ) <EOL> element = request . toElement ( ) <EOL> children = element . query . elements ( ) <EOL> child = children . next ( ) <EOL> self . assertEqual ( NS_ROSTER , child . uri ) <EOL> self . assertEqual ( '<STR_LIT>' , child . name ) <EOL> class FakeClient ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , xmlstream , jid ) : <EOL> self . xmlstream = xmlstream <EOL> self . jid = jid <EOL> def request ( self , request ) : <EOL> element = request . toElement ( ) <EOL> self . xmlstream . send ( element ) <EOL> return defer . Deferred ( ) <EOL> def addHandler ( self , handler ) : <EOL> handler . makeConnection ( self . xmlstream ) <EOL> handler . connectionInitialized ( ) <EOL> def test_toElementVersion ( self ) : <EOL> """<STR_LIT>""" <EOL> request = xmppim . RosterRequest ( ) <EOL> request . version = '<STR_LIT>' <EOL> element = request . toElement ( ) <EOL> self . assertEqual ( '<STR_LIT>' , element . query . getAttribute ( '<STR_LIT>' ) ) <EOL> def test_toElementVersionEmpty ( self ) : <EOL> """<STR_LIT>""" <EOL> request = xmppim . RosterRequest ( ) <EOL> request . version = '<STR_LIT>' <EOL> element = request . toElement ( ) <EOL> self . assertEqual ( '<STR_LIT>' , element . query . getAttribute ( '<STR_LIT>' ) ) <EOL> class RosterClientProtocolTest ( unittest . TestCase , TestableRequestHandlerMixin ) : <EOL> """<STR_LIT>""" <EOL> def setUp ( self ) : <EOL> self . stub = XmlStreamStub ( ) <EOL> self . client = FakeClient ( self . stub . xmlstream , JID ( '<STR_LIT>' ) ) <EOL> self . service = xmppim . RosterClientProtocol ( ) <EOL> self . service . setHandlerParent ( self . client ) <EOL> def test_setItem ( self ) : <EOL> """<STR_LIT>""" <EOL> item = xmppim . RosterItem ( JID ( '<STR_LIT>' ) , <EOL> name = '<STR_LIT>' , <EOL> groups = set ( [ '<STR_LIT>' , '<STR_LIT>' ] ) ) <EOL> d = self . service . setItem ( item ) <EOL> iq = self . stub . output [ - <NUM_LIT:1> ] <EOL> self . assertEqual ( '<STR_LIT>' , iq . getAttribute ( '<STR_LIT:type>' ) ) <EOL> self . assertNotIdentical ( None , iq . query ) <EOL> self . assertEqual ( NS_ROSTER , iq . query . uri ) <EOL> children = list ( domish . generateElementsQNamed ( iq . query . children , <EOL> '<STR_LIT>' , NS_ROSTER ) ) <EOL> self . assertEqual ( <NUM_LIT:1> , len ( children ) ) <EOL> child = children [ <NUM_LIT:0> ] <EOL> self . assertEqual ( '<STR_LIT>' , child [ '<STR_LIT>' ] ) <EOL> self . assertIdentical ( None , child . getAttribute ( '<STR_LIT>' ) ) <EOL> response = toResponse ( iq , '<STR_LIT:result>' ) <EOL> d . callback ( response ) <EOL> return d <EOL> def test_setItemIgnoreAttributes ( self ) : <EOL> """<STR_LIT>""" <EOL> item = xmppim . RosterItem ( JID ( '<STR_LIT>' ) , <EOL> subscriptionTo = True , <EOL> subscriptionFrom = False , <EOL> name = '<STR_LIT>' , <EOL> groups = set ( [ '<STR_LIT>' , '<STR_LIT>' ] ) ) <EOL> item . pendingOut = True <EOL> item . approved = True <EOL> d = self . service . setItem ( item ) <EOL> iq = self . stub . output [ - <NUM_LIT:1> ] <EOL> self . assertEqual ( '<STR_LIT>' , iq . getAttribute ( '<STR_LIT:type>' ) ) <EOL> self . assertNotIdentical ( None , iq . query ) <EOL> self . assertEqual ( NS_ROSTER , iq . query . uri ) <EOL> children = list ( domish . generateElementsQNamed ( iq . query . children , <EOL> '<STR_LIT>' , NS_ROSTER ) ) <EOL> self . assertEqual ( <NUM_LIT:1> , len ( children ) ) <EOL> child = children [ <NUM_LIT:0> ] <EOL> self . assertIdentical ( None , child . getAttribute ( '<STR_LIT>' ) ) <EOL> self . assertIdentical ( None , child . getAttribute ( '<STR_LIT>' ) ) <EOL> self . assertIdentical ( None , child . getAttribute ( '<STR_LIT>' ) ) <EOL> response = toResponse ( iq , '<STR_LIT:result>' ) <EOL> d . callback ( response ) <EOL> return d <EOL> def test_removeItem ( self ) : <EOL> """<STR_LIT>""" <EOL> d = self . service . removeItem ( JID ( '<STR_LIT>' ) ) <EOL> iq = self . stub . output [ - <NUM_LIT:1> ] <EOL> self . assertEqual ( '<STR_LIT>' , iq . getAttribute ( '<STR_LIT:type>' ) ) <EOL> self . assertNotIdentical ( None , iq . query ) <EOL> self . assertEqual ( NS_ROSTER , iq . query . uri ) <EOL> children = list ( domish . generateElementsQNamed ( iq . query . children , <EOL> '<STR_LIT>' , NS_ROSTER ) ) <EOL> self . assertEqual ( <NUM_LIT:1> , len ( children ) ) <EOL> child = children [ <NUM_LIT:0> ] <EOL> self . assertEqual ( '<STR_LIT>' , child [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( '<STR_LIT>' , child . getAttribute ( '<STR_LIT>' ) ) <EOL> response = toResponse ( iq , '<STR_LIT:result>' ) <EOL> d . callback ( response ) <EOL> return d <EOL> def test_getRoster ( self ) : <EOL> """<STR_LIT>""" <EOL> def cb ( roster ) : <EOL> self . assertIn ( JID ( '<STR_LIT>' ) , roster ) <EOL> self . assertIdentical ( None , getattr ( roster , '<STR_LIT:version>' ) ) <EOL> d = self . service . getRoster ( ) <EOL> d . addCallback ( cb ) <EOL> iq = self . stub . output [ - <NUM_LIT:1> ] <EOL> self . assertEqual ( '<STR_LIT>' , iq . getAttribute ( '<STR_LIT:type>' ) ) <EOL> self . assertNotIdentical ( None , iq . query ) <EOL> self . assertEqual ( NS_ROSTER , iq . query . uri ) <EOL> self . assertFalse ( iq . query . hasAttribute ( '<STR_LIT>' ) ) <EOL> response = toResponse ( iq , '<STR_LIT:result>' ) <EOL> query = response . addElement ( ( NS_ROSTER , '<STR_LIT>' ) ) <EOL> item = query . addElement ( '<STR_LIT>' ) <EOL> item [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> d . callback ( response ) <EOL> return d <EOL> def test_getRosterVer ( self ) : <EOL> """<STR_LIT>""" <EOL> def cb ( roster ) : <EOL> self . assertEqual ( '<STR_LIT>' , getattr ( roster , '<STR_LIT:version>' ) ) <EOL> d = self . service . getRoster ( version = '<STR_LIT>' ) <EOL> d . addCallback ( cb ) <EOL> iq = self . stub . output [ - <NUM_LIT:1> ] <EOL> self . assertEqual ( '<STR_LIT>' , iq . query . getAttribute ( '<STR_LIT>' ) ) <EOL> response = toResponse ( iq , '<STR_LIT:result>' ) <EOL> query = response . addElement ( ( NS_ROSTER , '<STR_LIT>' ) ) <EOL> query [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> item = query . addElement ( '<STR_LIT>' ) <EOL> item [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> d . callback ( response ) <EOL> return d <EOL> def test_getRosterVerEmptyResult ( self ) : <EOL> """<STR_LIT>""" <EOL> def cb ( response ) : <EOL> self . assertIdentical ( None , response ) <EOL> d = self . service . getRoster ( version = '<STR_LIT>' ) <EOL> d . addCallback ( cb ) <EOL> iq = self . stub . output [ - <NUM_LIT:1> ] <EOL> response = toResponse ( iq , '<STR_LIT:result>' ) <EOL> d . callback ( response ) <EOL> return d <EOL> def test_onRosterSet ( self ) : <EOL> """<STR_LIT>""" <EOL> xml = """<STR_LIT>""" <EOL> items = [ ] <EOL> def onRosterSet ( item ) : <EOL> items . append ( item ) <EOL> def cb ( result ) : <EOL> self . assertEqual ( <NUM_LIT:1> , len ( items ) ) <EOL> self . assertEqual ( JID ( '<STR_LIT>' ) , items [ <NUM_LIT:0> ] . entity ) <EOL> self . service . onRosterSet = onRosterSet <EOL> d = self . assertWarns ( DeprecationWarning , <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> xmppim . __file__ , <EOL> self . handleRequest , xml ) <EOL> d . addCallback ( cb ) <EOL> return d <EOL> def test_onRosterRemove ( self ) : <EOL> """<STR_LIT>""" <EOL> xml = """<STR_LIT>""" <EOL> entities = [ ] <EOL> def onRosterRemove ( entity ) : <EOL> entities . append ( entity ) <EOL> def cb ( result ) : <EOL> self . assertEqual ( [ JID ( '<STR_LIT>' ) ] , entities ) <EOL> self . service . onRosterRemove = onRosterRemove <EOL> d = self . assertWarns ( DeprecationWarning , <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> xmppim . __file__ , <EOL> self . handleRequest , xml ) <EOL> d . addCallback ( cb ) <EOL> return d <EOL> def test_setReceived ( self ) : <EOL> """<STR_LIT>""" <EOL> xml = """<STR_LIT>""" <EOL> requests = [ ] <EOL> def setReceived ( request ) : <EOL> requests . append ( request ) <EOL> def cb ( result ) : <EOL> self . assertEqual ( <NUM_LIT:1> , len ( requests ) , "<STR_LIT>" ) <EOL> self . assertEqual ( JID ( '<STR_LIT>' ) , requests [ <NUM_LIT:0> ] . item . entity ) <EOL> self . service . setReceived = setReceived <EOL> d = self . handleRequest ( xml ) <EOL> d . addCallback ( cb ) <EOL> return d <EOL> def test_setReceivedOtherSource ( self ) : <EOL> """<STR_LIT>""" <EOL> xml = """<STR_LIT>""" <EOL> def cb ( result ) : <EOL> self . assertEquals ( '<STR_LIT>' , result . condition ) <EOL> d = self . handleRequest ( xml ) <EOL> self . assertFailure ( d , error . StanzaError ) <EOL> d . addCallback ( cb ) <EOL> return d <EOL> def test_setReceivedOtherSourceAllowed ( self ) : <EOL> """<STR_LIT>""" <EOL> xml = """<STR_LIT>""" <EOL> self . service . allowAnySender = True <EOL> requests = [ ] <EOL> def setReceived ( request ) : <EOL> requests . append ( request ) <EOL> def cb ( result ) : <EOL> self . assertEqual ( <NUM_LIT:1> , len ( requests ) , "<STR_LIT>" ) <EOL> self . service . setReceived = setReceived <EOL> d = self . handleRequest ( xml ) <EOL> d . addCallback ( cb ) <EOL> return d <EOL> def test_setReceivedOtherSourceIgnored ( self ) : <EOL> """<STR_LIT>""" <EOL> xml = """<STR_LIT>""" <EOL> self . service . allowAnySender = True <EOL> def setReceived ( request ) : <EOL> if request . sender == JID ( '<STR_LIT>' ) : <EOL> raise xmppim . RosterPushIgnored ( ) <EOL> def cb ( result ) : <EOL> self . assertEquals ( '<STR_LIT>' , result . condition ) <EOL> self . service . setReceived = setReceived <EOL> d = self . handleRequest ( xml ) <EOL> self . assertFailure ( d , error . StanzaError ) <EOL> d . addCallback ( cb ) <EOL> return d <EOL> def test_removeReceived ( self ) : <EOL> """<STR_LIT>""" <EOL> xml = """<STR_LIT>""" <EOL> requests = [ ] <EOL> def removeReceived ( request ) : <EOL> requests . append ( request ) <EOL> def cb ( result ) : <EOL> self . assertEqual ( <NUM_LIT:1> , len ( requests ) , "<STR_LIT>" ) <EOL> self . assertEqual ( JID ( '<STR_LIT>' ) , requests [ <NUM_LIT:0> ] . item . entity ) <EOL> self . service . removeReceived = removeReceived <EOL> d = self . handleRequest ( xml ) <EOL> d . addCallback ( cb ) <EOL> return d </s>
