# Copyright (c) Twisted Matrix Laboratories. # See LICENSE for details. """ Tests for implementations of L{IReactorTCP}. """ __metaclass__ = type import socket from zope.interface import implements from zope.interface.verify import verifyObject from twisted.internet.test.reactormixins import ReactorBuilder from twisted.internet.error import DNSLookupError from twisted.internet.interfaces import ( IResolverSimple, IConnector, IReactorFDSet) from twisted.internet.address import IPv4Address from twisted.internet.defer import Deferred, DeferredList, succeed, fail, maybeDeferred from twisted.internet.endpoints import TCP4ClientEndpoint from twisted.internet.protocol import ServerFactory, ClientFactory, Protocol from twisted.python.runtime import platform from twisted.python.failure import Failure from twisted.python import log from twisted.trial.unittest import SkipTest, TestCase from twisted.internet.tcp import Connection from twisted.test.test_tcp import ClosingProtocol from twisted.internet.test.test_core import ObjectModelIntegrationMixin def findFreePort(interface='127.0.0.1', type=socket.SOCK_STREAM): """ Ask the platform to allocate a free port on the specified interface, then release the socket and return the address which was allocated. @param interface: The local address to try to bind the port on. @type interface: C{str} @param type: The socket type which will use the resulting port. @return: A two-tuple of address and port, like that returned by L{socket.getsockname}. """ family = socket.AF_INET probe = socket.socket(family, type) try: probe.bind((interface, 0)) return probe.getsockname() finally: probe.close() 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 _SimplePullProducer(object): """ A pull producer which writes one byte whenever it is resumed. For use by L{test_unregisterProducerAfterDisconnect}. """ def __init__(self, consumer): self.consumer = consumer def stopProducing(self): pass def resumeProducing(self): log.msg("Producer.resumeProducing") self.consumer.write('x') def _getWriters(reactor): """ Like L{IReactorFDSet.getWriters}, but with support for IOCP reactor as well. """ if IReactorFDSet.providedBy(reactor): return reactor.getWriters() elif 'IOCP' in reactor.__class__.__name__: return reactor.handles else: # Cannot tell what is going on. raise Exception("Cannot find writers on %r" % (reactor,)) def serverFactoryFor(protocol): """ Helper function which provides the signature L{ServerFactory} should provide. """ factory = ServerFactory() factory.protocol = protocol return factory class FakeSocket(object): """ A Fake Socket object """ fileno = 1 def __init__(self, data): self.data = data def setblocking(self, blocking): self.blocking = blocking def recv(self, size): return self.data class TestFakeSocket(TestCase): """ Test that the FakeSocket can be used by the doRead method of L{Connection} """ def test_blocking(self): skt = FakeSocket("someData") skt.setblocking(0) self.assertEquals(skt.blocking, 0) def test_recv(self): skt = FakeSocket("someData") self.assertEquals(skt.recv(10), "someData") class FakeProtocol(Protocol): """ An L{IProtocol} that returns a value from its dataReceived method. """ def dataReceived(self, data): """ Return something other than C{None} to trigger a deprecation warning for that behavior. """ return () class TCPConnectionTests(TestCase): """ Whitebox tests for L{twisted.internet.tcp.Connection}. """ def test_doReadWarningIsRaised(self): """ When an L{IProtocol} implementation that returns a value from its C{dataReceived} method, a deprecated warning is emitted. """ skt = FakeSocket("someData") protocol = FakeProtocol() conn = Connection(skt, protocol) conn.doRead() warnings = self.flushWarnings([FakeProtocol.dataReceived]) self.assertEquals(warnings[0]['category'], DeprecationWarning) self.assertEquals( warnings[0]["message"], "Returning a value other than None from " "twisted.internet.test.test_tcp.FakeProtocol.dataReceived " "is deprecated since Twisted 11.0.0.") self.assertEquals(len(warnings), 1) class TCPClientTestsBuilder(ReactorBuilder): """ Builder defining tests relating to L{IReactorTCP.connectTCP}. """ def test_interface(self): """ L{IReactorTCP.connectTCP} returns an object providing L{IConnector}. """ reactor = self.buildReactor() connector = reactor.connectTCP("127.0.0.1", 1234, ClientFactory()) self.assertTrue(verifyObject(IConnector, connector)) def test_clientConnectionFailedStopsReactor(self): """ The reactor can be stopped by a client factory's C{clientConnectionFailed} method. """ host, port = findFreePort() reactor = self.buildReactor() reactor.connectTCP(host, port, Stop(reactor)) self.runReactor(reactor) 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 = findFreePort() reactor = self.buildReactor() server = reactor.listenTCP( 0, serverFactoryFor(Protocol), 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 reactor.connectTCP( 'localhost', server.getHost().port, clientFactory, bindAddress=('127.0.0.1', port)) reactor.installResolver(FakeResolver({'localhost': '127.0.0.1'})) 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() server = reactor.listenTCP(0, serverFactoryFor(Protocol)) connected = [] class CheckConnection(Protocol): def connectionMade(self): connected.append(self) reactor.stop() clientFactory = Stop(reactor) clientFactory.protocol = CheckConnection reactor.connectTCP( '127.0.0.1', server.getHost().port, clientFactory) reactor.run() self.assertTrue(connected) def test_unregisterProducerAfterDisconnect(self): """ If a producer is unregistered from a L{ITCPTransport} provider after the transport has been disconnected (by the peer) and after L{ITCPTransport.loseConnection} has been called, the transport is not re-added to the reactor as a writer as would be necessary if the transport were still connected. """ reactor = self.buildReactor() port = reactor.listenTCP(0, serverFactoryFor(ClosingProtocol)) finished = Deferred() finished.addErrback(log.err) finished.addCallback(lambda ign: reactor.stop()) writing = [] class ClientProtocol(Protocol): """ Protocol to connect, register a producer, try to lose the connection, wait for the server to disconnect from us, and then unregister the producer. """ def connectionMade(self): log.msg("ClientProtocol.connectionMade") self.transport.registerProducer( _SimplePullProducer(self.transport), False) self.transport.loseConnection() def connectionLost(self, reason): log.msg("ClientProtocol.connectionLost") self.unregister() writing.append(self.transport in _getWriters(reactor)) finished.callback(None) def unregister(self): log.msg("ClientProtocol unregister") self.transport.unregisterProducer() clientFactory = ClientFactory() clientFactory.protocol = ClientProtocol reactor.connectTCP('127.0.0.1', port.getHost().port, clientFactory) self.runReactor(reactor) self.assertFalse( writing[0], "Transport was writing after unregisterProducer.") def test_disconnectWhileProducing(self): """ If L{ITCPTransport.loseConnection} is called while a producer is registered with the transport, the connection is closed after the producer is unregistered. """ reactor = self.buildReactor() # XXX For some reason, pyobject/pygtk will not deliver the close # notification that should happen after the unregisterProducer call in # this test. The selectable is in the write notification set, but no # notification ever arrives. skippedReactors = ["Glib2Reactor", "Gtk2Reactor"] reactorClassName = reactor.__class__.__name__ if reactorClassName in skippedReactors and platform.isWindows(): raise SkipTest( "A pygobject/pygtk bug disables this functionality on Windows.") class Producer: def resumeProducing(self): log.msg("Producer.resumeProducing") port = reactor.listenTCP(0, serverFactoryFor(Protocol)) finished = Deferred() finished.addErrback(log.err) finished.addCallback(lambda ign: reactor.stop()) class ClientProtocol(Protocol): """ Protocol to connect, register a producer, try to lose the connection, unregister the producer, and wait for the connection to actually be lost. """ def connectionMade(self): log.msg("ClientProtocol.connectionMade") self.transport.registerProducer(Producer(), False) self.transport.loseConnection() # Let the reactor tick over, in case synchronously calling # loseConnection and then unregisterProducer is the same as # synchronously calling unregisterProducer and then # loseConnection (as it is in several reactors). reactor.callLater(0, reactor.callLater, 0, self.unregister) def unregister(self): log.msg("ClientProtocol unregister") self.transport.unregisterProducer() # This should all be pretty quick. Fail the test # if we don't get a connectionLost event really # soon. reactor.callLater( 1.0, finished.errback, Failure(Exception("Connection was not lost"))) def connectionLost(self, reason): log.msg("ClientProtocol.connectionLost") finished.callback(None) clientFactory = ClientFactory() clientFactory.protocol = ClientProtocol reactor.connectTCP('127.0.0.1', port.getHost().port, clientFactory) self.runReactor(reactor) # If the test failed, we logged an error already and trial # will catch it. class TCPPortTestsBuilder(ReactorBuilder, ObjectModelIntegrationMixin): """ Tests for L{IReactorRCP.listenTCP} """ def getListeningPort(self, reactor): """ Get a TCP port from a reactor """ return reactor.listenTCP(0, ServerFactory()) def getExpectedConnectionLostLogMsg(self, port): """ Get the expected connection lost message for a TCP port """ return "(TCP Port %s Closed)" % (port.getHost().port,) def test_connectionLostLogMsg(self): """ When a connection is lost, an informative message should be logged (see L{getExpectedConnectionLostLogMsg}): an address identifying the port and the fact that it was closed. """ loggedMessages = [] def logConnectionLostMsg(eventDict): loggedMessages.append(log.textFromEventDict(eventDict)) reactor = self.buildReactor() p = self.getListeningPort(reactor) expectedMessage = self.getExpectedConnectionLostLogMsg(p) log.addObserver(logConnectionLostMsg) def stopReactor(ignored): log.removeObserver(logConnectionLostMsg) reactor.stop() def doStopListening(): log.addObserver(logConnectionLostMsg) maybeDeferred(p.stopListening).addCallback(stopReactor) reactor.callWhenRunning(doStopListening) reactor.run() self.assertIn(expectedMessage, loggedMessages) def test_allNewStyle(self): """ The L{IListeningPort} object is an instance of a class with no classic classes in its hierarchy. """ reactor = self.buildReactor() port = self.getListeningPort(reactor) self.assertFullyNewStyle(port) class StopStartReadingProtocol(Protocol): """ Protocol that pauses and resumes the transport a few times """ def connectionMade(self): self.data = '' self.pauseResumeProducing(3) def pauseResumeProducing(self, counter): """ Toggle transport read state, then count down. """ self.transport.pauseProducing() self.transport.resumeProducing() if counter: self.factory.reactor.callLater(0, self.pauseResumeProducing, counter - 1) else: self.factory.reactor.callLater(0, self.factory.ready.callback, self) def dataReceived(self, data): log.msg('got data', len(data)) self.data += data if len(self.data) == 4*4096: self.factory.stop.callback(self.data) class TCPConnectionTestsBuilder(ReactorBuilder): """ Builder defining tests relating to L{twisted.internet.tcp.Connection}. """ def test_stopStartReading(self): """ This test verifies transport socket read state after multiple pause/resumeProducing calls. """ sf = ServerFactory() reactor = sf.reactor = self.buildReactor() skippedReactors = ["Glib2Reactor", "Gtk2Reactor"] reactorClassName = reactor.__class__.__name__ if reactorClassName in skippedReactors and platform.isWindows(): raise SkipTest( "This test is broken on gtk/glib under Windows.") sf.protocol = StopStartReadingProtocol sf.ready = Deferred() sf.stop = Deferred() p = reactor.listenTCP(0, sf) port = p.getHost().port def proceed(protos, port): """ Send several IOCPReactor's buffers' worth of data. """ self.assertTrue(protos[0]) self.assertTrue(protos[1]) protos = protos[0][1], protos[1][1] protos[0].transport.write('x' * (2 * 4096) + 'y' * (2 * 4096)) return (sf.stop.addCallback(cleanup, protos, port) .addCallback(lambda ign: reactor.stop())) def cleanup(data, protos, port): """ Make sure IOCPReactor didn't start several WSARecv operations that clobbered each other's results. """ self.assertEquals(data, 'x'*(2*4096) + 'y'*(2*4096), 'did not get the right data') return DeferredList([ maybeDeferred(protos[0].transport.loseConnection), maybeDeferred(protos[1].transport.loseConnection), maybeDeferred(port.stopListening)]) cc = TCP4ClientEndpoint(reactor, '127.0.0.1', port) cf = ClientFactory() cf.protocol = Protocol d = DeferredList([cc.connect(cf), sf.ready]).addCallback(proceed, p) self.runReactor(reactor) return d globals().update(TCPClientTestsBuilder.makeTestCaseClasses()) globals().update(TCPPortTestsBuilder.makeTestCaseClasses()) globals().update(TCPConnectionTestsBuilder.makeTestCaseClasses())