1. Pypy
  2. Untitled project
  3. pypy

Commits

Amaury Forgeot d'Arc  committed 0e4c3da

BlockingIOError is now in the exceptions module.

  • Participants
  • Parent commits 4deadf4
  • Branches py3.3

Comments (0)

Files changed (7)

File pypy/interpreter/baseobjspace.py

View file
  • Ignore whitespace
     'BaseException',
     'BufferError',
     'BytesWarning',
+    'BlockingIOError',
     'DeprecationWarning',
     'EOFError',
     'EnvironmentError',

File pypy/module/_io/__init__.py

View file
  • Ignore whitespace
 
     interpleveldefs = {
         'DEFAULT_BUFFER_SIZE': 'space.wrap(interp_iobase.DEFAULT_BUFFER_SIZE)',
-        'BlockingIOError': 'interp_io.W_BlockingIOError',
+        'BlockingIOError': 'space.w_BlockingIOError',
         'UnsupportedOperation':
             'space.fromcache(interp_io.Cache).w_unsupportedoperation',
         '_IOBase': 'interp_iobase.W_IOBase',

File pypy/module/_io/interp_bufferedio.py

View file
  • Ignore whitespace
 from pypy.module._io.interp_iobase import (
     W_IOBase, DEFAULT_BUFFER_SIZE, convert_size, trap_eintr,
     check_readable_w, check_writable_w, check_seekable_w)
-from pypy.module._io.interp_io import W_BlockingIOError
 from rpython.rlib import rthread
 
 STATE_ZERO, STATE_OK, STATE_DETACHED = range(3)
 
 
 def make_write_blocking_error(space, written):
-    w_type = space.gettypeobject(W_BlockingIOError.typedef)
     w_value = space.call_function(
-        w_type,
+        space.w_BlockingIOError,
         space.wrap(rposix.get_errno()),
         space.wrap("write could not complete without blocking"),
         space.wrap(written))
-    return OperationError(w_type, w_value)
+    return OperationError(space.w_BlockingIOError, w_value)
 
 
 class TryLock(object):
             try:
                 self._writer_flush_unlocked(space)
             except OperationError, e:
-                if not e.match(space, space.gettypeobject(
-                    W_BlockingIOError.typedef)):
+                if not e.match(space, space.w_BlockingIOError):
                     raise
-                w_exc = e.get_w_value(space)
-                assert isinstance(w_exc, W_BlockingIOError)
                 if self.readable:
                     self._reader_reset_buf()
                 # Make some place by shifting the buffer

File pypy/module/_io/interp_io.py

View file
  • Ignore whitespace
             "io.UnsupportedOperation",
             space.newtuple([space.w_ValueError, space.w_IOError]))
 
-class W_BlockingIOError(W_IOError):
-    def __init__(self, space):
-        W_IOError.__init__(self, space)
-        self.written = 0
-
-    @unwrap_spec(written=int)
-    def descr_init(self, space, w_errno, w_strerror, written=0):
-        W_IOError.descr_init(self, space, [w_errno, w_strerror])
-        self.written = written
-
-W_BlockingIOError.typedef = TypeDef(
-    'BlockingIOError', W_IOError.typedef,
-    __doc__ = ("Exception raised when I/O would block on a non-blocking "
-               "I/O stream"),
-    __new__  = generic_new_descr(W_BlockingIOError),
-    __init__ = interp2app(W_BlockingIOError.descr_init),
-    characters_written = interp_attrproperty('written', W_BlockingIOError),
-    )
-
 DEFAULT_BUFFER_SIZE = 8 * 1024
 
 @unwrap_spec(mode=str, buffering=int,

File pypy/module/_io/test/test_io.py

View file
  • Ignore whitespace
         import _io
         try:
             raise _io.BlockingIOError(42, "test blocking", 123)
-        except IOError as e:
+        except OSError as e:
             assert isinstance(e, _io.BlockingIOError)
             assert e.errno == 42
             assert e.strerror == "test blocking"

File pypy/module/exceptions/interp_exceptions.py

View file
  • Ignore whitespace
 """
 
 from pypy.interpreter.baseobjspace import W_Root
-from pypy.interpreter.typedef import (TypeDef, GetSetProperty, descr_get_dict,
-    descr_set_dict, descr_del_dict)
-from pypy.interpreter.gateway import interp2app
+from pypy.interpreter.typedef import (
+    TypeDef, GetSetProperty, interp_attrproperty,
+    descr_get_dict, descr_set_dict, descr_del_dict)
+from pypy.interpreter.gateway import interp2app, unwrap_spec
 from pypy.interpreter.error import OperationError, setup_context
 from pypy.interpreter.pytraceback import PyTraceback, check_traceback
 from rpython.rlib import rwin32
     )
 
 # Various OSError subclasses added in Python 3.3
-W_BlockingIOError = _new_exception(
-    "BlockingIOError", W_OSError, "I/O operation would block.")
+class W_BlockingIOError(W_OSError):
+    "I/O operation would block."
+
+    def __init__(self, space):
+        W_OSError.__init__(self, space)
+        self.written = -1
+
+    def descr_init(self, space, args_w):
+        W_OSError.descr_init(self, space, args_w)
+        # BlockingIOError's 3rd argument can be the number of
+        # characters written.
+        if len(args_w) >= 3:
+            try:
+                written = space.int_w(args_w[2])
+            except OperationError:
+                pass
+            else:
+                self.written = written
+
+    def descr_get_written(self, space):
+        if self.written == -1:
+            raise OperationError(space.w_AttributeError,
+                                 space.wrap("characters_written"))
+        return space.wrap(self.written)
+
+    def descr_set_written(self, space, w_written):
+        self.written = space.int_w(w_written)
+
+
+W_BlockingIOError.typedef = TypeDef(
+    'BlockingIOError', W_OSError.typedef,
+    __doc__ = ("Exception raised when I/O would block on a non-blocking "
+               "I/O stream"),
+    __new__  = _new(W_BlockingIOError),
+    __init__ = interp2app(W_BlockingIOError.descr_init),
+    characters_written = GetSetProperty(W_BlockingIOError.descr_get_written,
+                                        W_BlockingIOError.descr_set_written),
+    )
+
 W_ConnectionError = _new_exception(
     "ConnectionError", W_OSError, "Connection error.")
 W_ChildProcessError = _new_exception(

File pypy/module/exceptions/test/test_exc.py

View file
  • Ignore whitespace
         assert ImportError("message", path="y").path == "y"
         raises(TypeError, ImportError, invalid="z")
 
+    def test_blockingioerror(self):
+        args = ("a", "b", "c", "d", "e")
+        for n in range(6):
+            e = BlockingIOError(*args[:n])
+            raises(AttributeError, getattr, e, 'characters_written')
+        e = BlockingIOError("a", "b", 3)
+        assert e.characters_written == 3
+        e.characters_written = 5
+        assert e.characters_written == 5
+
+