source: wokkel/test/test_compat.py @ 96:8e6130587088

Last change on this file since 96:8e6130587088 was 96:8e6130587088, checked in by Ralph Meijer <ralphm@…>, 10 years ago

Remove copyright dates from individual source files, only update LICENSE.

  • Property exe set to *
File size: 5.7 KB
Line 
1# Copyright (c) 2001-2008 Twisted Matrix Laboratories.
2# Copyright (c) Ralph Meijer.
3# See LICENSE for details.
4
5"""
6Tests for L{wokkel.compat}.
7"""
8
9from zope.interface import implements
10from zope.interface.verify import verifyObject
11from twisted.internet import protocol, task
12from twisted.internet.interfaces import IProtocolFactory, IReactorTime
13from twisted.trial import unittest
14from twisted.words.xish import utility
15from twisted.words.protocols.jabber import xmlstream
16from wokkel.compat import BootstrapMixin, IQ, XmlStreamServerFactory
17
18class DummyProtocol(protocol.Protocol, utility.EventDispatcher):
19    """
20    I am a protocol with an event dispatcher without further processing.
21
22    This protocol is only used for testing BootstrapMixin to make
23    sure the bootstrap observers are added to the protocol instance.
24    """
25
26    def __init__(self, *args, **kwargs):
27        self.args = args
28        self.kwargs = kwargs
29        self.observers = []
30
31        utility.EventDispatcher.__init__(self)
32
33
34
35class BootstrapMixinTest(unittest.TestCase):
36    """
37    Tests for L{BootstrapMixin}.
38
39    @ivar factory: Instance of the factory or mixin under test.
40    """
41
42    def setUp(self):
43        self.factory = BootstrapMixin()
44
45
46    def test_installBootstraps(self):
47        """
48        Dispatching an event should fire registered bootstrap observers.
49        """
50        called = []
51
52        def cb(data):
53            called.append(data)
54
55        dispatcher = DummyProtocol()
56        self.factory.addBootstrap('//event/myevent', cb)
57        self.factory.installBootstraps(dispatcher)
58
59        dispatcher.dispatch(None, '//event/myevent')
60        self.assertEquals(1, len(called))
61
62
63    def test_addAndRemoveBootstrap(self):
64        """
65        Test addition and removal of a bootstrap event handler.
66        """
67
68        called = []
69
70        def cb(data):
71            called.append(data)
72
73        self.factory.addBootstrap('//event/myevent', cb)
74        self.factory.removeBootstrap('//event/myevent', cb)
75
76        dispatcher = DummyProtocol()
77        self.factory.installBootstraps(dispatcher)
78
79        dispatcher.dispatch(None, '//event/myevent')
80        self.assertFalse(called)
81
82
83
84class XmlStreamServerFactoryTest(BootstrapMixinTest):
85    """
86    Tests for L{XmlStreamServerFactory}.
87    """
88
89    def setUp(self):
90        """
91        Set up a server factory with a authenticator factory function.
92        """
93        class TestAuthenticator(object):
94            def __init__(self):
95                self.xmlstreams = []
96
97            def associateWithStream(self, xs):
98                self.xmlstreams.append(xs)
99
100        def authenticatorFactory():
101            return TestAuthenticator()
102
103        self.factory = XmlStreamServerFactory(authenticatorFactory)
104
105
106    def test_interface(self):
107        """
108        L{XmlStreamServerFactory} is a L{Factory}.
109        """
110        verifyObject(IProtocolFactory, self.factory)
111
112
113    def test_buildProtocolAuthenticatorInstantiation(self):
114        """
115        The authenticator factory should be used to instantiate the
116        authenticator and pass it to the protocol.
117
118        The default protocol, L{XmlStream} stores the authenticator it is
119        passed, and calls its C{associateWithStream} method. so we use that to
120        check whether our authenticator factory is used and the protocol
121        instance gets an authenticator.
122        """
123        xs = self.factory.buildProtocol(None)
124        self.assertEquals([xs], xs.authenticator.xmlstreams)
125
126
127    def test_buildProtocolXmlStream(self):
128        """
129        The protocol factory creates Jabber XML Stream protocols by default.
130        """
131        xs = self.factory.buildProtocol(None)
132        self.assertIsInstance(xs, xmlstream.XmlStream)
133
134
135    def test_buildProtocolTwice(self):
136        """
137        Subsequent calls to buildProtocol should result in different instances
138        of the protocol, as well as their authenticators.
139        """
140        xs1 = self.factory.buildProtocol(None)
141        xs2 = self.factory.buildProtocol(None)
142        self.assertNotIdentical(xs1, xs2)
143        self.assertNotIdentical(xs1.authenticator, xs2.authenticator)
144
145
146    def test_buildProtocolInstallsBootstraps(self):
147        """
148        The protocol factory installs bootstrap event handlers on the protocol.
149        """
150        called = []
151
152        def cb(data):
153            called.append(data)
154
155        self.factory.addBootstrap('//event/myevent', cb)
156
157        xs = self.factory.buildProtocol(None)
158        xs.dispatch(None, '//event/myevent')
159
160        self.assertEquals(1, len(called))
161
162
163    def test_buildProtocolStoresFactory(self):
164        """
165        The protocol factory is saved in the protocol.
166        """
167        xs = self.factory.buildProtocol(None)
168        self.assertIdentical(self.factory, xs.factory)
169
170
171
172class FakeReactor(object):
173
174    implements(IReactorTime)
175    def __init__(self):
176        self.clock = task.Clock()
177        self.callLater = self.clock.callLater
178        self.getDelayedCalls = self.clock.getDelayedCalls
179
180
181
182class IQTest(unittest.TestCase):
183    """
184    Tests for L{IQ}.
185    """
186
187    def setUp(self):
188        self.reactor = FakeReactor()
189        self.clock = self.reactor.clock
190
191
192    def testRequestTimingOutEventDispatcher(self):
193        """
194        Test that an iq request with a defined timeout times out.
195        """
196        from twisted.words.xish import utility
197        output = []
198        xs = utility.EventDispatcher()
199        xs.send = output.append
200
201        self.iq = IQ(xs, reactor=self.reactor)
202        self.iq.timeout = 60
203        d = self.iq.send()
204        self.assertFailure(d, xmlstream.TimeoutError)
205
206        self.clock.pump([1, 60])
207        self.assertFalse(self.reactor.getDelayedCalls())
208        self.assertFalse(xs.iqDeferreds)
209        return d
Note: See TracBrowser for help on using the repository browser.