[pypy-svn] r21021 - in pypy/dist/pypy: module/_socket/test translator/c/test

[nik at codespeak.net]

Author: nik
Date: Sun Dec 11 11:36:05 2005
New Revision: 21021

Removed:
   pypy/dist/pypy/translator/c/test/test_ext__socket_conn.py
Modified:
   pypy/dist/pypy/module/_socket/test/echoserver.py
   pypy/dist/pypy/translator/c/test/test_ext__socket.py
Log:
(ale, nik)
test reorganisation. merged the two test files, we can now reuse the ipv4 tests for
ipv6 and unix sockets.


Modified: pypy/dist/pypy/module/_socket/test/echoserver.py
==============================================================================
--- pypy/dist/pypy/module/_socket/test/echoserver.py	(original)
+++ pypy/dist/pypy/module/_socket/test/echoserver.py	Sun Dec 11 11:36:05 2005
@@ -1,4 +1,4 @@
-import SocketServer
+import socket, SocketServer
 import sys, time
 
 # user-accessible port
@@ -7,6 +7,8 @@
 class EchoServer(SocketServer.TCPServer):
 
     def __init__(self, *args, **kwargs):
+        self.address_family = kwargs["address_family"]
+        del kwargs["address_family"]
         SocketServer.TCPServer.__init__(self, *args, **kwargs)
         self.stop = False
         
@@ -30,8 +32,8 @@
                 raise RuntimeError()
             self.wfile.write(client_string)
 
-def start_server():
-    server = EchoServer(("", PORT), EchoRequestHandler)
+def start_server(address_family=socket.AF_INET):
+    server = EchoServer(("", PORT), EchoRequestHandler, address_family=address_family)
     server.serve()    
 
 if __name__ == "__main__":

Modified: pypy/dist/pypy/translator/c/test/test_ext__socket.py
==============================================================================
--- pypy/dist/pypy/translator/c/test/test_ext__socket.py	(original)
+++ pypy/dist/pypy/translator/c/test/test_ext__socket.py	Sun Dec 11 11:36:05 2005
@@ -1,8 +1,12 @@
 import autopath
 import py
-import _socket
+import _socket, thread
 from pypy.translator.c.test.test_genc import compile
 from pypy.translator.translator import Translator
+from pypy.module._socket.test import echoserver
+
+HOST = "localhost"
+PORT = 8037
 
 def setup_module(mod):
     import pypy.module._socket.rpython.exttable   # for declare()/declaretype()
@@ -105,3 +109,36 @@
     f1 = compile(does_stuff, [str, int])
     for args in tests:
         py.test.raises(OSError, f1, *args)
+
+
+class TestConnectedIPv4:
+
+    family = _socket.AF_INET
+    
+    def setup_class(cls):    
+        thread.start_new_thread(echoserver.start_server, (),
+                                            {"address_family": cls.family})
+
+    def teardown_class(cls):
+        import telnetlib
+        tn = telnetlib.Telnet(HOST, PORT)
+        tn.write("shutdown\n")
+        tn.close()
+
+    def test_connect(self):
+        import os
+        from pypy.module._socket.rpython import rsocket
+        def does_stuff():
+            fd = rsocket.newsocket(self.family, _socket.SOCK_STREAM, 0)
+            rsocket.connect(fd, (HOST, PORT, 0, 0))
+            sockname = rsocket.getpeername(fd)
+            os.close(fd)
+            return sockname[1]
+        f1 = compile(does_stuff, [])
+        res = f1()
+        assert res == PORT
+
+class DONOT_TestConnectedIPv6(TestConnectedIPv4):
+    
+    disabled = not _socket.has_ipv6
+    family = _socket.AF_INET6