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

File fix-ipv6-literals-udp.3.patch, 8.8 KB (added by marto1_, 15 months ago)
  • 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 
  • 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/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/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 test_writeToIPv6Interface(self): 
     182        """ 
     183        Send packets to ipv6 address. 
     184        """ 
     185        #FIXME Probably a more elegant way of testing this should be done 
     186        server = Server() 
     187        serverStarted = server.startedDeferred = defer.Deferred() 
     188        port1 = reactor.listenUDP(0, server, interface="::1") 
     189 
     190        client = GoodClient() 
     191        clientStarted = client.startedDeferred = defer.Deferred() 
     192 
     193        def cbServerStarted(ignored): 
     194            self.port2 = reactor.listenUDP(0, client, interface="::1") 
     195            return clientStarted 
     196 
     197        d = serverStarted.addCallback(cbServerStarted) 
     198 
     199        def cbClientStarted(ignored): 
     200            client.transport.connect("::1", 
     201                                     server.transport.getHost().port) 
     202            cAddr = client.transport.getHost() 
     203            sAddr = server.transport.getHost() 
     204 
     205            serverSend = client.packetReceived = defer.Deferred() 
     206            server.transport.write(b"hello", (cAddr.host, cAddr.port)) 
     207 
     208            clientWrites = [ 
     209                (b"a",), 
     210                (b"b", None), 
     211                (b"c", (sAddr.host, sAddr.port))] 
     212 
     213            def cbClientSend(ignored): 
     214                if clientWrites: 
     215                    nextClientWrite = server.packetReceived = defer.Deferred() 
     216                    nextClientWrite.addCallback(cbClientSend) 
     217                    client.transport.write(*clientWrites.pop(0)) 
     218                    return nextClientWrite 
     219 
     220            # No one will ever call .errback on either of these Deferreds, 
     221            # but there is a non-trivial amount of test code which might 
     222            # cause them to fail somehow.  So fireOnOneErrback=True. 
     223            return defer.DeferredList([ 
     224                cbClientSend(None), 
     225                serverSend], 
     226                fireOnOneErrback=True) 
     227 
     228        d.addCallback(cbClientStarted) 
     229 
     230        def cbSendsFinished(ignored): 
     231            cAddr = client.transport.getHost() 
     232            sAddr = server.transport.getHost() 
     233            self.assertEqual( 
     234                client.packets, 
     235                [(b"hello", (sAddr.host, sAddr.port, 0, 0))]) 
     236            clientAddr = (cAddr.host, cAddr.port, 0, 0) 
     237            self.assertEqual( 
     238                server.packets, 
     239                [(b"a", clientAddr), 
     240                 (b"b", clientAddr), 
     241                 (b"c", clientAddr)]) 
     242 
     243        d.addCallback(cbSendsFinished) 
     244 
     245        def cbFinished(ignored): 
     246            return defer.DeferredList([ 
     247                defer.maybeDeferred(port1.stopListening), 
     248                defer.maybeDeferred(self.port2.stopListening)], 
     249                fireOnOneErrback=True) 
     250 
     251        d.addCallback(cbFinished) 
     252        return d 
     253 
    172254    def test_bindError(self): 
    173255        """ 
    174256        A L{CannotListenError} exception is raised when attempting to bind a