[pypy-commit] pypy default: Various new tests. Fix for test_readinto_small_parts.

[arigo]

Author: Armin Rigo <arigo at tunes.org>
Branch: 
Changeset: r90554:c415b3b5820c
Date: 2017-03-05 09:50 +0100
http://bitbucket.org/pypy/pypy/changeset/c415b3b5820c/

Log:	Various new tests. Fix for test_readinto_small_parts.

diff --git a/pypy/module/_io/interp_bufferedio.py b/pypy/module/_io/interp_bufferedio.py
--- a/pypy/module/_io/interp_bufferedio.py
+++ b/pypy/module/_io/interp_bufferedio.py
@@ -16,6 +16,7 @@
     check_readable_w, check_writable_w, check_seekable_w)
 from pypy.module._io.interp_io import W_BlockingIOError
 from rpython.rlib import rthread
+from rpython.rtyper.lltypesystem import rffi
 
 STATE_ZERO, STATE_OK, STATE_DETACHED = range(3)
 
@@ -123,7 +124,8 @@
         self.buf[self.start + index] = char
 
     def get_raw_address(self):
-        return nonmoving_raw_ptr_for_resizable_list(self.buf)
+        ptr = nonmoving_raw_ptr_for_resizable_list(self.buf)
+        return rffi.ptradd(ptr, self.start)
 
 class BufferedMixin:
     _mixin_ = True
diff --git a/pypy/module/_io/test/test_bufferedio.py b/pypy/module/_io/test/test_bufferedio.py
--- a/pypy/module/_io/test/test_bufferedio.py
+++ b/pypy/module/_io/test/test_bufferedio.py
@@ -12,6 +12,9 @@
         tmpfile = udir.join('tmpfile')
         tmpfile.write("a\nb\nc", mode='wb')
         cls.w_tmpfile = cls.space.wrap(str(tmpfile))
+        bigtmpfile = udir.join('bigtmpfile')
+        bigtmpfile.write("a\nb\nc" * 20, mode='wb')
+        cls.w_bigtmpfile = cls.space.wrap(str(bigtmpfile))
 
     def test_simple_read(self):
         import _io
@@ -135,10 +138,14 @@
 
     def test_readinto(self):
         import _io
-        a = bytearray('x' * 10)
+        a1 = bytearray('x')
+        a = bytearray('x' * 9)
         raw = _io.FileIO(self.tmpfile)
         f = _io.BufferedReader(raw)
-        assert f.readinto(a) == 5
+        assert f.readinto(a1) == 1
+        assert a1 == 'a'
+        assert f.readinto(a) == 4
+        assert a == '\nb\ncxxxxx'
         exc = raises(TypeError, f.readinto, u"hello")
         assert str(exc.value) == "cannot use unicode as modifiable buffer"
         exc = raises(TypeError, f.readinto, buffer(b"hello"))
@@ -148,7 +155,17 @@
         exc = raises(TypeError, f.readinto, memoryview(b"hello"))
         assert str(exc.value) == "must be read-write buffer, not memoryview"
         f.close()
-        assert a == 'a\nb\ncxxxxx'
+
+    def test_readinto_big(self):
+        import _io
+        a1 = bytearray('x')
+        a = bytearray('x' * 199)
+        raw = _io.FileIO(self.bigtmpfile)
+        f = _io.BufferedReader(raw)
+        assert f.readinto(a1) == 1
+        assert a1 == 'a'
+        assert f.readinto(a) == 99
+        assert a == '\nb\nc' + 'a\nb\nc' * 19 + 'x' * 100
 
     def test_seek(self):
         import _io
@@ -237,7 +254,21 @@
         assert rawio.count == 4
 
 class AppTestBufferedReaderWithThreads(AppTestBufferedReader):
-    spaceconfig = dict(usemodules=['_io', 'thread'])
+    spaceconfig = dict(usemodules=['_io', 'thread', 'time'])
+
+    def test_readinto_small_parts(self):
+        import _io, os, thread, time
+        read_fd, write_fd = os.pipe()
+        raw = _io.FileIO(read_fd)
+        f = _io.BufferedReader(raw)
+        a = bytearray(b'x' * 10)
+        os.write(write_fd, b"abcde")
+        def write_more():
+            time.sleep(0.5)
+            os.write(write_fd, b"fghij")
+        thread.start_new_thread(write_more, ())
+        assert f.readinto(a) == 10
+        assert a == 'abcdefghij'
 
 
 class AppTestBufferedWriter: