[pypy-commit] pypy py3.5: Implement os.get_blocking() and os.set_blocking()

[rlamy]

Author: Ronan Lamy <ronan.lamy at gmail.com>
Branch: py3.5
Changeset: r87951:675ae4c0f6d1
Date: 2016-10-26 20:03 +0100
http://bitbucket.org/pypy/pypy/changeset/675ae4c0f6d1/

Log:	Implement os.get_blocking() and os.set_blocking()

diff --git a/pypy/module/posix/__init__.py b/pypy/module/posix/__init__.py
--- a/pypy/module/posix/__init__.py
+++ b/pypy/module/posix/__init__.py
@@ -203,6 +203,8 @@
 
     if not rposix._WIN32:
         interpleveldefs['sync'] = 'interp_posix.sync'
+        interpleveldefs['get_blocking'] = 'interp_posix.get_blocking'
+        interpleveldefs['set_blocking'] = 'interp_posix.set_blocking'
 
     def startup(self, space):
         from pypy.module.posix import interp_posix
diff --git a/pypy/module/posix/interp_posix.py b/pypy/module/posix/interp_posix.py
--- a/pypy/module/posix/interp_posix.py
+++ b/pypy/module/posix/interp_posix.py
@@ -2217,3 +2217,16 @@
     if count <= 0:
         return space.w_None
     return space.wrap(count)
+
+ at unwrap_spec(fd=c_int)
+def get_blocking(space, fd):
+    return space.newbool(rposix.get_status_flags(fd) & rposix.O_NONBLOCK == 0)
+
+ at unwrap_spec(fd=c_int, blocking=int)
+def set_blocking(space, fd, blocking):
+    flags = rposix.get_status_flags(fd)
+    if blocking:
+        flags &= ~rposix.O_NONBLOCK
+    else:
+        flags |= rposix.O_NONBLOCK
+    rposix.set_status_flags(fd, flags)
diff --git a/pypy/module/posix/test/test_posix2.py b/pypy/module/posix/test/test_posix2.py
--- a/pypy/module/posix/test/test_posix2.py
+++ b/pypy/module/posix/test/test_posix2.py
@@ -1140,6 +1140,17 @@
         def test_sync(self):
             self.posix.sync()   # does not raise
 
+        def test_blocking(self):
+            posix = self.posix
+            fd = posix.open(self.path, posix.O_RDONLY)
+            assert posix.get_blocking(fd) is True
+            posix.set_blocking(fd, False)
+            assert posix.get_blocking(fd) is False
+            posix.set_blocking(fd, True)
+            assert posix.get_blocking(fd) is True
+            posix.close(fd)
+
+
     def test_urandom(self):
         os = self.posix
         s = os.urandom(5)