← Back to branch summary

~justin-fathomdb/nova/justinsb-openstack-api-volumes

« back to all changes in this revision

Viewing changes to vendor/Twisted-10.0.0/twisted/pair/test/test_rawudp.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/pair/test/test_rawudp.py
 
1
# Copyright (c) 2001-2004 Twisted Matrix Laboratories.
 
2
# See LICENSE for details.
 
3
 
 
4
#
 
5
from twisted.trial import unittest
 
6
 
 
7
from twisted.internet import protocol, reactor, error
 
8
from twisted.python import failure
 
9
from twisted.pair import rawudp
 
10
 
 
11
class MyProtocol(protocol.DatagramProtocol):
 
12
    def __init__(self, expecting):
 
13
        self.expecting = list(expecting)
 
14
 
 
15
    def datagramReceived(self, data, (host, port)):
 
16
        assert self.expecting, 'Got a packet when not expecting anymore.'
 
17
        expectData, expectHost, expectPort = self.expecting.pop(0)
 
18
 
 
19
        assert expectData == data, "Expected data %r, got %r" % (expectData, data)
 
20
        assert expectHost == host, "Expected host %r, got %r" % (expectHost, host)
 
21
        assert expectPort == port, "Expected port %d=0x%04x, got %d=0x%04x" % (expectPort, expectPort, port, port)
 
22
 
 
23
class RawUDPTestCase(unittest.TestCase):
 
24
    def testPacketParsing(self):
 
25
        proto = rawudp.RawUDPProtocol()
 
26
        p1 = MyProtocol([
 
27
 
 
28
            ('foobar', 'testHost', 0x43A2),
 
29
 
 
30
            ])
 
31
        proto.addProto(0xF00F, p1)
 
32
 
 
33
        proto.datagramReceived("\x43\xA2" #source
 
34
                               + "\xf0\x0f" #dest
 
35
                               + "\x00\x06" #len
 
36
                               + "\xDE\xAD" #check
 
37
                               + "foobar",
 
38
                               partial=0,
 
39
                               dest='dummy',
 
40
                               source='testHost',
 
41
                               protocol='dummy',
 
42
                               version='dummy',
 
43
                               ihl='dummy',
 
44
                               tos='dummy',
 
45
                               tot_len='dummy',
 
46
                               fragment_id='dummy',
 
47
                               fragment_offset='dummy',
 
48
                               dont_fragment='dummy',
 
49
                               more_fragments='dummy',
 
50
                               ttl='dummy',
 
51
                               )
 
52
 
 
53
        assert not p1.expecting, \
 
54
               'Should not expect any more packets, but still want %r' % p1.expecting
 
55
 
 
56
    def testMultiplePackets(self):
 
57
        proto = rawudp.RawUDPProtocol()
 
58
        p1 = MyProtocol([
 
59
 
 
60
            ('foobar', 'testHost', 0x43A2),
 
61
            ('quux', 'otherHost', 0x33FE),
 
62
 
 
63
            ])
 
64
        proto.addProto(0xF00F, p1)
 
65
        proto.datagramReceived("\x43\xA2" #source
 
66
                               + "\xf0\x0f" #dest
 
67
                               + "\x00\x06" #len
 
68
                               + "\xDE\xAD" #check
 
69
                               + "foobar",
 
70
                               partial=0,
 
71
                               dest='dummy',
 
72
                               source='testHost',
 
73
                               protocol='dummy',
 
74
                               version='dummy',
 
75
                               ihl='dummy',
 
76
                               tos='dummy',
 
77
                               tot_len='dummy',
 
78
                               fragment_id='dummy',
 
79
                               fragment_offset='dummy',
 
80
                               dont_fragment='dummy',
 
81
                               more_fragments='dummy',
 
82
                               ttl='dummy',
 
83
                               )
 
84
        proto.datagramReceived("\x33\xFE" #source
 
85
                               + "\xf0\x0f" #dest
 
86
                               + "\x00\x05" #len
 
87
                               + "\xDE\xAD" #check
 
88
                               + "quux",
 
89
                               partial=0,
 
90
                               dest='dummy',
 
91
                               source='otherHost',
 
92
                               protocol='dummy',
 
93
                               version='dummy',
 
94
                               ihl='dummy',
 
95
                               tos='dummy',
 
96
                               tot_len='dummy',
 
97
                               fragment_id='dummy',
 
98
                               fragment_offset='dummy',
 
99
                               dont_fragment='dummy',
 
100
                               more_fragments='dummy',
 
101
                               ttl='dummy',
 
102
                               )
 
