Ticket #5086: fix-ipv6-literals-udp.4.patch

File fix-ipv6-literals-udp.4.patch, 12.1 KB (added by marto1_, 15 months ago)

Polished tests in twisted.test.test_udp

  • twisted/test/test_udp.py

     
    168168 
    169169        return d.addCallback(cbStarted, p) 
    170170 
     171    def test_bindToIPv6Interface(self): 
     172        """ 
     173        Bind to ipv6 interface. 
     174        """ 
     175        server = Server() 
     176        p = reactor.listenUDP(0, server, interface="::1") 
     177        self.assertEqual(p.getHost().host, "::1") 
    171178 
     179        return p.stopListening() 
     180 
     181    def cbServerStarted(self, ignored, client, interface, d): 
     182        self.port2 = reactor.listenUDP(0, client, interface=interface) 
     183        return d 
     184 
     185    def cbClientStarted(self, ignored, client, server, interface): 
     186        client.transport.connect(interface, 
     187                                 server.transport.getHost().port) 
     188        cAddr = client.transport.getHost() 
     189        sAddr = server.transport.getHost() 
     190 
     191        serverSend = client.packetReceived = defer.Deferred() 
     192        server.transport.write(b"hello", (cAddr.host, cAddr.port)) 
     193 
     194        clientWrites = [ 
     195            (b"a",), 
     196            (b"b", None), 
     197            (b"c", (sAddr.host, sAddr.port))] 
     198 
     199        def cbClientSend(ignored): 
     200            if clientWrites: 
     201                nextClientWrite = server.packetReceived = defer.Deferred() 
     202                nextClientWrite.addCallback(cbClientSend) 
     203                client.transport.write(*clientWrites.pop(0)) 
     204                return nextClientWrite 
     205 
     206        # No one will ever call .errback on either of these Deferreds, 
     207        # but there is a non-trivial amount of test code which might 
     208        # cause them to fail somehow.  So fireOnOneErrback=True. 
     209        return defer.DeferredList([ 
     210                cbClientSend(None), 
     211                serverSend], fireOnOneErrback=True) 
     212 
     213 
     214    def cbSendsFinished(self, ignored, client, server, addressAddition=()): 
     215        cAddr = client.transport.getHost() 
     216        sAddr = server.transport.getHost() 
     217        clientAddr = (cAddr.host, cAddr.port) + addressAddition 
     218        serverAddr = (sAddr.host, sAddr.port) + addressAddition 
     219        self.assertEqual( 
     220            client.packets, 
     221            [(b"hello", serverAddr)]) 
     222        self.assertEqual( 
     223            server.packets, 
     224            [(b"a", clientAddr), 
     225             (b"b", clientAddr), 
     226             (b"c", clientAddr)]) 
     227 
     228 
     229    def cbFinished(self, ignored, port1, port2): 
     230        return defer.DeferredList([ 
     231            defer.maybeDeferred(port1.stopListening), 
     232            defer.maybeDeferred(port2.stopListening)], 
     233            fireOnOneErrback=True) 
     234 
     235 
     236 
    172237    def test_bindError(self): 
    173238        """ 
    174239        A L{CannotListenError} exception is raised when attempting to bind a 
     
    192257        d.addCallback(cbFinished) 
    193258        return d 
    194259 
    195  
    196     def test_sendPackets(self): 
     260    def performTransferServerClient(self, interface, addressAddition=()): 
    197261        """ 
    198         Datagrams can be sent with the transport's C{write} method and 
    199         received via the C{datagramReceived} callback method. 
     262        Helper method that creates UDP server and client, 
     263        connects the client to the server, sends some packets 
     264        and asserts addresses. 
    200265        """ 
    201266        server = Server() 
    202267        serverStarted = server.startedDeferred = defer.Deferred() 
    203         port1 = reactor.listenUDP(0, server, interface="127.0.0.1") 
     268        port1 = reactor.listenUDP(0, server, interface=interface) 
    204269 
    205270        client = GoodClient() 
    206271        clientStarted = client.startedDeferred = defer.Deferred() 
    207272 
    208         def cbServerStarted(ignored): 
    209             self.port2 = reactor.listenUDP(0, client, interface="127.0.0.1") 
    210             return clientStarted 
     273        d = serverStarted.addCallback(self.cbServerStarted, client, interface, 
     274                                      clientStarted) 
     275        d.addCallback(self.cbClientStarted, client, server, interface) 
     276        d.addCallback(self.cbSendsFinished, client, server, addressAddition) 
     277        d.addCallback(self.cbFinished, port1, self.port2) 
    211278 
    212         d = serverStarted.addCallback(cbServerStarted) 
     279        return d 
    213280 
    214         def cbClientStarted(ignored): 
    215             client.transport.connect("127.0.0.1", 
    216                                      server.transport.getHost().port) 
    217             cAddr = client.transport.getHost() 
    218             sAddr = server.transport.getHost() 
     281    def test_writeToIPv6Interface(self): 
     282        """ 
     283        Send packets to ipv6 address. 
     284        """ 
     285        return self.performTransferServerClient("::1", (0, 0)) 
    219286 
    220             serverSend = client.packetReceived = defer.Deferred() 
    221             server.transport.write(b"hello", (cAddr.host, cAddr.port)) 
    222287 
    223             clientWrites = [ 
    224                 (b"a",), 
    225                 (b"b", None), 
    226                 (b"c", (sAddr.host, sAddr.port))] 
     288    def test_sendPackets(self): 
     289        """ 
     290        Datagrams can be sent with the transport's C{write} method and 
     291        received via the C{datagramReceived} callback method. 
     292        """ 
     293        return self.performTransferServerClient("127.0.0.1") 
    227294 
    228             def cbClientSend(ignored): 
    229                 if clientWrites: 
    230                     nextClientWrite = server.packetReceived = defer.Deferred() 
    231                     nextClientWrite.addCallback(cbClientSend) 
    232                     client.transport.write(*clientWrites.pop(0)) 
    233                     return nextClientWrite 
    234295 
    235             # No one will ever call .errback on either of these Deferreds, 
    236             # but there is a non-trivial amount of test code which might 
    237             # cause them to fail somehow.  So fireOnOneErrback=True. 
    238             return defer.DeferredList([ 
    239                 cbClientSend(None), 
    240                 serverSend], 
    241                 fireOnOneErrback=True) 
    242  
    243         d.addCallback(cbClientStarted) 
    244  
    245         def cbSendsFinished(ignored): 
    246             cAddr = client.transport.getHost() 
    247             sAddr = server.transport.getHost() 
    248             self.assertEqual( 
    249                 client.packets, 
    250                 [(b"hello", (sAddr.host, sAddr.port))]) 
    251             clientAddr = (cAddr.host, cAddr.port) 
    252             self.assertEqual( 
    253                 server.packets, 
    254                 [(b"a", clientAddr), 
    255                  (b"b", clientAddr), 
    256                  (b"c", clientAddr)]) 
    257  
    258         d.addCallback(cbSendsFinished) 
    259  
    260         def cbFinished(ignored): 
    261             return defer.DeferredList([ 
    262                 defer.maybeDeferred(port1.stopListening), 
    263                 defer.maybeDeferred(self.port2.stopListening)], 
    264                 fireOnOneErrback=True) 
    265  
    266         d.addCallback(cbFinished) 
    267         return d 
    268  
    269  
    270296    def test_connectionRefused(self): 
    271297        """ 
    272298        A L{ConnectionRefusedError} exception is raised when a connection 
  • twisted/internet/address.py

     
    7777        IPv6 address; for example, "::1". 
    7878    @type host: C{str} 
    7979    """ 
     80    def __init__(self, type, host, port, flowInfo=0, scopeId=0, _bwHack=None): 
     81        _IPAddress.__init__(self, type, host, port) 
     82        self.flowInfo = flowInfo 
     83        self.scopeId = scopeId 
     84        if _bwHack is not None: 
     85            warnings.warn("twisted.internet.address.IPv4Address._bwHack " 
     86                          "is deprecated since Twisted 11.0", 
     87                          DeprecationWarning, stacklevel=2) 
    8088 
    8189 
    8290 
  • twisted/internet/udp.py

     
    8080        self.interface = interface 
    8181        self.setLogStr() 
    8282        self._connectedAddr = None 
     83        self.setAddressFamily() 
    8384 
    8485    def __repr__(self): 
    8586        if self._realPortNumber is not None: 
     
    207208        """ 
    208209        if self._connectedAddr: 
    209210            raise RuntimeError("already connected, reconnecting is not currently supported") 
    210         if not abstract.isIPAddress(host): 
     211        if not abstract.isIPAddress(host) and not abstract.isIPv6Address(host): 
    211212            raise ValueError("please pass only IP addresses, not domain names") 
    212213        self._connectedAddr = (host, port) 
    213214        self.socket.connect((host, port)) 
     
    252253        logPrefix = self._getLogPrefix(self.protocol) 
    253254        self.logstr = "%s (UDP)" % logPrefix 
    254255 
     256    def setAddressFamily(self): 
     257        """ 
     258        Resolve address family for the socket. 
     259        """ 
     260        if abstract.isIPv6Address(self.interface): 
     261            if not socket.has_ipv6: 
     262                raise RuntimeError("IPV6 is not supported, "+\ 
     263                                       "compile with ipv6 support") 
     264            self.addressFamily = socket.AF_INET6 
     265        elif abstract.isIPAddress(self.interface): 
     266            self.addressFamily = socket.AF_INET 
     267         
    255268 
     269 
    256270    def logPrefix(self): 
    257271        """ 
    258272        Return the prefix to log with. 
     
    262276 
    263277    def getHost(self): 
    264278        """ 
    265         Returns an IPv4Address. 
     279        Returns an IPv4Address or IPv6Address. 
    266280 
    267281        This indicates the address from which I am connecting. 
    268282        """ 
    269         return address.IPv4Address('UDP', *self.socket.getsockname()) 
     283        if self.addressFamily == socket.AF_INET: 
     284            return address.IPv4Address('UDP', *self.socket.getsockname()) 
     285        elif self.addressFamily == socket.AF_INET6: 
     286            return address.IPv6Address('UDP', *self.socket.getsockname()) 
    270287 
    271288 
    272289 
  • twisted/internet/test/test_udp.py

     
    2020from twisted.internet.defer import Deferred, maybeDeferred 
    2121from twisted.internet.interfaces import ( 
    2222    ILoggingContext, IListeningPort, IReactorUDP) 
    23 from twisted.internet.address import IPv4Address 
     23from twisted.internet.address import IPv4Address, IPv6Address 
    2424from twisted.internet.protocol import DatagramProtocol 
    2525 
    2626from twisted.internet.test.connectionmixins import (LogObserverMixin, 
     
    4848        """ 
    4949        return "(UDP Port %s Closed)" % (port.getHost().port,) 
    5050 
     51    def getIPv6ListeningPort(self, reactor, protocol): 
     52        """ 
     53        Get a UDP port binded to ipv6 interface. 
     54        """ 
     55        return reactor.listenUDP(0, protocol, "::1") 
    5156 
    5257 
     58 
    5359class DatagramTransportTestsMixin(LogObserverMixin): 
    5460    """ 
    5561    Mixin defining tests which apply to any port/datagram based transport. 
     
    162168        self.assertEqual( 
    163169            port.getHost(), IPv4Address('UDP', host, portNumber)) 
    164170 
     171    def test_getHostIPv6(self): 
     172        """ 
     173        L{IListeningPort.getHost} returns an L{IPv6Address} giving a 
     174        IPv6 address, the port number that the protocol is listening on 
     175        and the port number. 
     176        """ 
     177        reactor = self.buildReactor() 
     178        port = self.getIPv6ListeningPort(reactor, DatagramProtocol()) 
     179        addr = port.getHost() 
     180        self.assertEqual(addr.host, "::1") 
     181        self.assertEqual(addr.flowInfo, 0) 
     182        self.assertEqual(addr.scopeId, 0) 
    165183 
     184 
     185 
    166186    def test_logPrefix(self): 
    167187        """ 
    168188        Datagram transports implement L{ILoggingContext.logPrefix} to return a 
  • twisted/internet/test/test_address.py

     
    199199        """ 
    200200        return IPv6Address("TCP", "::2", 0) 
    201201 
     202    def buildWholeAddress(self): 
     203        """ 
     204        Like L{buldAddress}, but assigns all fields specific to 
     205        ipv6 (including flow info and scope id) 
     206        """ 
     207        return IPv6Address("TCP", "FE80::", 1, 1, None) 
    202208 
     209    def test_specificFields(self): 
     210        """ 
     211        Simply check if ipv6 specific fields exist. 
     212        """ 
     213        addr = self.buildWholeAddress() 
     214        self.assertTrue(hasattr(addr, "flowInfo")) 
     215        self.assertTrue(hasattr(addr, "scopeId")) 
    203216 
     217 
     218 
    204219class UNIXAddressTestCase(unittest.SynchronousTestCase, AddressTestCaseMixin): 
    205220    addressArgSpec = (("name", "%r"),) 
    206221