[telepathy-salut/master] ipv6.py: add client side code

[Guillaume Desmottes]

---
 tests/twisted/ipv6.py |  113 +++++++++++++++++++++++++++++++++++++++++++++++++
 1 files changed, 113 insertions(+), 0 deletions(-)

diff --git a/tests/twisted/ipv6.py b/tests/twisted/ipv6.py
index 9b9feea..694b83f 100644
--- a/tests/twisted/ipv6.py
+++ b/tests/twisted/ipv6.py
@@ -31,6 +31,11 @@ def listenTCP6(port,factory,backlog=5,interface='::',reactor = None):
 # stolen from http://twistedmatrix.com/trac/attachment/ticket/3014/ipv6.2.patch
 from zope.interface import implements
 from twisted.internet.interfaces import IAddress
+from twisted.internet import base, address, error
+from twisted.internet.tcp import Client
+from twisted.python.util import unsignedID
+import types
+
 class IPv6Address(object):
     """
     Object representing an IPv6 socket endpoint.
@@ -65,3 +70,111 @@ class IPv6Address(object):
     def __str__(self):
         return 'IPv6Address(%s, %r, %d, %s, %s, %s)' % (self.type, self.host,
           self.port, self.flowinfo, self.scopeid)
+
+class Client6(Client):
+    """
+    A TCP6 client.
+    """
+    addressFamily = socket.AF_INET6
+
+    def __init__(self, host, port, bindAddress, connector, reactor=None,
+                 flowinfo=0, scopeid=0):
+        Client.__init__(self, host, port, bindAddress, connector, reactor)
+        self.addr = (host, port, flowinfo, scopeid)
+
+    def resolveAddress(self):
+        """
+        Lookup the IPv6 address for self.addr[0] if necessary, then set
+        self.realAddress to that IPv6 address.
+        """
+        if isIPv6Address(self.addr[0]):
+            self._setRealAddress(self.addr[0])
+        else:
+            d = self.reactor.resolve(self.addr[0])
+            d.addCallbacks(self._setRealAddress, self.failIfNotConnected)
+
+    def _setRealAddress(self, address):
+        """
+        Set self.realAddress[0] to address.  Set the remaining parts of 
+        self.realAddress to the corresponding parts of self.addr.
+        """
+        self.realAddress = (address, self.addr[1], self.addr[2], self.addr[3])
+        self.doConnect()
+
+    def getHost(self):
+        """
+        Returns an IPv6Address.
+
+        This indicates the address from which I am connecting.
+        """
+        return address.IPv6Address('TCP', *(self.socket.getsockname()))
+
+    def getPeer(self):
+        """
+        Returns an IPv6Address.
+
+        This indicates the address that I am connected to.
+        """
+        return IPv6Address('TCP', *(self.addr))
+
+    def __repr__(self):
+        s = '<%s to %s at %x>' % (self.__class__, self.addr, unsignedID(self))
+        return s
+
+class Connector6(base.BaseConnector):
+    """
+    IPv6 implementation of connector
+
+    @ivar flowinfo An integer representing the sockaddr-in6 flowinfo
+    @ivar scopeid An integer representing the sockaddr-in6 scopeid
+    """
+
+    def __init__(self, host, port, factory, timeout, bindAddress,
+                 reactor=None, flowinfo=0, scopeid=0):
+        self.host = host
+        if isinstance(port, types.StringTypes):
+            try:
+                port = socket.getservbyname(port, 'tcp')
+            except socket.error, e:
+                raise error.ServiceNameUnknownError(string="%s (%r)" % (e, port))
+        self.port = port
+        self.bindAddress = bindAddress
+        self.flowinfo = flowinfo
+        self.scopeid = scopeid
+        base.BaseConnector.__init__(self, factory, timeout, reactor)
+
+    def _makeTransport(self):
+        """
+        Build and return a TCP6 client for the connector's transport.
+        """
+        return Client6(self.host, self.port, self.bindAddress, self,
+                      self.reactor, self.flowinfo, self.scopeid)
+
+    def getDestination(self):
+        """
+        @see twisted.internet.interfaces.IConnector.getDestination
+        """
+        return address.IPv6Address('TCP', self.host, self.port, self.flowinfo,
+                                   self.scopeid)
+
+def connectTCP6(reactor, host, port, factory, timeout=30, bindAddress=None,
+        flowinfo=0, scopeid=0):
+        """
+        @see: twisted.internet.interfaces.IReactorTCP.connectTCP6
+        """
+        c = Connector6(host, port, factory, timeout, bindAddress, reactor,
+                           flowinfo, scopeid)
+        c.connect()
+        return c
+
+def isIPv6Address(ip):
+    """
+    Return True iff ip is a valid bare IPv6 address.
+
+    Return False for 'enhanced' IPv6 addresses like '::1%lo' and '::1/128'
+    """
+    try:
+        socket.inet_pton(socket.AF_INET6, ip)
+    except (ValueError, socket.error):
+        return False
+    return True
-- 
1.5.6.5