103
 
 
104
        assert not p1.expecting, \
 
105
               'Should not expect any more packets, but still want %r' % p1.expecting
 
106
 
 
107
 
 
108
    def testMultipleSameProtos(self):
 
109
        proto = rawudp.RawUDPProtocol()
 
110
        p1 = MyProtocol([
 
111
 
 
112
            ('foobar', 'testHost', 0x43A2),
 
113
 
 
114
            ])
 
115
 
 
116
        p2 = MyProtocol([
 
117
 
 
118
            ('foobar', 'testHost', 0x43A2),
 
119
 
 
120
            ])
 
121
 
 
122
        proto.addProto(0xF00F, p1)
 
123
        proto.addProto(0xF00F, p2)
 
124
 
 
125
        proto.datagramReceived("\x43\xA2" #source
 
126
                               + "\xf0\x0f" #dest
 
127
                               + "\x00\x06" #len
 
128
                               + "\xDE\xAD" #check
 
129
                               + "foobar",
 
130
                               partial=0,
 
131
                               dest='dummy',
 
132
                               source='testHost',
 
133
                               protocol='dummy',
 
134
                               version='dummy',
 
135
                               ihl='dummy',
 
136
                               tos='dummy',
 
137
                               tot_len='dummy',
 
138
                               fragment_id='dummy',
 
139
                               fragment_offset='dummy',
 
140
                               dont_fragment='dummy',
 
141
                               more_fragments='dummy',
 
142
                               ttl='dummy',
 
143
                               )
 
144
 
 
145
        assert not p1.expecting, \
 
146
               'Should not expect any more packets, but still want %r' % p1.expecting
 
147
        assert not p2.expecting, \
 
148
               'Should not expect any more packets, but still want %r' % p2.expecting
 
149
 
 
150
    def testWrongProtoNotSeen(self):
 
151
        proto = rawudp.RawUDPProtocol()
 
152
        p1 = MyProtocol([])
 
153
        proto.addProto(1, p1)
 
154
 
 
155
        proto.datagramReceived("\x43\xA2" #source
 
156
                               + "\xf0\x0f" #dest
 
157
                               + "\x00\x06" #len
 
158
                               + "\xDE\xAD" #check
 
159
                               + "foobar",
 
160
                               partial=0,
 
161
                               dest='dummy',
 
162
                               source='testHost',
 
163
                               protocol='dummy',
 
164
                               version='dummy',
 
165
                               ihl='dummy',
 
166
                               tos='dummy',
 
167
                               tot_len='dummy',
 
168
                               fragment_id='dummy',
 
169
                               fragment_offset='dummy',
 
170
                               dont_fragment='dummy',
 
171
                               more_fragments='dummy',
 
172
                               ttl='dummy',
 
173
                               )
 
174
 
 
175
    def testDemuxing(self):
 
176
        proto = rawudp.RawUDPProtocol()
 
177
        p1 = MyProtocol([
 
178
 
 
179
            ('foobar', 'testHost', 0x43A2),
 
180
            ('quux', 'otherHost', 0x33FE),
 
181
 
 
182
            ])
 
183
        proto.addProto(0xF00F, p1)
 
184
 
 
185
        p2 = MyProtocol([
 
186
 
 
187
            ('quux', 'otherHost', 0xA401),
 
188
            ('foobar', 'testHost', 0xA302),
 
189
 
 
190
            ])
 
191
        proto.addProto(0xB050, p2)
 
192
 
 
193
        proto.datagramReceived("\xA4\x01" #source
 
194
                               + "\xB0\x50" #dest
 
195
                               + "\x00\x05" #len
 
196
                               + "\xDE\xAD" #check
 
197
                               + "quux",
 
198
                               partial=0,
 
199
                               dest='dummy',
 
200
                               source='otherHost',
 
201
                               protocol='dummy',
 
202
                               version='dummy',
 
203
                               ihl='dummy',
 
204
                               tos='dummy',
 
205
                               tot_len='dummy',
 
206
                               fragment_id='dummy',
 
207
                               fragment_offset='dummy',
 
208
                               dont_fragment='dummy',
 
209
                               more_fragments='dummy',
 
210
                               ttl='dummy',
 
211
                               )
 
