[pypy-svn] r33499 - in pypy/dist/pypy/module/rsocket: . test
ac at codespeak.net
ac at codespeak.net
Fri Oct 20 14:38:37 CEST 2006
Author: ac
Date: Fri Oct 20 14:38:35 2006
New Revision: 33499
Modified:
pypy/dist/pypy/module/rsocket/interp_socket.py
pypy/dist/pypy/module/rsocket/rsocket.py
pypy/dist/pypy/module/rsocket/test/test_rsocket.py
pypy/dist/pypy/module/rsocket/test/test_sock_app.py
Log:
Add getsockopt() and setsockopt().
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 Fri Oct 20 14:38:35 2006
@@ -110,6 +110,44 @@
return addr.as_object(space)
getsockname_w.unwrap_spec = ['self', ObjSpace]
+ def getsockopt_w(self, space, level, optname, w_buflen=NoneNotWrapped):
+ """getsockopt(level, option[, buffersize]) -> value
+
+ Get a socket option. See the Unix manual for level and option.
+ If a nonzero buffersize argument is given, the return value is a
+ string of that length; otherwise it is an integer.
+ """
+ if w_buflen is None:
+ try:
+ return space.wrap(self.getsockopt_int(level, optname))
+ except SocketError, e:
+ raise converted_error(space, e)
+ buflen = space.int_w(w_buflen)
+ return space.wrap(self.getsockopt(level, optname, buflen))
+ getsockopt_w.unwrap_spec = ['self', ObjSpace, int, int, W_Root]
+
+ def setsockopt_w(self, space, level, optname, w_optval):
+ """setsockopt(level, option, value)
+
+ Set a socket option. See the Unix manual for level and option.
+ The value argument can either be an integer or a string.
+ """
+ try:
+ optval = space.int_w(w_optval)
+ except:
+ optval = space.str_w(w_optval)
+ try:
+ self.setsockopt(level, optname, optval)
+ except SocketError, e:
+ raise converted_error(space, e)
+ return
+ try:
+ self.setsockopt_int(level, optname, optval)
+ except SocketError, e:
+ raise converted_error(space, e)
+
+ setsockopt_w.unwrap_spec = ['self', ObjSpace, int, int, W_Root]
+
def listen_w(self, space, backlog):
"""listen(backlog)
@@ -255,10 +293,10 @@
socketmethodnames = """
accept bind close connect connect_ex fileno
-getpeername getsockname listen recv
+getpeername getsockname getsockopt listen recv
recvfrom send sendall sendto setblocking
-shutdown
-""".split() # dup makefile gettimeout settimeout getsockopt setsockopt
+setsockopt shutdown
+""".split() # dup makefile gettimeout settimeout
socketmethods = {}
for methodname in socketmethodnames:
method = getattr(W_RSocket, methodname + '_w')
Modified: pypy/dist/pypy/module/rsocket/rsocket.py
==============================================================================
--- pypy/dist/pypy/module/rsocket/rsocket.py (original)
+++ pypy/dist/pypy/module/rsocket/rsocket.py Fri Oct 20 14:38:35 2006
@@ -588,6 +588,36 @@
if res < 0:
raise self.error_handler()
+ def getsockopt_int(self, level, option):
+ flag = _c.c_int()
+ flagsize = _c.socklen_t()
+ flagsize.value = _c.sizeof(flag)
+ res = _c.socketgetsockopt(self.fd, level, option,
+ byref(flag), byref(flagsize))
+ if res < 0:
+ raise self.error_handler()
+ return flag.value
+
+ def getsockopt(self, level, option, maxlen):
+ buf = _c.create_string_buffer(maxlen)
+ bufsize = _c.socklen_t()
+ bufsize.value = maxlen
+ res = _c.socketgetsockopt(self.fd, level, option, byref(buf), byref(bufsize))
+ if res < 0:
+ raise self.error_handler()
+ return buf.raw[:bufsize.value]
+
+ def setsockopt_int(self, level, option, value):
+ flag = _c.c_int(value)
+ res = _c.socketsetsockopt(self.fd, level, option,
+ byref(flag), _c.sizeof(flag))
+ if res < 0:
+ raise self.error_handler()
+
+ def setsockopt(self, level, option, value):
+ res = _c.socketsetsockopt(self.fd, level, option, value, len(value))
+ if res < 0:
+ raise self.error_handler()
# ____________________________________________________________
def make_socket(fd, family, type, proto, SocketClass=RSocket):
Modified: pypy/dist/pypy/module/rsocket/test/test_rsocket.py
==============================================================================
--- pypy/dist/pypy/module/rsocket/test/test_rsocket.py (original)
+++ pypy/dist/pypy/module/rsocket/test/test_rsocket.py Fri Oct 20 14:38:35 2006
@@ -182,3 +182,25 @@
s = RSocket()
err = s.connect_ex(s.getsockname()) # should not work
assert err in (errno.ECONNREFUSED, errno.EADDRNOTAVAIL)
+
+
+def test_getsetsockopt():
+ # A socket sould start with reuse == 0
+ s = RSocket(_c.AF_INET, _c.SOCK_STREAM)
+ reuse = s.getsockopt_int(_c.SOL_SOCKET, _c.SO_REUSEADDR)
+ assert reuse == 0
+ s.setsockopt_int(_c.SOL_SOCKET, _c.SO_REUSEADDR, 1)
+ reuse = s.getsockopt_int(_c.SOL_SOCKET, _c.SO_REUSEADDR)
+ assert reuse != 0
+ # Test string case
+ s = RSocket(_c.AF_INET, _c.SOCK_STREAM)
+ reusestr = s.getsockopt(_c.SOL_SOCKET, _c.SO_REUSEADDR, sizeof(_c.c_int))
+ reuseptr = _c.cast(_c.c_char_p(reusestr), _c.POINTER(_c.c_int))
+ assert reuseptr[0] == 0
+ optval = _c.c_int(1)
+ optvalp = _c.cast(_c.pointer(optval), _c.POINTER(_c.c_char))
+ optstr = optvalp[:sizeof(_c.c_int)]
+ s.setsockopt(_c.SOL_SOCKET, _c.SO_REUSEADDR, optstr)
+ reusestr = s.getsockopt(_c.SOL_SOCKET, _c.SO_REUSEADDR, sizeof(_c.c_int))
+ reuseptr = _c.cast(_c.c_char_p(reusestr), _c.POINTER(_c.c_int))
+ assert reuseptr[0] != 0
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 Fri Oct 20 14:38:35 2006
@@ -353,3 +353,27 @@
import socket
s = socket.socket()
+
+ def test_getsetsockopt(self):
+ import _socket as socket
+ import struct
+ # A socket sould start with reuse == 0
+ s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+ reuse = s.getsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR)
+ assert reuse == 0
+ s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+ reuse = s.getsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR)
+ assert reuse != 0
+ # String case
+ intsize = struct.calcsize('i')
+ s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+ reusestr = s.getsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR,
+ intsize)
+ (reuse,) = struct.unpack('i', reusestr)
+ assert reuse == 0
+ reusestr = struct.pack('i', 1)
+ s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, reusestr)
+ reusestr = s.getsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR,
+ intsize)
+ (reuse,) = struct.unpack('i', reusestr)
+ assert reuse != 0
More information about the pypy-svn
mailing list