<s> """<STR_LIT>""" <EOL> from zope . interface import implements <EOL> from zope . interface . verify import verifyObject <EOL> from twisted . internet import defer <EOL> from twisted . trial import unittest <EOL> from twisted . words . protocols . jabber import jid <EOL> from twisted . words . protocols . jabber . error import StanzaError <EOL> from wokkel import iwokkel , pubsub <EOL> from idavoll import backend , error , iidavoll <EOL> OWNER = jid . JID ( '<STR_LIT>' ) <EOL> OWNER_FULL = jid . JID ( '<STR_LIT>' ) <EOL> SERVICE = jid . JID ( '<STR_LIT>' ) <EOL> NS_PUBSUB = '<STR_LIT>' <EOL> class BackendTest ( unittest . TestCase ) : <EOL> def test_interfaceIBackend ( self ) : <EOL> self . assertTrue ( verifyObject ( iidavoll . IBackendService , <EOL> backend . BackendService ( None ) ) ) <EOL> def test_deleteNode ( self ) : <EOL> class TestNode : <EOL> nodeIdentifier = '<STR_LIT>' <EOL> def getAffiliation ( self , entity ) : <EOL> if entity . userhostJID ( ) == OWNER : <EOL> return defer . succeed ( '<STR_LIT>' ) <EOL> class TestStorage : <EOL> def __init__ ( self ) : <EOL> self . deleteCalled = [ ] <EOL> def getNode ( self , nodeIdentifier ) : <EOL> return defer . succeed ( TestNode ( ) ) <EOL> def deleteNode ( self , nodeIdentifier ) : <EOL> if nodeIdentifier in [ '<STR_LIT>' ] : <EOL> self . deleteCalled . append ( nodeIdentifier ) <EOL> return defer . succeed ( None ) <EOL> else : <EOL> return defer . fail ( error . NodeNotFound ( ) ) <EOL> def preDelete ( data ) : <EOL> self . assertFalse ( self . storage . deleteCalled ) <EOL> preDeleteCalled . append ( data ) <EOL> return defer . succeed ( None ) <EOL> def cb ( result ) : <EOL> self . assertEquals ( <NUM_LIT:1> , len ( preDeleteCalled ) ) <EOL> data = preDeleteCalled [ - <NUM_LIT:1> ] <EOL> self . assertEquals ( '<STR_LIT>' , data [ '<STR_LIT>' ] ) <EOL> self . assertTrue ( self . storage . deleteCalled ) <EOL> self . storage = TestStorage ( ) <EOL> self . backend = backend . BackendService ( self . storage ) <EOL> preDeleteCalled = [ ] <EOL> self . backend . registerPreDelete ( preDelete ) <EOL> d = self . backend . deleteNode ( '<STR_LIT>' , OWNER_FULL ) <EOL> d . addCallback ( cb ) <EOL> return d <EOL> def test_deleteNodeRedirect ( self ) : <EOL> uri = '<STR_LIT>' % ( SERVICE . full ( ) , ) <EOL> class TestNode : <EOL> nodeIdentifier = '<STR_LIT>' <EOL> def getAffiliation ( self , entity ) : <EOL> if entity . userhostJID ( ) == OWNER : <EOL> return defer . succeed ( '<STR_LIT>' ) <EOL> class TestStorage : <EOL> def __init__ ( self ) : <EOL> self . deleteCalled = [ ] <EOL> def getNode ( self , nodeIdentifier ) : <EOL> return defer . succeed ( TestNode ( ) ) <EOL> def deleteNode ( self , nodeIdentifier ) : <EOL> if nodeIdentifier in [ '<STR_LIT>' ] : <EOL> self . deleteCalled . append ( nodeIdentifier ) <EOL> return defer . succeed ( None ) <EOL> else : <EOL> return defer . fail ( error . NodeNotFound ( ) ) <EOL> def preDelete ( data ) : <EOL> self . assertFalse ( self . storage . deleteCalled ) <EOL> preDeleteCalled . append ( data ) <EOL> return defer . succeed ( None ) <EOL> def cb ( result ) : <EOL> self . assertEquals ( <NUM_LIT:1> , len ( preDeleteCalled ) ) <EOL> data = preDeleteCalled [ - <NUM_LIT:1> ] <EOL> self . assertEquals ( '<STR_LIT>' , data [ '<STR_LIT>' ] ) <EOL> self . assertEquals ( uri , data [ '<STR_LIT>' ] ) <EOL> self . assertTrue ( self . storage . deleteCalled ) <EOL> self . storage = TestStorage ( ) <EOL> self . backend = backend . BackendService ( self . storage ) <EOL> preDeleteCalled = [ ] <EOL> self . backend . registerPreDelete ( preDelete ) <EOL> d = self . backend . deleteNode ( '<STR_LIT>' , OWNER , redirectURI = uri ) <EOL> d . addCallback ( cb ) <EOL> return d <EOL> def test_createNodeNoID ( self ) : <EOL> """<STR_LIT>""" <EOL> class TestStorage : <EOL> def getDefaultConfiguration ( self , nodeType ) : <EOL> return { } <EOL> def createNode ( self , nodeIdentifier , requestor , config ) : <EOL> self . nodeIdentifier = nodeIdentifier <EOL> return defer . succeed ( None ) <EOL> self . storage = TestStorage ( ) <EOL> self . backend = backend . BackendService ( self . storage ) <EOL> self . storage . backend = self . backend <EOL> def checkID ( nodeIdentifier ) : <EOL> self . assertNotIdentical ( None , nodeIdentifier ) <EOL> self . assertIdentical ( self . storage . nodeIdentifier , nodeIdentifier ) <EOL> d = self . backend . createNode ( None , OWNER_FULL ) <EOL> d . addCallback ( checkID ) <EOL> return d <EOL> class NodeStore : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , nodes ) : <EOL> self . nodes = nodes <EOL> def getNode ( self , nodeIdentifier ) : <EOL> try : <EOL> return defer . succeed ( self . nodes [ nodeIdentifier ] ) <EOL> except KeyError : <EOL> return defer . fail ( error . NodeNotFound ( ) ) <EOL> def test_getNotifications ( self ) : <EOL> """<STR_LIT>""" <EOL> item = pubsub . Item ( ) <EOL> sub = pubsub . Subscription ( '<STR_LIT:test>' , OWNER , '<STR_LIT>' ) <EOL> class TestNode : <EOL> def getSubscriptions ( self , state = None ) : <EOL> return [ sub ] <EOL> def cb ( result ) : <EOL> self . assertEquals ( <NUM_LIT:1> , len ( result ) ) <EOL> subscriber , subscriptions , items = result [ - <NUM_LIT:1> ] <EOL> self . assertEquals ( OWNER , subscriber ) <EOL> self . assertEquals ( set ( [ sub ] ) , subscriptions ) <EOL> self . assertEquals ( [ item ] , items ) <EOL> self . storage = self . NodeStore ( { '<STR_LIT:test>' : TestNode ( ) } ) <EOL> self . backend = backend . BackendService ( self . storage ) <EOL> d = self . backend . getNotifications ( '<STR_LIT:test>' , [ item ] ) <EOL> d . addCallback ( cb ) <EOL> return d <EOL> def test_getNotificationsRoot ( self ) : <EOL> """<STR_LIT>""" <EOL> item = pubsub . Item ( ) <EOL> subRoot = pubsub . Subscription ( '<STR_LIT>' , OWNER , '<STR_LIT>' ) <EOL> class TestNode : <EOL> def getSubscriptions ( self , state = None ) : <EOL> return [ ] <EOL> class TestRootNode : <EOL> def getSubscriptions ( self , state = None ) : <EOL> return [ subRoot ] <EOL> def cb ( result ) : <EOL> self . assertEquals ( <NUM_LIT:1> , len ( result ) ) <EOL> subscriber , subscriptions , items = result [ - <NUM_LIT:1> ] <EOL> self . assertEquals ( OWNER , subscriber ) <EOL> self . assertEquals ( set ( [ subRoot ] ) , subscriptions ) <EOL> self . assertEquals ( [ item ] , items ) <EOL> self . storage = self . NodeStore ( { '<STR_LIT:test>' : TestNode ( ) , <EOL> '<STR_LIT>' : TestRootNode ( ) } ) <EOL> self . backend = backend . BackendService ( self . storage ) <EOL> d = self . backend . getNotifications ( '<STR_LIT:test>' , [ item ] ) <EOL> d . addCallback ( cb ) <EOL> return d <EOL> def test_getNotificationsMultipleNodes ( self ) : <EOL> """<STR_LIT>""" <EOL> item = pubsub . Item ( ) <EOL> sub = pubsub . Subscription ( '<STR_LIT:test>' , OWNER , '<STR_LIT>' ) <EOL> subRoot = pubsub . Subscription ( '<STR_LIT>' , OWNER , '<STR_LIT>' ) <EOL> class TestNode : <EOL> def getSubscriptions ( self , state = None ) : <EOL> return [ sub ] <EOL> class TestRootNode : <EOL> def getSubscriptions ( self , state = None ) : <EOL> return [ subRoot ] <EOL> def cb ( result ) : <EOL> self . assertEquals ( <NUM_LIT:1> , len ( result ) ) <EOL> subscriber , subscriptions , items = result [ - <NUM_LIT:1> ] <EOL> self . assertEquals ( OWNER , subscriber ) <EOL> self . assertEquals ( set ( [ sub , subRoot ] ) , subscriptions ) <EOL> self . assertEquals ( [ item ] , items ) <EOL> self . storage = self . NodeStore ( { '<STR_LIT:test>' : TestNode ( ) , <EOL> '<STR_LIT>' : TestRootNode ( ) } ) <EOL> self . backend = backend . BackendService ( self . storage ) <EOL> d = self . backend . getNotifications ( '<STR_LIT:test>' , [ item ] ) <EOL> d . addCallback ( cb ) <EOL> return d <EOL> def test_getDefaultConfiguration ( self ) : <EOL> """<STR_LIT>""" <EOL> class TestStorage : <EOL> def getDefaultConfiguration ( self , nodeType ) : <EOL> return { <EOL> "<STR_LIT>" : True , <EOL> "<STR_LIT>" : True } <EOL> def cb ( options ) : <EOL> self . assertIn ( "<STR_LIT>" , options ) <EOL> self . assertEqual ( True , options [ "<STR_LIT>" ] ) <EOL> self . backend = backend . BackendService ( TestStorage ( ) ) <EOL> d = self . backend . getDefaultConfiguration ( '<STR_LIT>' ) <EOL> d . addCallback ( cb ) <EOL> return d <EOL> def test_getNodeConfiguration ( self ) : <EOL> class testNode : <EOL> nodeIdentifier = '<STR_LIT>' <EOL> def getConfiguration ( self ) : <EOL> return { '<STR_LIT>' : True , <EOL> '<STR_LIT>' : False } <EOL> class testStorage : <EOL> def getNode ( self , nodeIdentifier ) : <EOL> return defer . succeed ( testNode ( ) ) <EOL> def cb ( options ) : <EOL> self . assertIn ( "<STR_LIT>" , options ) <EOL> self . assertEqual ( True , options [ "<STR_LIT>" ] ) <EOL> self . assertIn ( "<STR_LIT>" , options ) <EOL> self . assertEqual ( False , options [ "<STR_LIT>" ] ) <EOL> self . storage = testStorage ( ) <EOL> self . backend = backend . BackendService ( self . storage ) <EOL> self . storage . backend = self . backend <EOL> d = self . backend . getNodeConfiguration ( '<STR_LIT>' ) <EOL> d . addCallback ( cb ) <EOL> return d <EOL> def test_setNodeConfiguration ( self ) : <EOL> class testNode : <EOL> nodeIdentifier = '<STR_LIT>' <EOL> def getAffiliation ( self , entity ) : <EOL> if entity . userhostJID ( ) == OWNER : <EOL> return defer . succeed ( '<STR_LIT>' ) <EOL> def setConfiguration ( self , options ) : <EOL> self . options = options <EOL> class testStorage : <EOL> def __init__ ( self ) : <EOL> self . nodes = { '<STR_LIT>' : testNode ( ) } <EOL> def getNode ( self , nodeIdentifier ) : <EOL> return defer . succeed ( self . nodes [ nodeIdentifier ] ) <EOL> def checkOptions ( node ) : <EOL> options = node . options <EOL> self . assertIn ( "<STR_LIT>" , options ) <EOL> self . assertEqual ( True , options [ "<STR_LIT>" ] ) <EOL> self . assertIn ( "<STR_LIT>" , options ) <EOL> self . assertEqual ( False , options [ "<STR_LIT>" ] ) <EOL> def cb ( result ) : <EOL> d = self . storage . getNode ( '<STR_LIT>' ) <EOL> d . addCallback ( checkOptions ) <EOL> return d <EOL> self . storage = testStorage ( ) <EOL> self . backend = backend . BackendService ( self . storage ) <EOL> self . storage . backend = self . backend <EOL> options = { '<STR_LIT>' : True , <EOL> '<STR_LIT>' : False } <EOL> d = self . backend . setNodeConfiguration ( '<STR_LIT>' , options , OWNER_FULL ) <EOL> d . addCallback ( cb ) <EOL> return d <EOL> def test_publishNoID ( self ) : <EOL> """<STR_LIT>""" <EOL> class TestNode : <EOL> nodeType = '<STR_LIT>' <EOL> nodeIdentifier = '<STR_LIT>' <EOL> def getAffiliation ( self , entity ) : <EOL> if entity . userhostJID ( ) == OWNER : <EOL> return defer . succeed ( '<STR_LIT>' ) <EOL> def getConfiguration ( self ) : <EOL> return { '<STR_LIT>' : True , <EOL> '<STR_LIT>' : False } <EOL> class TestStorage : <EOL> def getNode ( self , nodeIdentifier ) : <EOL> return defer . succeed ( TestNode ( ) ) <EOL> def checkID ( notification ) : <EOL> self . assertNotIdentical ( None , notification [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT:id>' ] ) <EOL> self . storage = TestStorage ( ) <EOL> self . backend = backend . BackendService ( self . storage ) <EOL> self . storage . backend = self . backend <EOL> self . backend . registerNotifier ( checkID ) <EOL> items = [ pubsub . Item ( ) ] <EOL> d = self . backend . publish ( '<STR_LIT>' , items , OWNER_FULL ) <EOL> return d <EOL> def test_notifyOnSubscription ( self ) : <EOL> """<STR_LIT>""" <EOL> ITEM = "<STR_LIT>" % NS_PUBSUB <EOL> class TestNode : <EOL> implements ( iidavoll . ILeafNode ) <EOL> nodeIdentifier = '<STR_LIT>' <EOL> nodeType = '<STR_LIT>' <EOL> def getAffiliation ( self , entity ) : <EOL> if entity is OWNER : <EOL> return defer . succeed ( '<STR_LIT>' ) <EOL> def getConfiguration ( self ) : <EOL> return { '<STR_LIT>' : True , <EOL> '<STR_LIT>' : False , <EOL> '<STR_LIT>' : '<STR_LIT>' } <EOL> def getItems ( self , maxItems ) : <EOL> return [ ITEM ] <EOL> def addSubscription ( self , subscriber , state , options ) : <EOL> self . subscription = pubsub . Subscription ( '<STR_LIT>' , subscriber , <EOL> state , options ) <EOL> return defer . succeed ( None ) <EOL> def getSubscription ( self , subscriber ) : <EOL> return defer . succeed ( self . subscription ) <EOL> class TestStorage : <EOL> def getNode ( self , nodeIdentifier ) : <EOL> return defer . succeed ( TestNode ( ) ) <EOL> def cb ( data ) : <EOL> self . assertEquals ( '<STR_LIT>' , data [ '<STR_LIT>' ] ) <EOL> self . assertEquals ( [ ITEM ] , data [ '<STR_LIT>' ] ) <EOL> self . assertEquals ( OWNER , data [ '<STR_LIT>' ] . subscriber ) <EOL> self . storage = TestStorage ( ) <EOL> self . backend = backend . BackendService ( self . storage ) <EOL> self . storage . backend = self . backend <EOL> d1 = defer . Deferred ( ) <EOL> d1 . addCallback ( cb ) <EOL> self . backend . registerNotifier ( d1 . callback ) <EOL> d2 = self . backend . subscribe ( '<STR_LIT>' , OWNER , OWNER_FULL ) <EOL> return defer . gatherResults ( [ d1 , d2 ] ) <EOL> test_notifyOnSubscription . timeout = <NUM_LIT:2> <EOL> class BaseTestBackend ( object ) : <EOL> """<STR_LIT>""" <EOL> def supportsPublisherAffiliation ( self ) : <EOL> return True <EOL> def supportsOutcastAffiliation ( self ) : <EOL> return True <EOL> def supportsPersistentItems ( self ) : <EOL> return True <EOL> def supportsInstantNodes ( self ) : <EOL> return True <EOL> def registerNotifier ( self , observerfn , * args , ** kwargs ) : <EOL> return <EOL> def registerPreDelete ( self , preDeleteFn ) : <EOL> return <EOL> class PubSubResourceFromBackendTest ( unittest . TestCase ) : <EOL> def test_interface ( self ) : <EOL> resource = backend . PubSubResourceFromBackend ( BaseTestBackend ( ) ) <EOL> self . assertTrue ( verifyObject ( iwokkel . IPubSubResource , resource ) ) <EOL> def test_preDelete ( self ) : <EOL> """<STR_LIT>""" <EOL> class TestBackend ( BaseTestBackend ) : <EOL> preDeleteFn = None <EOL> def registerPreDelete ( self , preDeleteFn ) : <EOL> self . preDeleteFn = preDeleteFn <EOL> def getSubscribers ( self , nodeIdentifier ) : <EOL> return defer . succeed ( [ OWNER ] ) <EOL> def notifyDelete ( service , nodeIdentifier , subscribers , <EOL> redirectURI = None ) : <EOL> self . assertEqual ( SERVICE , service ) <EOL> self . assertEqual ( '<STR_LIT:test>' , nodeIdentifier ) <EOL> self . assertEqual ( [ OWNER ] , subscribers ) <EOL> self . assertIdentical ( None , redirectURI ) <EOL> d1 . callback ( None ) <EOL> d1 = defer . Deferred ( ) <EOL> resource = backend . PubSubResourceFromBackend ( TestBackend ( ) ) <EOL> resource . serviceJID = SERVICE <EOL> resource . pubsubService = pubsub . PubSubService ( ) <EOL> resource . pubsubService . notifyDelete = notifyDelete <EOL> self . assertTrue ( verifyObject ( iwokkel . IPubSubResource , resource ) ) <EOL> self . assertNotIdentical ( None , resource . backend . preDeleteFn ) <EOL> data = { '<STR_LIT>' : '<STR_LIT:test>' } <EOL> d2 = resource . backend . preDeleteFn ( data ) <EOL> return defer . DeferredList ( [ d1 , d2 ] , fireOnOneErrback = <NUM_LIT:1> ) <EOL> def test_preDeleteRedirect ( self ) : <EOL> """<STR_LIT>""" <EOL> uri = '<STR_LIT>' % ( SERVICE . full ( ) , ) <EOL> class TestBackend ( BaseTestBackend ) : <EOL> preDeleteFn = None <EOL> def registerPreDelete ( self , preDeleteFn ) : <EOL> self . preDeleteFn = preDeleteFn <EOL> def getSubscribers ( self , nodeIdentifier ) : <EOL> return defer . succeed ( [ OWNER ] ) <EOL> def notifyDelete ( service , nodeIdentifier , subscribers , <EOL> redirectURI = None ) : <EOL> self . assertEqual ( SERVICE , service ) <EOL> self . assertEqual ( '<STR_LIT:test>' , nodeIdentifier ) <EOL> self . assertEqual ( [ OWNER ] , subscribers ) <EOL> self . assertEqual ( uri , redirectURI ) <EOL> d1 . callback ( None ) <EOL> d1 = defer . Deferred ( ) <EOL> resource = backend . PubSubResourceFromBackend ( TestBackend ( ) ) <EOL> resource . serviceJID = SERVICE <EOL> resource . pubsubService = pubsub . PubSubService ( ) <EOL> resource . pubsubService . notifyDelete = notifyDelete <EOL> self . assertTrue ( verifyObject ( iwokkel . IPubSubResource , resource ) ) <EOL> self . assertNotIdentical ( None , resource . backend . preDeleteFn ) <EOL> data = { '<STR_LIT>' : '<STR_LIT:test>' , <EOL> '<STR_LIT>' : uri } <EOL> d2 = resource . backend . preDeleteFn ( data ) <EOL> return defer . DeferredList ( [ d1 , d2 ] , fireOnOneErrback = <NUM_LIT:1> ) <EOL> def test_unsubscribeNotSubscribed ( self ) : <EOL> """<STR_LIT>""" <EOL> class TestBackend ( BaseTestBackend ) : <EOL> def unsubscribe ( self , nodeIdentifier , subscriber , requestor ) : <EOL> return defer . fail ( error . NotSubscribed ( ) ) <EOL> def cb ( e ) : <EOL> self . assertEquals ( '<STR_LIT>' , e . condition ) <EOL> resource = backend . PubSubResourceFromBackend ( TestBackend ( ) ) <EOL> request = pubsub . PubSubRequest ( ) <EOL> request . sender = OWNER <EOL> request . recipient = SERVICE <EOL> request . nodeIdentifier = '<STR_LIT:test>' <EOL> request . subscriber = OWNER <EOL> d = resource . unsubscribe ( request ) <EOL> self . assertFailure ( d , StanzaError ) <EOL> d . addCallback ( cb ) <EOL> return d <EOL> def test_getInfo ( self ) : <EOL> """<STR_LIT>""" <EOL> class TestBackend ( BaseTestBackend ) : <EOL> def getNodeType ( self , nodeIdentifier ) : <EOL> return defer . succeed ( '<STR_LIT>' ) <EOL> def getNodeMetaData ( self , nodeIdentifier ) : <EOL> return defer . succeed ( { '<STR_LIT>' : True } ) <EOL> def cb ( info ) : <EOL> self . assertIn ( '<STR_LIT:type>' , info ) <EOL> self . assertEquals ( '<STR_LIT>' , info [ '<STR_LIT:type>' ] ) <EOL> self . assertIn ( '<STR_LIT>' , info ) <EOL> self . assertEquals ( { '<STR_LIT>' : True } , info [ '<STR_LIT>' ] ) <EOL> resource = backend . PubSubResourceFromBackend ( TestBackend ( ) ) <EOL> d = resource . getInfo ( OWNER , SERVICE , '<STR_LIT:test>' ) <EOL> d . addCallback ( cb ) <EOL> return d <EOL> def test_getConfigurationOptions ( self ) : <EOL> class TestBackend ( BaseTestBackend ) : <EOL> nodeOptions = { <EOL> "<STR_LIT>" : <EOL> { "<STR_LIT:type>" : "<STR_LIT>" , <EOL> "<STR_LIT:label>" : "<STR_LIT>" } , <EOL> "<STR_LIT>" : <EOL> { "<STR_LIT:type>" : "<STR_LIT>" , <EOL> "<STR_LIT:label>" : "<STR_LIT>" } <EOL> } <EOL> resource = backend . PubSubResourceFromBackend ( TestBackend ( ) ) <EOL> options = resource . getConfigurationOptions ( ) <EOL> self . assertIn ( "<STR_LIT>" , options ) <EOL> def test_default ( self ) : <EOL> class TestBackend ( BaseTestBackend ) : <EOL> def getDefaultConfiguration ( self , nodeType ) : <EOL> options = { "<STR_LIT>" : True , <EOL> "<STR_LIT>" : True , <EOL> "<STR_LIT>" : '<STR_LIT>' , <EOL> } <EOL> return defer . succeed ( options ) <EOL> def cb ( options ) : <EOL> self . assertEquals ( True , options [ "<STR_LIT>" ] ) <EOL> resource = backend . PubSubResourceFromBackend ( TestBackend ( ) ) <EOL> request = pubsub . PubSubRequest ( ) <EOL> request . sender = OWNER <EOL> request . recipient = SERVICE <EOL> request . nodeType = '<STR_LIT>' <EOL> d = resource . default ( request ) <EOL> d . addCallback ( cb ) <EOL> return d </s>
