[pypy-svn] r42136 - in pypy/dist/pypy: module/rsocket module/rsocket/test rlib rlib/test

[afa at codespeak.net]

Author: afa
Date: Tue Apr 17 21:46:41 2007
New Revision: 42136

Modified:
   pypy/dist/pypy/module/rsocket/interp_socket.py
   pypy/dist/pypy/module/rsocket/test/test_sock_app.py
   pypy/dist/pypy/rlib/rsocket.py
   pypy/dist/pypy/rlib/test/test_rsocket.py
Log:
rsocket progress on win32: hide socket.dup, add implementation of inet_aton


Modified: pypy/dist/pypy/module/rsocket/interp_socket.py
==============================================================================
--- pypy/dist/pypy/module/rsocket/interp_socket.py	(original)
+++ pypy/dist/pypy/module/rsocket/interp_socket.py	Tue Apr 17 21:46:41 2007
@@ -474,6 +474,11 @@
 recv recvfrom send sendall sendto setblocking
 setsockopt settimeout shutdown _reuse _drop
 """.split()
+# Remove non-implemented methods
+for name in ('dup',):
+    if not hasattr(RSocket, name):
+        socketmethodnames.remove(name)
+
 socketmethods = {}
 for methodname in socketmethodnames:
     method = getattr(W_RSocket, methodname + '_w')

Modified: pypy/dist/pypy/module/rsocket/test/test_sock_app.py
==============================================================================
--- pypy/dist/pypy/module/rsocket/test/test_sock_app.py	(original)
+++ pypy/dist/pypy/module/rsocket/test/test_sock_app.py	Tue Apr 17 21:46:41 2007
@@ -377,6 +377,8 @@
 
     def test_dup(self):
         import _socket as socket
+        if not hasattr(socket.socket, 'dup'):
+            skip('No dup() on this platform')
         s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
         s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
         s.bind(('localhost', 50007))

Modified: pypy/dist/pypy/rlib/rsocket.py
==============================================================================
--- pypy/dist/pypy/rlib/rsocket.py	(original)
+++ pypy/dist/pypy/rlib/rsocket.py	Tue Apr 17 21:46:41 2007
@@ -613,14 +613,15 @@
             return _c.geterrno()
         return res
 
-    def dup(self, SocketClass=None):
-        if SocketClass is None:
-            SocketClass = RSocket
-        fd = _c.dup(self.fd)
-        if fd < 0:
-            raise self.error_handler()
-        return make_socket(fd, self.family, self.type, self.proto,
-                           SocketClass=SocketClass)
+    if hasattr(_c, 'dup'):
+        def dup(self, SocketClass=None):
+            if SocketClass is None:
+                SocketClass = RSocket
+            fd = _c.dup(self.fd)
+            if fd < 0:
+                raise self.error_handler()
+            return make_socket(fd, self.family, self.type, self.proto,
+                               SocketClass=SocketClass)
         
     def fileno(self):
         fd = self.fd
@@ -1011,13 +1012,25 @@
         raise GAIError(error)
     return host.value, serv.value
 
-def inet_aton(ip):
-    "IPv4 dotted string -> packed 32-bits string"
-    buf = create_string_buffer(sizeof(_c.in_addr))
-    if _c.inet_aton(ip, cast(buf, POINTER(_c.in_addr))):
+if hasattr(_c, 'inet_aton'):
+    def inet_aton(ip):
+        "IPv4 dotted string -> packed 32-bits string"
+        buf = create_string_buffer(sizeof(_c.in_addr))
+        if _c.inet_aton(ip, cast(buf, POINTER(_c.in_addr))):
+            return buf.raw
+        else:
+            raise RSocketError("illegal IP address string passed to inet_aton")
+else:
+    def inet_aton(ip):
+        "IPv4 dotted string -> packed 32-bits string"
+        if ip == "255.255.255.255":
+            return "\xff\xff\xff\xff"
+        packed_addr = _c.inet_addr(ip)
+        if _c.c_long(packed_addr).value == INADDR_NONE:
+            raise RSocketError("illegal IP address string passed to inet_aton")
+        buf = copy_buffer(cast(pointer(c_ulong(packed_addr)),
+                               POINTER(c_char)), 4)
         return buf.raw
-    else:
-        raise RSocketError("illegal IP address string passed to inet_aton")
 
 def inet_ntoa(packed):
     "packet 32-bits string -> IPv4 dotted string"

Modified: pypy/dist/pypy/rlib/test/test_rsocket.py
==============================================================================
--- pypy/dist/pypy/rlib/test/test_rsocket.py	(original)
+++ pypy/dist/pypy/rlib/test/test_rsocket.py	Tue Apr 17 21:46:41 2007
@@ -257,6 +257,14 @@
     s2 = s.dup()
     assert s.fileno() != s2.fileno()
     assert s.getsockname().eq(s2.getsockname())
+
+def test_inet_aton():
+    assert inet_aton('1.2.3.4') == '\x01\x02\x03\x04'
+    assert inet_aton('127.0.0.1') == '\x7f\x00\x00\x01'
+    tests = ["127.0.0.256", "127.0.0.255555555555555555", "127.2b.0.0",
+        "127.2.0.0.1", "127.2..0"]
+    for ip in tests:
+        py.test.raises(SocketError, inet_aton, ip)
     
 class TestTCP:
     PORT = 50007