| OLD | NEW |
| (Empty) |
| 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, components | |
| 9 from twisted.pair import ethernet, raw | |
| 10 from zope.interface import implements | |
| 11 | |
| 12 class MyProtocol: | |
| 13 implements(raw.IRawPacketProtocol) | |
| 14 | |
| 15 def __init__(self, expecting): | |
| 16 self.expecting = list(expecting) | |
| 17 | |
| 18 def datagramReceived(self, data, **kw): | |
| 19 assert self.expecting, 'Got a packet when not expecting anymore.' | |
| 20 expect = self.expecting.pop(0) | |
| 21 assert expect == (data, kw), \ | |
| 22 "Expected %r, got %r" % ( | |
| 23 expect, (data, kw), | |
| 24 ) | |
| 25 | |
| 26 class EthernetTestCase(unittest.TestCase): | |
| 27 def testPacketParsing(self): | |
| 28 proto = ethernet.EthernetProtocol() | |
| 29 p1 = MyProtocol([ | |
| 30 | |
| 31 ('foobar', { | |
| 32 'partial': 0, | |
| 33 'dest': "123456", | |
| 34 'source': "987654", | |
| 35 'protocol': 0x0800, | |
| 36 }), | |
| 37 | |
| 38 ]) | |
| 39 proto.addProto(0x0800, p1) | |
| 40 | |
| 41 proto.datagramReceived("123456987654\x08\x00foobar", | |
| 42 partial=0) | |
| 43 | |
| 44 assert not p1.expecting, \ | |
| 45 'Should not expect any more packets, but still want %r' % p1.expe
cting | |
| 46 | |
| 47 | |
| 48 def testMultiplePackets(self): | |
| 49 proto = ethernet.EthernetProtocol() | |
| 50 p1 = MyProtocol([ | |
| 51 | |
| 52 ('foobar', { | |
| 53 'partial': 0, | |
| 54 'dest': "123456", | |
| 55 'source': "987654", | |
| 56 'protocol': 0x0800, | |
| 57 }), | |
| 58 | |
| 59 ('quux', { | |
| 60 'partial': 1, | |
| 61 'dest': "012345", | |
| 62 'source': "abcdef", | |
| 63 'protocol': 0x0800, | |
| 64 }), | |
| 65 | |
| 66 ]) | |
| 67 proto.addProto(0x0800, p1) | |
| 68 | |
| 69 proto.datagramReceived("123456987654\x08\x00foobar", | |
| 70 partial=0) | |
| 71 proto.datagramReceived("012345abcdef\x08\x00quux", | |
| 72 partial=1) | |
| 73 | |
| 74 assert not p1.expecting, \ | |
| 75 'Should not expect any more packets, but still want %r' % p1.expe
cting | |
| 76 | |
| 77 | |
| 78 def testMultipleSameProtos(self): | |
| 79 proto = ethernet.EthernetProtocol() | |
| 80 p1 = MyProtocol([ | |
| 81 | |
| 82 ('foobar', { | |
| 83 'partial': 0, | |
| 84 'dest': "123456", | |
| 85 'source': "987654", | |
| 86 'protocol': 0x0800, | |
| 87 }), | |
| 88 | |
| 89 ]) | |
| 90 | |
| 91 p2 = MyProtocol([ | |
| 92 | |
| 93 ('foobar', { | |
| 94 'partial': 0, | |
| 95 'dest': "123456", | |
| 96 'source': "987654", | |
| 97 'protocol': 0x0800, | |
| 98 }), | |
| 99 | |
| 100 ]) | |
| 101 | |
| 102 proto.addProto(0x0800, p1) | |
| 103 proto.addProto(0x0800, p2) | |
| 104 | |
| 105 proto.datagramReceived("123456987654\x08\x00foobar", | |
| 106 partial=0) | |
| 107 | |
| 108 assert not p1.expecting, \ | |
| 109 'Should not expect any more packets, but still want %r' % p1.expe
cting | |
| 110 assert not p2.expecting, \ | |
| 111 'Should not expect any more packets, but still want %r' % p2.expe
cting | |
| 112 | |
| 113 def testWrongProtoNotSeen(self): | |
| 114 proto = ethernet.EthernetProtocol() | |
| 115 p1 = MyProtocol([]) | |
| 116 proto.addProto(0x0801, p1) | |
| 117 | |
| 118 proto.datagramReceived("123456987654\x08\x00foobar", | |
| 119 partial=0) | |
| 120 proto.datagramReceived("012345abcdef\x08\x00quux", | |
| 121 partial=1) | |
| 122 | |
| 123 def testDemuxing(self): | |
| 124 proto = ethernet.EthernetProtocol() | |
| 125 p1 = MyProtocol([ | |
| 126 | |
| 127 ('foobar', { | |
| 128 'partial': 0, | |
| 129 'dest': "123456", | |
| 130 'source': "987654", | |
| 131 'protocol': 0x0800, | |
| 132 }), | |
| 133 | |
| 134 ('quux', { | |
| 135 'partial': 1, | |
| 136 'dest': "012345", | |
| 137 'source': "abcdef", | |
| 138 'protocol': 0x0800, | |
| 139 }), | |
| 140 | |
| 141 ]) | |
| 142 proto.addProto(0x0800, p1) | |
| 143 | |
| 144 p2 = MyProtocol([ | |
| 145 | |
| 146 ('quux', { | |
| 147 'partial': 1, | |
| 148 'dest': "012345", | |
| 149 'source': "abcdef", | |
| 150 'protocol': 0x0806, | |
| 151 }), | |
| 152 | |
| 153 ('foobar', { | |
| 154 'partial': 0, | |
| 155 'dest': "123456", | |
| 156 'source': "987654", | |
| 157 'protocol': 0x0806, | |
| 158 }), | |
| 159 | |
| 160 ]) | |
| 161 proto.addProto(0x0806, p2) | |
| 162 | |
| 163 proto.datagramReceived("123456987654\x08\x00foobar", | |
| 164 partial=0) | |
| 165 proto.datagramReceived("012345abcdef\x08\x06quux", | |
| 166 partial=1) | |
| 167 proto.datagramReceived("123456987654\x08\x06foobar", | |
| 168 partial=0) | |
| 169 proto.datagramReceived("012345abcdef\x08\x00quux", | |
| 170 partial=1) | |
| 171 | |
| 172 assert not p1.expecting, \ | |
| 173 'Should not expect any more packets, but still want %r' % p1.expe
cting | |
| 174 assert not p2.expecting, \ | |
| 175 'Should not expect any more packets, but still want %r' % p2.expe
cting | |
| 176 | |
| 177 def testAddingBadProtos_WrongLevel(self): | |
| 178 """Adding a wrong level protocol raises an exception.""" | |
| 179 e = ethernet.EthernetProtocol() | |
| 180 try: | |
| 181 e.addProto(42, "silliness") | |
| 182 except components.CannotAdapt: | |
| 183 pass | |
| 184 else: | |
| 185 raise AssertionError, 'addProto must raise an exception for bad prot
ocols' | |
| 186 | |
| 187 | |
| 188 def testAddingBadProtos_TooSmall(self): | |
| 189 """Adding a protocol with a negative number raises an exception.""" | |
| 190 e = ethernet.EthernetProtocol() | |
| 191 try: | |
| 192 e.addProto(-1, MyProtocol([])) | |
| 193 except TypeError, e: | |
| 194 if e.args == ('Added protocol must be positive or zero',): | |
| 195 pass | |
| 196 else: | |
| 197 raise | |
| 198 else: | |
| 199 raise AssertionError, 'addProto must raise an exception for bad prot
ocols' | |
| 200 | |
| 201 | |
| 202 def testAddingBadProtos_TooBig(self): | |
| 203 """Adding a protocol with a number >=2**16 raises an exception.""" | |
| 204 e = ethernet.EthernetProtocol() | |
| 205 try: | |
| 206 e.addProto(2**16, MyProtocol([])) | |
| 207 except TypeError, e: | |
| 208 if e.args == ('Added protocol must fit in 16 bits',): | |
| 209 pass | |
| 210 else: | |
| 211 raise | |
| 212 else: | |
| 213 raise AssertionError, 'addProto must raise an exception for bad prot
ocols' | |
| 214 | |
| 215 def testAddingBadProtos_TooBig2(self): | |
| 216 """Adding a protocol with a number >=2**16 raises an exception.""" | |
| 217 e = ethernet.EthernetProtocol() | |
| 218 try: | |
| 219 e.addProto(2**16+1, MyProtocol([])) | |
| 220 except TypeError, e: | |
| 221 if e.args == ('Added protocol must fit in 16 bits',): | |
| 222 pass | |
| 223 else: | |
| 224 raise | |
| 225 else: | |
| 226 raise AssertionError, 'addProto must raise an exception for bad prot
ocols' | |
| OLD | NEW |