212
        proto.datagramReceived("\x43\xA2" #source
 
213
                               + "\xf0\x0f" #dest
 
214
                               + "\x00\x06" #len
 
215
                               + "\xDE\xAD" #check
 
216
                               + "foobar",
 
217
                               partial=0,
 
218
                               dest='dummy',
 
219
                               source='testHost',
 
220
                               protocol='dummy',
 
221
                               version='dummy',
 
222
                               ihl='dummy',
 
223
                               tos='dummy',
 
224
                               tot_len='dummy',
 
225
                               fragment_id='dummy',
 
226
                               fragment_offset='dummy',
 
227
                               dont_fragment='dummy',
 
228
                               more_fragments='dummy',
 
229
                               ttl='dummy',
 
230
                               )
 
231
        proto.datagramReceived("\x33\xFE" #source
 
232
                               + "\xf0\x0f" #dest
 
233
                               + "\x00\x05" #len
 
234
                               + "\xDE\xAD" #check
 
235
                               + "quux",
 
236
                               partial=0,
 
237
                               dest='dummy',
 
238
                               source='otherHost',
 
239
                               protocol='dummy',
 
240
                               version='dummy',
 
241
                               ihl='dummy',
 
242
                               tos='dummy',
 
243
                               tot_len='dummy',
 
244
                               fragment_id='dummy',
 
245
                               fragment_offset='dummy',
 
246
                               dont_fragment='dummy',
 
247
                               more_fragments='dummy',
 
248
                               ttl='dummy',
 
249
                               )
 
250
        proto.datagramReceived("\xA3\x02" #source
 
251
                               + "\xB0\x50" #dest
 
252
                               + "\x00\x06" #len
 
253
                               + "\xDE\xAD" #check
 
254
                               + "foobar",
 
255
                               partial=0,
 
256
                               dest='dummy',
 
257
                               source='testHost',
 
258
                               protocol='dummy',
 
259
                               version='dummy',
 
260
                               ihl='dummy',
 
261
                               tos='dummy',
 
262
                               tot_len='dummy',
 
263
                               fragment_id='dummy',
 
264
                               fragment_offset='dummy',
 
265
                               dont_fragment='dummy',
 
266
                               more_fragments='dummy',
 
267
                               ttl='dummy',
 
268
                               )
 
269
 
 
270
        assert not p1.expecting, \
 
271
               'Should not expect any more packets, but still want %r' % p1.expecting
 
272
        assert not p2.expecting, \
 
273
               'Should not expect any more packets, but still want %r' % p2.expecting
 
274
 
 
275
    def testAddingBadProtos_WrongLevel(self):
 
276
        """Adding a wrong level protocol raises an exception."""
 
277
        e = rawudp.RawUDPProtocol()
 
278
        try:
 
279
            e.addProto(42, "silliness")
 
280
        except TypeError, e:
 
281
            if e.args == ('Added protocol must be an instance of DatagramProtocol',):
 
282
                pass
 
283
            else:
 
284
                raise
 
285
        else:
 
286
            raise AssertionError, 'addProto must raise an exception for bad protocols'
 
287
 
 
288
 
 
289
    def testAddingBadProtos_TooSmall(self):
 
290
        """Adding a protocol with a negative number raises an exception."""
 
291
        e = rawudp.RawUDPProtocol()
 
292
        try:
 
293
            e.addProto(-1, protocol.DatagramProtocol())
 
294
        except TypeError, e:
 
295
            if e.args == ('Added protocol must be positive or zero',):
 
296
                pass
 
297
            else:
 
298
                raise
 
299
        else:
 
300
            raise AssertionError, 'addProto must raise an exception for bad protocols'
 
301
 
 
302
 
 
303
    def testAddingBadProtos_TooBig(self):
 
304
        """Adding a protocol with a number >=2**16 raises an exception."""
 
305
        e = rawudp.RawUDPProtocol()
 
306
        try:
 
307
            e.addProto(2**16, protocol.DatagramProtocol())
 
308
        except TypeError, e:
 
309
            if e.args == ('Added protocol must fit in 16 bits',):
 
310
                pass
 
311
            else:
 
312
                raise
 
313
        else:
 
314
            raise AssertionError, 'addProto must raise an exception for bad protocols'
 
315
 
 
316
    def testAddingBadProtos_TooBig2(self):
 
317
        """Adding a protocol with a number >=2**16 raises an exception."""
 
318
        e = rawudp.RawUDPProtocol()
 
319
        try:
 
320
            e.addProto(2**16+1, protocol.DatagramProtocol())
 
321
        except TypeError, e:
 
322
            if e.args == ('Added protocol must fit in 16 bits',):
 
323
                pass
 
324
            else:
 
325
                raise
 
326
        else:
 
327
            raise AssertionError, 'addProto must raise an exception for bad protocols'