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

File fix-ipv6-literals-udp.3.patch, 8.8 KB (added by marto1_, 3 years 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