Commits

Amaury Forgeot d'Arc  committed ac389af

Implement BytesIO.getbuffer()

  • Participants
  • Parent commits 8207e13
  • Branches py3k

Comments (0)

Files changed (3)

File lib-python/3.2/test/test_memoryio.py

         self.assertEqual(bytes(buf), b"1234567890")
         # Trying to change the size of the BytesIO while a buffer is exported
         # raises a BufferError.
-        self.assertRaises(BufferError, memio.write, b'x' * 100)
-        self.assertRaises(BufferError, memio.truncate)
+        if support.check_impl_detail(pypy=False):
+            # PyPy export buffers differently, and allows reallocation
+            # of the underlying object.
+            self.assertRaises(BufferError, memio.write, b'x' * 100)
+            self.assertRaises(BufferError, memio.truncate)
         # Mutating the buffer updates the BytesIO
         buf[3:6] = b"abc"
         self.assertEqual(bytes(buf), b"123abc7890")

File pypy/module/_io/interp_bytesio.py

     TypeDef, generic_new_descr, GetSetProperty)
 from pypy.interpreter.gateway import interp2app, unwrap_spec
 from pypy.interpreter.error import OperationError, operationerrfmt
+from pypy.interpreter.buffer import RWBuffer
 from rpython.rlib.rarithmetic import r_longlong
 from pypy.module._io.interp_bufferedio import W_BufferedIOBase
 from pypy.module._io.interp_iobase import convert_size
         builder.append(buffer[i])
     return builder.build()
 
+class BytesIOBuffer(RWBuffer):
+    def __init__(self, w_bytesio):
+        self.w_bytesio = w_bytesio
+
+    def getlength(self):
+        return int(self.w_bytesio.string_size)
+
+    def getitem(self, index):
+        return self.w_bytesio.buf[index]
+
+    def setitem(self, index, char):
+        self.w_bytesio.buf[index] = char
+
+
 class W_BytesIO(W_BufferedIOBase):
     def __init__(self, space):
         W_BufferedIOBase.__init__(self, space)
 
         return space.wrap(size)
 
+    def getbuffer_w(self, space):
+        return space.wrap(BytesIOBuffer(self))
+
     def getvalue_w(self, space):
         self._check_closed(space)
         return space.wrapbytes(buffer2string(self.buf, 0, self.string_size))
     readinto = interp2app(W_BytesIO.readinto_w),
     write = interp2app(W_BytesIO.write_w),
     truncate = interp2app(W_BytesIO.truncate_w),
+    getbuffer = interp2app(W_BytesIO.getbuffer_w),
     getvalue = interp2app(W_BytesIO.getvalue_w),
     seek = interp2app(W_BytesIO.seek_w),
     tell = interp2app(W_BytesIO.tell_w),

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

         b = _io.BytesIO(b"hello")
         b.close()
         raises(ValueError, b.readinto, bytearray(b"hello"))
+
+    def test_getbuffer(self):
+        import _io
+        memio = _io.BytesIO(b"1234567890")
+        buf = memio.getbuffer()
+        assert bytes(buf) == b"1234567890"
+        memio.seek(5)
+        buf = memio.getbuffer()
+        assert bytes(buf) == b"1234567890"
+        # Mutating the buffer updates the BytesIO
+        buf[3:6] = b"abc"
+        assert bytes(buf) == b"123abc7890"
+        assert memio.getvalue() == b"123abc7890"
+        # After the buffer gets released, we can resize the BytesIO again
+        del buf
+        memio.truncate()
+