[pypy-commit] pypy buffer-readline: Tests and fixes.

[Armin Rigo]

Author: Armin Rigo <arigo at tunes.org>
Branch: buffer-readline
Changeset: r44656:967778208b73
Date: 2011-06-02 22:16 +0200
http://bitbucket.org/pypy/pypy/changeset/967778208b73/

Log:	Tests and fixes.

diff --git a/pypy/rlib/streamio.py b/pypy/rlib/streamio.py
--- a/pypy/rlib/streamio.py
+++ b/pypy/rlib/streamio.py
@@ -810,10 +810,35 @@
             self.bufstart = 0
         return self.buf
 
-    tell       = PassThrough("tell",      flush_buffers=True)
+    def tell(self):
+        return self.base.tell() - (len(self.buf) - self.bufstart)
+
+    def readall(self):
+        result = self.base.readall()
+        if self.buf:
+            result = self.buf[self.bufstart:] + result
+            self.buf = ''
+            self.bufstart = 0
+        return result
+
+    def read(self, n):
+        if not self.buf:
+            return self.do_read(n)
+        else:
+            m = n - (len(self.buf) - self.bufstart)
+            start = self.bufstart
+            if m > 0:
+                result = self.buf[start:] + self.do_read(m)
+                self.buf = ''
+                self.bufstart = 0
+                return result
+            elif n >= 0:
+                self.bufstart = start + n
+                return self.buf[start : self.bufstart]
+            else:
+                return ''
+
     seek       = PassThrough("seek",      flush_buffers=True)
-    readall    = PassThrough("readall",   flush_buffers=True)
-    read       = PassThrough("read",      flush_buffers=True)
     write      = PassThrough("write",     flush_buffers=True)
     truncate   = PassThrough("truncate",  flush_buffers=True)
     flush      = PassThrough("flush",     flush_buffers=True)
diff --git a/pypy/rlib/test/test_streamio.py b/pypy/rlib/test/test_streamio.py
--- a/pypy/rlib/test/test_streamio.py
+++ b/pypy/rlib/test/test_streamio.py
@@ -1013,12 +1013,17 @@
     packets = ["a", "b", "\n", "def", "\nxy\npq\nuv", "wx"]
     lines = ["ab\n", "def\n", "xy\n", "pq\n", "uvwx"]
 
-    def makeStream(self, seek=False, bufsize=-1):
+    def makeStream(self, seek=False, tell=False, bufsize=-1):
         base = TSource(self.packets)
         self.source = base
         def f(*args):
-            raise NotImplementedError
-        base.tell = f
+            if seek is False:
+                raise NotImplementedError     # a bug!
+            if seek is None:
+                raise streamio.MyNotImplementedError   # can be caught
+            raise ValueError(seek)  # uh?
+        if not tell:
+            base.tell = f
         if not seek:
             base.seek = f
         return streamio.ReadlineInputStream(base, bufsize)
@@ -1048,6 +1053,30 @@
                 i += 1
             assert i == len(self.lines)
 
+    def test_readline_and_read_interleaved_no_seek(self):
+        for file in [self.makeStream(seek=None),
+                     self.makeStream(seek=None, bufsize=2)]:
+            i = 0
+            while 1:
+                firstchar = file.read(1)
+                if firstchar == "":
+                    break
+                r = file.readline()
+                assert r != ""
+                assert self.lines[i] == firstchar + r
+                i += 1
+            assert i == len(self.lines)
+
+    def test_readline_and_readall(self):
+        file = self.makeStream(seek=True, tell=True, bufsize=2)
+        r = file.readline()
+        assert r == 'ab\n'
+        assert file.tell() == 3
+        r = file.readall()
+        assert r == 'def\nxy\npq\nuvwx'
+        r = file.readall()
+        assert r == ''
+
 
 # Speed test