[pypy-svn] r78402 - in pypy/branch/fast-forward/pypy/module/_io: . test

[afa at codespeak.net]

Author: afa
Date: Thu Oct 28 16:04:19 2010
New Revision: 78402

Modified:
   pypy/branch/fast-forward/pypy/module/_io/interp_bufferedio.py
   pypy/branch/fast-forward/pypy/module/_io/test/test_bufferedio.py
Log:
BufferedReader.readinto()


Modified: pypy/branch/fast-forward/pypy/module/_io/interp_bufferedio.py
==============================================================================
--- pypy/branch/fast-forward/pypy/module/_io/interp_bufferedio.py	(original)
+++ pypy/branch/fast-forward/pypy/module/_io/interp_bufferedio.py	Thu Oct 28 16:04:19 2010
@@ -38,6 +38,19 @@
     def detach_w(self, space):
         self._unsupportedoperation(space, "detach")
 
+    @unwrap_spec('self', ObjSpace, W_Root)
+    def readinto_w(self, space, w_buffer):
+        rwbuffer = space.rwbuffer_w(w_buffer)
+        length = rwbuffer.getlength()
+        w_data = space.call_method(self, "read", space.wrap(length))
+
+        if not space.isinstance_w(w_data, space.w_str):
+            raise OperationError(space.w_TypeError, space.wrap(
+                "read() should return bytes"))
+        data = space.str_w(w_data)
+        rwbuffer.setslice(0, data)
+        return space.wrap(len(data))
+
 W_BufferedIOBase.typedef = TypeDef(
     '_BufferedIOBase', W_IOBase.typedef,
     __new__ = generic_new_descr(W_BufferedIOBase),
@@ -45,6 +58,7 @@
     read1 = interp2app(W_BufferedIOBase.read1_w),
     write = interp2app(W_BufferedIOBase.write_w),
     detach = interp2app(W_BufferedIOBase.detach_w),
+    readinto = interp2app(W_BufferedIOBase.readinto_w),
     )
 
 class BufferedMixin:

Modified: pypy/branch/fast-forward/pypy/module/_io/test/test_bufferedio.py
==============================================================================
--- pypy/branch/fast-forward/pypy/module/_io/test/test_bufferedio.py	(original)
+++ pypy/branch/fast-forward/pypy/module/_io/test/test_bufferedio.py	Thu Oct 28 16:04:19 2010
@@ -55,6 +55,15 @@
         assert raw.nbreads == 3
         f.close()
 
+    def test_readinto(self):
+        import _io
+        a = bytearray('x' * 10)
+        raw = _io.FileIO(self.tmpfile)
+        f = _io.BufferedReader(raw)
+        assert f.readinto(a) == 5
+        f.close()
+        assert a == 'a\nb\ncxxxxx'
+
     def test_seek(self):
         import _io
         raw = _io.FileIO(self.tmpfile)