Ticket #2060: iaddress-connect-listen.diff

File iaddress-connect-listen.diff, 5.4 KB (added by rwall, 10 years ago)
  • twisted/test/test_unix.py

     
    299299
    300300        return defer.maybeDeferred(p.stopListening).addCallback(stoppedListening)
    301301
     302class UNIXAddressTestCase(PortCleanerUpper):
     303    socketPath = "/tmp/UNIXAddressTestCase.socket"
     304    def testConnect(self):
     305        """Test that UNIXAddress.connect returns an IConnector"""
     306        listeningPort = reactor.listenUNIX(self.socketPath, Factory(self, self.socketPath))
     307        self.ports.append(listeningPort)
     308        addr = address.UNIXAddress(self.socketPath)
     309        conn = addr.connect(TestClientFactory(self, self.socketPath))
     310        self.assertTrue(interfaces.IConnector.providedBy(conn))
    302311
     312    def testListen(self):
     313        """Test that UNIXAddress.listen returns an IListeningPort"""
     314        addr = address.UNIXAddress(self.socketPath)
     315        listeningPort = addr.listen(Factory(self, self.socketPath))
     316        self.ports.append(listeningPort)
     317        self.assertTrue(interfaces.IListeningPort.providedBy(listeningPort))
     318
    303319if not interfaces.IReactorUNIX(reactor, None):
    304320    UnixSocketTestCase.skip = "This reactor does not support UNIX domain sockets"
    305321if not interfaces.IReactorUNIXDatagram(reactor, None):
  • twisted/test/test_tcp.py

     
    1515from twisted.internet import protocol, reactor, defer, interfaces
    1616from twisted.internet import error
    1717from twisted.internet.address import IPv4Address
    18 from twisted.internet.interfaces import IHalfCloseableProtocol
     18from twisted.internet.interfaces import (IConnector, IHalfCloseableProtocol,
     19                                         IListeningPort)
    1920from twisted.protocols import policies
    2021
    2122
     
    11421143        d.addCallback(lambda _: log.flushErrors(RuntimeError))
    11431144        return d
    11441145
     1146class IPv4AddressTestCase(PortCleanerUpper):
     1147    def testConnect(self):
     1148        """Test that IPv4Address.connect returns an IConnector"""
     1149        factory = ClosingFactory()
     1150        listeningPort = reactor.listenTCP(0, factory)
     1151        self.ports.append(listeningPort)
     1152        factory.port = listeningPort
     1153        portNo = listeningPort.getHost().port
     1154        addr = IPv4Address("TCP", "127.0.0.1", portNo)
     1155        conn = addr.connect(MyClientFactory())
     1156        self.assertTrue(IConnector.providedBy(conn))
    11451157
     1158    def testListen(self):
     1159        """Test that IPv4Address.listen returns an IListeningPort"""
     1160        factory = ClosingFactory()
     1161        addr = IPv4Address("TCP", "0.0.0.0", 0)
     1162        listeningPort = addr.listen(factory)
     1163        self.ports.append(listeningPort)
     1164        factory.port = listeningPort
     1165        self.assertTrue(IListeningPort.providedBy(listeningPort))
     1166
    11461167try:
    11471168    import resource
    11481169except ImportError:
  • twisted/internet/interfaces.py

     
    1919    Default implementations are in L{twisted.internet.address}.
    2020    """
    2121
    22 
     22    def connect(factory, timeout=30):
     23        """Attempt to connect to my address
     24       
     25        @param factory: A protocol factory
     26        @return: An object which provides L{IConnector}.
     27        """
     28       
     29    def listen(factory, backlog=50):
     30        """Attempt to listen at my address
     31       
     32        @param factory: A protocol factory
     33        @return: An object which provides L{IListeningPort}.
     34        """       
    2335### Reactor Interfaces
    2436
    2537class IConnector(Interface):
  • twisted/internet/address.py

     
    3434        self.port = port
    3535        self._bwHack = _bwHack
    3636
     37    def connect(self, factory, timeout=30):
     38        from twisted.internet import reactor
     39        return reactor.connectTCP(self.host, self.port, factory, timeout)
     40
     41    def listen(self, factory, backlog=50):
     42        from twisted.internet import reactor
     43        return reactor.listenTCP(self.port, factory, backlog, interface=self.host)
     44
    3745    def __getitem__(self, index):
    3846        warnings.warn("IPv4Address.__getitem__ is deprecated.  Use attributes instead.",
    3947                      category=DeprecationWarning, stacklevel=2)
     
    7078    def __init__(self, name, _bwHack='UNIX'):
    7179        self.name = name
    7280        self._bwHack = _bwHack
    73    
     81
     82    def connect(self, factory, timeout=30):
     83        from twisted.internet import reactor
     84        return reactor.connectUNIX(self.name, factory, timeout)
     85
     86    def listen(self, factory, backlog=50):
     87        from twisted.internet import reactor
     88        return reactor.listenUNIX(self.name, factory, backlog)
     89
    7490    def __getitem__(self, index):
    7591        warnings.warn("UNIXAddress.__getitem__ is deprecated.  Use attributes instead.",
    7692                      category=DeprecationWarning, stacklevel=2)
     
    92108        return False
    93109
    94110    def __str__(self):
    95         return 'UNIXSocket(%r)' % (self.name,)
     111        return 'UNIXAddress(%r)' % (self.name,)
    96112
    97113
    98114# These are for buildFactory backwards compatability due to