[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):

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__)

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)

            def connectionMade(self):
                log.debug('trace, writing "%s"' % payload)
                assert self.transport

            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='', port = 10040)
        t = self.sendTraffic( vip)
        reactor.callLater(1, self.killClient, t)

More information about the Twisted-Python mailing list