[telepathy-salut/master] add test-receive-file-ipv6.py

[Guillaume Desmottes]

---
 tests/twisted/avahi/test-receive-file-ipv6.py |   60 +++++++++++++++++++++++++
 1 files changed, 60 insertions(+), 0 deletions(-)
 create mode 100644 tests/twisted/avahi/test-receive-file-ipv6.py

diff --git a/tests/twisted/avahi/test-receive-file-ipv6.py b/tests/twisted/avahi/test-receive-file-ipv6.py
new file mode 100644
index 0000000..0f11d8c
--- /dev/null
+++ b/tests/twisted/avahi/test-receive-file-ipv6.py
@@ -0,0 +1,60 @@
+import avahi
+import urllib
+import BaseHTTPServer
+import SocketServer
+import socket
+
+from saluttest import exec_test
+from file_transfer_helper import ReceiveFileTest
+
+from avahitest import AvahiListener
+from xmppstream import connect_to_stream6
+
+from twisted.words.xish import domish
+
+class TestReceiveFileIPv6(ReceiveFileTest):
+    def _resolve_salut_presence(self):
+        AvahiListener(self.q).listen_for_service("_presence._tcp")
+        e = self.q.expect('service-added', name = self.self_handle_name,
+            protocol = avahi.PROTO_INET6)
+        service = e.service
+        service.resolve()
+
+        e = self.q.expect('service-resolved', service = service)
+        return str(e.pt), e.port
+
+    def connect_to_salut(self):
+        host, port = self._resolve_salut_presence()
+
+        self.outbound = connect_to_stream6(self.q, self.contact_name,
+            self.self_handle_name, host, port)
+
+        e = self.q.expect('connection-result')
+        assert e.succeeded, e.reason
+        self.q.expect('stream-opened', connection = self.outbound)
+
+    def send_ft_offer_iq(self):
+        iq = domish.Element((None, 'iq'))
+        iq['to'] = self.self_handle_name
+        iq['from'] = self.contact_name
+        iq['type'] = 'set'
+        iq['id'] = 'gibber-file-transfer-0'
+        query = iq.addElement(('jabber:iq:oob', 'query'))
+        url = 'http://[::1]:%u/gibber-file-transfer-0/%s' % \
+            (self.httpd.server_port, urllib.quote(self.file.name))
+        url_node = query.addElement('url', content=url)
+        url_node['type'] = 'file'
+        url_node['size'] = str(self.file.size)
+        url_node['mimeType'] = self.file.content_type
+        query.addElement('desc', content=self.file.description)
+        self.outbound.send(iq)
+
+    def _get_http_server_class(self):
+        class HTTPServer6(SocketServer.ThreadingMixIn, BaseHTTPServer.HTTPServer):
+            address_family = getattr(socket, 'AF_INET6', None)
+
+        return HTTPServer6
+
+if __name__ == '__main__':
+    test = TestReceiveFileIPv6()
+    exec_test(test.test)
-- 
1.5.6.5