← Back to branch summary

~ntt-pf-lab/nova/monkey_patch_notification

« back to all changes in this revision

Viewing changes to vendor/Twisted-10.0.0/twisted/words/test/test_xmlstream.py

  • Committer: Jesse Andrews
  • Date: 2010-05-28 06:05:26 UTC
  • Revision ID: git-v1:bf6e6e718cdc7488e2da87b21e258ccc065fe499
initial commit

Show diffs side-by-side

added added

removed removed

Lines of Context:
vendor/Twisted-10.0.0/twisted/words/test/test_xmlstream.py
 
1
# Copyright (c) 2001-2008 Twisted Matrix Laboratories.
 
2
# See LICENSE for details.
 
3
 
 
4
"""
 
5
Tests for L{twisted.words.xish.xmlstream}.
 
6
"""
 
7
 
 
8
from twisted.internet import protocol
 
9
from twisted.trial import unittest
 
10
from twisted.words.xish import domish, utility, xmlstream
 
11
 
 
12
class XmlStreamTest(unittest.TestCase):
 
13
    def setUp(self):
 
14
        self.outlist = []
 
15
        self.xmlstream = xmlstream.XmlStream()
 
16
        self.xmlstream.transport = self
 
17
        self.xmlstream.transport.write = self.outlist.append
 
18
 
 
19
 
 
20
    def loseConnection(self):
 
21
        """
 
22
        Stub loseConnection because we are a transport.
 
23
        """
 
24
        self.xmlstream.connectionLost("no reason")
 
25
 
 
26
 
 
27
    def test_send(self):
 
28
        """
 
29
        Sending data should result into it being written to the transport.
 
30
        """
 
31
        self.xmlstream.connectionMade()
 
32
        self.xmlstream.send("<root>")
 
33
        self.assertEquals(self.outlist[0], "<root>")
 
34
 
 
35
 
 
36
    def test_receiveRoot(self):
 
37
        """
 
38
        Receiving the starttag of the root element results in stream start.
 
39
        """
 
40
        streamStarted = []
 
41
 
 
42
        def streamStartEvent(rootelem):
 
43
            streamStarted.append(None)
 
44
 
 
45
        self.xmlstream.addObserver(xmlstream.STREAM_START_EVENT,
 
46
                                   streamStartEvent)
 
47
        self.xmlstream.connectionMade()
 
48
        self.xmlstream.dataReceived("<root>")
 
49
        self.assertEquals(1, len(streamStarted))
 
50
 
 
51
 
 
52
    def test_receiveBadXML(self):
 
53
        """
 
54
        Receiving malformed XML should result in in error.
 
55
        """
 
56
        streamError = []
 
57
        streamEnd = []
 
58
 
 
59
        def streamErrorEvent(reason):
 
60
            streamError.append(reason)
 
61
 
 
62
        def streamEndEvent(_):
 
63
            streamEnd.append(None)
 
64
 
 
65
        self.xmlstream.addObserver(xmlstream.STREAM_ERROR_EVENT,
 
66
                                   streamErrorEvent)
 
67
        self.xmlstream.addObserver(xmlstream.STREAM_END_EVENT,
 
68
                                   streamEndEvent)
 
69
        self.xmlstream.connectionMade()
 
70
 
 
71
        self.xmlstream.dataReceived("<root>")
 
72
        self.assertEquals(0, len(streamError))
 
73
        self.assertEquals(0, len(streamEnd))
 
74
 
 
75
        self.xmlstream.dataReceived("<child><unclosed></child>")
 
76
        self.assertEquals(1, len(streamError))
 
77
        self.assertTrue(streamError[0].check(domish.ParserError))
 
78
        self.assertEquals(1, len(streamEnd))
 
79
 
 
80
 
 
81
 
 
82
class DummyProtocol(protocol.Protocol, utility.EventDispatcher):
 
83
    """
 
84
    I am a protocol with an event dispatcher without further processing.
 
85
 
 
86
    This protocol is only used for testing XmlStreamFactoryMixin to make
 
87
    sure the bootstrap observers are added to the protocol instance.
 
88
    """
 
