Viewing file:  test_tcp.py (4.11 KB) -rw-r--r--
Select action/file-type:
 (+) |  (+) |  (+) | Code (+) | Session (+) |  (+) | SDB (+) |  (+) |  (+) |  (+) |  (+) |  (+) |
# Copyright (c) 2008 Twisted Matrix Laboratories.
# See LICENSE for details.
"""
Tests for implementations of L{IReactorTCP}.
"""
__metaclass__ = type
import socket
from zope.interface import implements
from twisted.internet.test.reactormixins import ReactorBuilder
from twisted.internet.error import DNSLookupError
from twisted.internet.interfaces import IResolverSimple
from twisted.internet.address import IPv4Address
from twisted.internet.defer import succeed, fail
from twisted.internet.protocol import ServerFactory, ClientFactory, Protocol
class Stop(ClientFactory):
"""
A client factory which stops a reactor when a connection attempt fails.
"""
def __init__(self, reactor):
self.reactor = reactor
def clientConnectionFailed(self, connector, reason):
self.reactor.stop()
class FakeResolver:
"""
A resolver implementation based on a C{dict} mapping names to addresses.
"""
implements(IResolverSimple)
def __init__(self, names):
self.names = names
def getHostByName(self, name, timeout):
try:
return succeed(self.names[name])
except KeyError:
return fail(DNSLookupError("FakeResolver couldn't find " + name))
class TCPClientTestsBuilder(ReactorBuilder):
"""
Builder defining tests relating to L{IReactorTCP.connectTCP}.
"""
def _freePort(self, interface='127.0.0.1'):
probe = socket.socket()
try:
probe.bind((interface, 0))
return probe.getsockname()
finally:
probe.close()
def test_clientConnectionFailedStopsReactor(self):
"""
The reactor can be stopped by a client factory's
C{clientConnectionFailed} method.
"""
host, port = self._freePort()
reactor = self.buildReactor()
reactor.connectTCP(host, port, Stop(reactor))
reactor.run()
def test_addresses(self):
"""
A client's transport's C{getHost} and C{getPeer} return L{IPv4Address}
instances which give the dotted-quad string form of the local and
remote endpoints of the connection respectively.
"""
host, port = self._freePort()
reactor = self.buildReactor()
serverFactory = ServerFactory()
serverFactory.protocol = Protocol
server = reactor.listenTCP(0, serverFactory, interface=host)
serverAddress = server.getHost()
addresses = {'host': None, 'peer': None}
class CheckAddress(Protocol):
def makeConnection(self, transport):
addresses['host'] = transport.getHost()
addresses['peer'] = transport.getPeer()
reactor.stop()
clientFactory = Stop(reactor)
clientFactory.protocol = CheckAddress
client = reactor.connectTCP(
'localhost', server.getHost().port, clientFactory,
bindAddress=('127.0.0.1', port))
reactor.installResolver(FakeResolver({'localhost': '127.0.0.1'}))
reactor.run() # self.runReactor(reactor)
self.assertEqual(
addresses['host'],
IPv4Address('TCP', '127.0.0.1', port))
self.assertEqual(
addresses['peer'],
IPv4Address('TCP', '127.0.0.1', serverAddress.port))
def test_connectEvent(self):
"""
This test checks that we correctly get notifications event for a
client. This ought to prevent a regression under Windows using the GTK2
reactor. See #3925.
"""
reactor = self.buildReactor()
serverFactory = ServerFactory()
serverFactory.protocol = Protocol
server = reactor.listenTCP(0, serverFactory)
connected = []
class CheckConnection(Protocol):
def connectionMade(self):
connected.append(self)
reactor.stop()
clientFactory = Stop(reactor)
clientFactory.protocol = CheckConnection
client = reactor.connectTCP(
'127.0.0.1', server.getHost().port, clientFactory)
reactor.run()
self.assertTrue(connected)
globals().update(TCPClientTestsBuilder.makeTestCaseClasses())
|