Author: Armin Rigo <ar...@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
_______________________________________________
pypy-commit mailing list
pypy-commit@python.org
http://mail.python.org/mailman/listinfo/pypy-commit
