Ticket #1442: endpoints3.diff

twisted/test/test_endpoints.py

@@ +1 @@
+from twisted.trial import unittest
+from twisted.internet import (defer, error, interfaces, reactor, 
+                              _sslverify as sslverify)
+from twisted.internet.address import IPv4Address, UNIXAddress
+from twisted.internet.protocol import ClientFactory, Protocol, ServerFactory
+from twisted.internet.endpoints import (TCPEndpoint, UNIXEndpoint)
+from twisted.test.test_sslverify import makeCertificate
+
+class ServerProtocol(Protocol):
+    def connectionMade(self):
+        self.factory.onConnectionMade.callback(self)
+
+    def connectionLost(self, *a):
+        self.factory.onConnectionLost.callback(self)
+
+class ClientProtocol(Protocol):
+    def connectionMade(self):
+        self.factory.onConnectionMade.callback(self)
+
+    def connectionLost(self, *a):
+        self.factory.onConnectionLost.callback(self)
+
+class MyServerFactory(ServerFactory):
+    protocol = ServerProtocol
+    
+    def __init__(self):
+        self.onConnectionMade = defer.Deferred()
+        self.onConnectionLost = defer.Deferred()
+
+class MyClientFactory(ClientFactory):
+    protocol = ClientProtocol
+    
+    def __init__(self):
+        self.onConnectionMade = defer.Deferred()
+        self.onConnectionLost = defer.Deferred()
+
+class PortAndConnectorCleanerUpper(unittest.TestCase):
+    def setUp(self):
+        self.listeningPorts = []
+        self.clientConnections = []
+    
+    def tearDown(self):
+        map(lambda p: p.stopListening(), self.listeningPorts)
+        map(lambda c: c.disconnect(), self.clientConnections)
+
+class EndpointTestCaseMixin(object):
+    def test_EndpointConnectSuccess(self):
+        """Test that Endpoint can connect and returns a deferred who
+        gets called back with a protocol instance. 
+        """
+        sf = MyServerFactory()
+        p = self.createServer(sf)
+        addr = p.getHost()
+        ep = self.createEndpoint(addr)
+        cf = MyClientFactory()
+        d = ep.connect(reactor, cf)
+        self.assertTrue(isinstance(d, defer.Deferred))
+        def onConnectSuccess(proto):
+            self.assertTrue(interfaces.IProtocol.providedBy(proto))
+            proto.transport.loseConnection()
+        d.addCallback(onConnectSuccess)
+        return defer.gatherResults([sf.onConnectionMade, cf.onConnectionMade, d])
+        
+    def test_EndpointConnectFailure(self):
+        """Test that if an Endpoint tries to connect to a none 
+        listening port that it gets a ConnectError failure.
+        """
+        p = self.createServer(MyServerFactory())
+        addr = p.getHost()
+        p.loseConnection()
+
+        ep = self.createEndpoint(addr)
+        d = ep.connect(reactor, MyClientFactory())
+        self.failUnlessFailure(d, error.ConnectError)
+        return d
+    
+    def test_EndpointListenSuccess(self):
+        """Test that Endpoint can listen and returns a deferred that
+        gets called back with a port instance. 
+        """
+        ep = self.createEndpoint()
+        sf = MyServerFactory()
+        d = ep.listen(reactor, sf)
+        self.assertTrue(isinstance(d, defer.Deferred))
+        def onListenSuccess(port):
+            self.assertTrue(interfaces.IListeningPort.providedBy(port))
+            self.listeningPorts.append(port)
+            return port.getHost()
+        d.addCallback(onListenSuccess)
+        def connectTo(addr):
+            self.createClient(addr, MyClientFactory())
+        d.addCallback(connectTo)
+        return defer.gatherResults([sf.onConnectionMade, d])
+
+    def test_EndpointListenFailure(self):
+        """Test that if Endpoint tries to listen on an already listening 
+        port, that a CannotListenError failure is errbacked. 
+        """
+        p = self.createServer(MyServerFactory())
+        addr = p.getHost()
+        ep = self.createEndpoint(addr)
+        d = ep.listen(reactor, MyServerFactory())
+        self.failUnlessFailure(d, error.CannotListenError)
+        return d
+
+class TCPEndpointsTestCase(PortAndConnectorCleanerUpper, EndpointTestCaseMixin):
+    def createServer(self, factory):
+        p = reactor.listenTCP(0, factory)
+        self.listeningPorts.append(p)
+        return p
+
+    def createClient(self, address, factory):
+        c = reactor.connectTCP(address.host, address.port, factory)
+        self.clientConnections.append(c)
+        return c
+
+    def createEndpoint(self, address=None):
+        if not address:
+            address = IPv4Address("TCP", "localhost", 0)
+        return TCPEndpoint(address.host, address.port)
+
+class SSLEndpointsTestCase(PortAndConnectorCleanerUpper, EndpointTestCaseMixin):
+    
+    def setUpClass(self):
+        self.sKey, self.sCert = makeCertificate(
+            O="Server Test Certificate",
+            CN="server")
+        self.cKey, self.cCert = makeCertificate(
+            O="Client Test Certificate",
+            CN="client")
+        self.serverSSLContext = sslverify.OpenSSLCertificateOptions(privateKey=self.sKey, certificate=self.sCert, requireCertificate=False)
+        self.clientSSLContext = sslverify.OpenSSLCertificateOptions(requireCertificate=False)
+        
+    def createServer(self, factory):
+        p = reactor.listenSSL(0, factory, self.serverSSLContext)
+        self.listeningPorts.append(p)
+        return p
+
+    def createClient(self, address, factory):
+        c = reactor.connectSSL(address.host, address.port, factory, 
+                               self.clientSSLContext)
+        self.clientConnections.append(c)
+        return c
+
+    def createEndpoint(self, address=None):
+        if not address:
+            address = IPv4Address("TCP", "localhost", 0)
+        return TCPEndpoint(address.host, address.port, 
+                           sslContextFactory=self.clientSSLContext)
+        
+
+class UNIXEndpointsTestCase(PortAndConnectorCleanerUpper, EndpointTestCaseMixin):
+    def createServer(self, factory):
+        p = reactor.listenUNIX(self.mktemp(), factory)
+        self.listeningPorts.append(p)
+        return p
+
+    def createClient(self, address, factory):
+        c = reactor.connectUNIX(address.name, factory)
+        self.clientConnections.append(c)
+        return c
+
+    def createEndpoint(self, address=None):
+        if not address:
+            address = UNIXAddress(self.mktemp())
+        return UNIXEndpoint(address.name)