89
 
 
90
    def __init__(self, *args, **kwargs):
 
91
        self.args = args
 
92
        self.kwargs = kwargs
 
93
        self.observers = []
 
94
 
 
95
        utility.EventDispatcher.__init__(self)
 
96
 
 
97
 
 
98
 
 
99
class BootstrapMixinTest(unittest.TestCase):
 
100
    """
 
101
    Tests for L{xmlstream.BootstrapMixin}.
 
102
 
 
103
    @ivar factory: Instance of the factory or mixin under test.
 
104
    """
 
105
 
 
106
    def setUp(self):
 
107
        self.factory = xmlstream.BootstrapMixin()
 
108
 
 
109
 
 
110
    def test_installBootstraps(self):
 
111
        """
 
112
        Dispatching an event should fire registered bootstrap observers.
 
113
        """
 
114
        called = []
 
115
 
 
116
        def cb(data):
 
117
            called.append(data)
 
118
 
 
119
        dispatcher = DummyProtocol()
 
120
        self.factory.addBootstrap('//event/myevent', cb)
 
121
        self.factory.installBootstraps(dispatcher)
 
122
 
 
123
        dispatcher.dispatch(None, '//event/myevent')
 
124
        self.assertEquals(1, len(called))
 
125
 
 
126
 
 
127
    def test_addAndRemoveBootstrap(self):
 
128
        """
 
129
        Test addition and removal of a bootstrap event handler.
 
130
        """
 
131
 
 
132
        called = []
 
133
 
 
134
        def cb(data):
 
135
            called.append(data)
 
136
 
 
137
        self.factory.addBootstrap('//event/myevent', cb)
 
138
        self.factory.removeBootstrap('//event/myevent', cb)
 
139
 
 
140
        dispatcher = DummyProtocol()
 
141
        self.factory.installBootstraps(dispatcher)
 
142
 
 
143
        dispatcher.dispatch(None, '//event/myevent')
 
144
        self.assertFalse(called)
 
145
 
 
146
 
 
147
 
 
148
class GenericXmlStreamFactoryTestsMixin(BootstrapMixinTest):
 
149
    """
 
150
    Generic tests for L{XmlStream} factories.
 
151
    """
 
152
 
 
153
    def setUp(self):
 
154
        self.factory = xmlstream.XmlStreamFactory()
 
155
 
 
156
 
 
157
    def test_buildProtocolInstallsBootstraps(self):
 
158
        """
 
159
        The protocol factory installs bootstrap event handlers on the protocol.
 
160
        """
 
161
        called = []
 
162
 
 
163
        def cb(data):
 
164
            called.append(data)
 
165
 
 
166
        self.factory.addBootstrap('//event/myevent', cb)
 
167
 
 
168
        xs = self.factory.buildProtocol(None)
 
169
        xs.dispatch(None, '//event/myevent')
 
170
 
 
171
        self.assertEquals(1, len(called))
 
172
 
 
173
 
 
174
    def test_buildProtocolStoresFactory(self):
 
175
        """
 
176
        The protocol factory is saved in the protocol.
 
177
        """
 
178
        xs = self.factory.buildProtocol(None)
 
179
        self.assertIdentical(self.factory, xs.factory)
 
180
 
 
181
 
 
182
 
 
183
class XmlStreamFactoryMixinTest(GenericXmlStreamFactoryTestsMixin):
 
184
    """
 
185
    Tests for L{xmlstream.XmlStreamFactoryMixin}.
 
186
    """
 
187
 
 
188
    def setUp(self):
 
189
        self.factory = xmlstream.XmlStreamFactoryMixin(None, test=None)
 
190
        self.factory.protocol = DummyProtocol
 
191
 
 
192
 
 
193
    def test_buildProtocolFactoryArguments(self):
 
194
        """
 
195
        Arguments passed to the factory should be passed to protocol on
 
196
        instantiation.
 
197
        """
 
198
        xs = self.factory.buildProtocol(None)
 
199
 
 
200
        self.assertEquals((None,), xs.args)
 
201
        self.assertEquals({'test': None}, xs.kwargs)