[Twisted-Python] telnet works, why wont this client/protocol test?
Mike Winter
miwinter at cisco.com
Thu Jan 12 03:40:17 EST 2012
i can telnet to the port given in test-code below and it interacts appropriately. The connection is to a regular portforwarder and the server is an Answer-server:
class Answer(LineReceiver):
answers = {'How are you?': 'Fine',
'no': 'No!?!',
None : "I don't know what you mean",
'hello': 'hi there',
'Whats Up': 'not much' }
def lineReceived(self, line):
log.debug('line: %s' % line)
if self.answers.has_key(line):
self.sendLine(self.answers[line])
else:
self.sendLine(self.answers[None])
Its not a unit-test, but it reasonably expresses code that I dont understand
from twisted.trial.unittest import TestCase
from twisted.internet import reactor, protocol, defer
from twisted.internet.protocol import ClientFactory,ReconnectingClientFactory
from twisted.protocols.policies import WrappingFactory
from twisted.protocols import basic
import localproxy
import settings
from state import State
import engine
from server import Factory, AssertingAnswer, AssertingFactoryMixin
from assertion import URL
import logging
global log
log = logging.getLogger(__name__)
log.setLevel(settings.loglevel)
class Test(TestCase):
def sendTraffic(self, vip, count = 1,
payload = 'Whats Up\n',
expectedResponse = 'not much'):
'''telnet to DUT and post payload'''
log.debug('vip: %s:%s' % (vip.host, vip.port))
t = self
class Hello(basic.LineReceiver):
def dataReceived(self, data):
log.debug('trace: %s' % data)
return basic.LineReceiver.dataReceived(self,data)
def lineReceived(self, line):
log.debug('trace: %s' % line)
t.assertEquals(line, expectedResponse)
self.transport.loseConnection()
def connectionMade(self):
log.debug('trace, writing "%s"' % payload)
assert self.transport
self.transport.write(payload)
def connectionLost(self, reason):
log.debug('reason: %s' % reason)
f = ReconnectingClientFactory()
f.protocol = Hello
factory = WrappingFactory(f)
reactor.connectTCP( vip.host, vip.port, factory)
log.debug('clientFactory: %s' % factory.__dict__)
return factory
def killClient(t):
log.debug('t: %s' % t)
def testClientConnect(self):
vip = State(host='127.0.0.1', port = 10040)
t = self.sendTraffic( vip)
reactor.callLater(1, self.killClient, t)
More information about the Twisted-Python
mailing list