twisted/internet/endpoints.py

@@ +1 @@
+# -*- test-case-name: twisted.test.test_endpoints -*-
+
+from zope.interface import implements, providedBy, directlyProvides
+
+from twisted.internet import interfaces
+from twisted.internet import defer, protocol
+from twisted.internet.protocol import ClientFactory, Protocol
+
+class _WrappingProtocol(Protocol):
+    """I wrap another protocol in order to notify my user when a connection has 
+    been made.
+    """
+    def __init__(self, factory, wrappedProtocol):
+        self.factory = factory
+        self.wrappedProtocol = wrappedProtocol
+        
+    def connectionMade(self):
+        """XXX: As soon as I am connected, I connect my wrappedProtocol, giving 
+        it my transport. Is it okay for a transport to be associated with more
+        than one protocol? Transport calls dataReceived on me and I in turn call
+        dataReceived on my wrappedProtocol. The wrappedProtocol may call 
+        transport.write or transport.loseConnection etc
+        """
+        
+        self.wrappedProtocol.makeConnection(self.transport)
+        self.factory.deferred.callback(self.wrappedProtocol)
+        
+    def dataReceived(self, data):
+        return self.wrappedProtocol.dataReceived(data)
+
+    def connectionLost(self, reason):
+        return self.wrappedProtocol.connectionLost(reason)
+        
+class _WrappingFactory(ClientFactory):
+    protocol = _WrappingProtocol
+
+    def __init__(self, wrappedFactory):
+        self.wrappedFactory = wrappedFactory
+        self.deferred = defer.Deferred()
+
+    def buildProtocol(self, addr):
+        try:
+            proto = self.wrappedFactory.buildProtocol(addr)
+        except:
+            self.deferred.errback()
+        else:
+            return self.protocol(self, proto)
+
+    def clientConnectionFailed(self, connector, reason):
+        self.deferred.errback(reason)
+
+
+class TCPEndpoint(object):
+    implements(interfaces.IClientEndpoint, interfaces.IServerEndpoint)
+
+    def __init__(self, host, port, connectArgs={}, listenArgs={},
+                 sslContextFactory=None):
+        """
+        @param host: A hostname, used only when connecting
+        @param port: The port number, used both when connecting and listening
+        @param connectArgs: An optional dict of keyword args that will be passed
+        to L{twisted.internet.interfaces.IReactorTCP.connectTCP}
+        @param listenArgs: An optional dict of keyword args that will be passed 
+        to L{twisted.internet.interfaces.IReactorTCP.listenTCP}
+        @param sslContextFactory: An optional instance of 
+        L{twisted.internet._sslverify.OpenSSLCertificateOptions}. If given, it 
+        makes L{connect} and L{listen} use the corresponding methods from 
+        L{twisted.internet.interfaces.IReactorSSL}
+        """
+        self.host = host
+        self.port = port
+        self.connectArgs = dict(timeout=30, bindAddress=None)
+        self.connectArgs.update(connectArgs)
+        self.listenArgs = dict(backlog=50, interface='')
+        self.listenArgs.update(listenArgs)
+        self.sslContextFactory = sslContextFactory
+        
+    def connect(self, reactor, clientFactory):
+        wf = _WrappingFactory(clientFactory)
+        connectArgs = self.connectArgs
+        connectMethod = reactor.connectTCP
+        if self.sslContextFactory:
+            connectMethod = reactor.connectSSL
+            connectArgs["contextFactory"] = self.sslContextFactory
+            
+        d = defer.execute(connectMethod, self.host, self.port, wf, **connectArgs)
+
+        d.addCallback(lambda _: wf.deferred)
+
+        return d
+
+    def listen(self, reactor, serverFactory):
+        wf = _WrappingFactory(serverFactory)
+        listenArgs = self.listenArgs
+        listenMethod = reactor.listenTCP
+        if self.sslContextFactory:
+            listenMethod = reactor.listenSSL
+            listenArgs["contextFactory"] = self.sslContextFactory
+        return defer.execute(listenMethod, self.port, wf, **listenArgs)
+
+class UNIXEndpoint(object):
+    implements(interfaces.IClientEndpoint, interfaces.IServerEndpoint)
+
+    def __init__(self, address, connectArgs={}, listenArgs={}):
+        """
+        @param address: The path to the Unix socket file, used both when 
+        connecting and listening
+        @param connectArgs: A dict of keyword args that will be passed to 
+        L{twisted.internet.interfaces.IReactorUNIX.connectUNIX}
+        @param listenArgs: A dict of keyword args that will be passed to 
+        L{twisted.internet.interfaces.IReactorUNIX.listenUNIX}
+        """
+
+        self.address = address
+        self.connectArgs = dict(timeout=30, checkPID=0)
+        self.connectArgs.update(connectArgs)
+        self.listenArgs = dict(backlog=50, mode=0666, wantPID=0)
+        self.listenArgs.update(listenArgs)
+
+    def connect(self, reactor, clientFactory):
+        wf = _WrappingFactory(clientFactory)
+        d = defer.execute(reactor.connectUNIX, self.address, wf, 
+                          **self.connectArgs)
+
+        d.addCallback(lambda _: wf.deferred)
+
+        return d
+
+    def listen(self, reactor, serverFactory):
+        wf = _WrappingFactory(serverFactory)
+        return defer.execute(reactor.listenUNIX, self.address, wf, 
+                             **self.listenArgs)

