1
# Copyright (c) 2001-2007 Twisted Matrix Laboratories.
2
# See LICENSE for details.
6
Test cases for twisted.protocols.stateful
9
from twisted.test import test_protocols
10
from twisted.protocols.stateful import StatefulProtocol
12
from struct import pack, unpack, calcsize
15
class MyInt32StringReceiver(StatefulProtocol):
17
A stateful Int32StringReceiver.
21
prefixLength = calcsize(structFormat)
23
def getInitialState(self):
24
return self._getHeader, 4
26
def lengthLimitExceeded(self, length):
27
self.transport.loseConnection()
29
def _getHeader(self, msg):
30
length, = unpack("!i", msg)
31
if length > self.MAX_LENGTH:
32
self.lengthLimitExceeded(length)
34
return self._getString, length
36
def _getString(self, msg):
37
self.stringReceived(msg)
38
return self._getHeader, 4
40
def stringReceived(self, msg):
44
raise NotImplementedError
46
def sendString(self, data):
48
Send an int32-prefixed string to the other end of the connection.
50
self.transport.write(pack(self.structFormat, len(data)) + data)
53
class TestInt32(MyInt32StringReceiver):
54
def connectionMade(self):
57
def stringReceived(self, s):
58
self.received.append(s)
63
def connectionLost(self, reason):
67
class Int32TestCase(test_protocols.Int32TestCase):
70
def test_bigReceive(self):
71
r = self.getProtocol()
73
for s in self.strings * 4:
74
big += pack("!i", len(s)) + s
76
self.assertEquals(r.received, self.strings * 4)