twisted/internet/interfaces.py

@@ -18 +18 @@
 
     Default implementations are in L{twisted.internet.address}.
     """
+    
+    def buildEndpoint():
+        """
+        @return: an instance providing both L{IClientEndpoint} and L{IServerEndpoint}
+        """
 
 
 ### Reactor Interfaces
@@ -1294 +1299 @@
 
     def leaveGroup(addr, interface=""):
         """Leave multicast group, return Deferred of success."""
+
+class IClientEndpoint(Interface):
+    """Object that represents a remote endpoint that we wish to connect to.
+    """
+    def connect(reactor, clientFactory):
+        """
+        @param reactor: The reactor
+        @param clientFactory: A provider of L{IProtocolFactory}
+        @return: A L{Deferred} that results in an L{IProtocol} upon successful
+        connection otherwise a L{ConnectError}
+        """
+
+class IServerEndpoint(Interface):
+    """Object representing an endpoint where we will listen for connections.
+    """
+
+    def listen(callable):
+        """
+        @param reactor: The reactor
+        @param serverFactory: A provider of L{IProtocolFactory}
+        @return: A L{Deferred} that results in an L{IListeningPort} or an 
+        L{CannotListenError}
+        """

twisted/internet/address.py

@@ -5 +5 @@
 """Address objects for network connections."""
 
 import warnings, os
+
 from zope.interface import implements
+
 from twisted.internet.interfaces import IAddress
+from twisted.internet.endpoints import TCPEndpoint, UNIXEndpoint
 
-
 class IPv4Address(object):
     """
     Object representing an IPv4 socket endpoint.
@@ -34 +36 @@
         self.port = port
         self._bwHack = _bwHack
 
+    def buildEndpoint(self):
+        if self.type == "TCP":
+            return TCPEndpoint(self.host, self.port)
+        else:
+            raise NotImplementedError
+
     def __getitem__(self, index):
         warnings.warn("IPv4Address.__getitem__ is deprecated.  Use attributes instead.",
                       category=DeprecationWarning, stacklevel=2)
@@ -70 +78 @@
     def __init__(self, name, _bwHack='UNIX'):
         self.name = name
         self._bwHack = _bwHack
-    
+
+    def buildEndpoint(self):
+        return UNIXEndpoint(self.name)
+
     def __getitem__(self, index):
         warnings.warn("UNIXAddress.__getitem__ is deprecated.  Use attributes instead.",
                       category=DeprecationWarning, stacklevel=2)