From 34c4d0c4a29c8b3c26925c2a05a9b0e50a83f617 Mon Sep 17 00:00:00 2001 From: achapp Date: Mon, 24 Nov 2014 18:08:26 -0600 Subject: Test update/Fix progress temp save Edited test to catch readline error. file.py code change in progress (DOES NOT WORK PROPERLY) so saving it temporarily. --- tests/test_file.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) (limited to 'tests/test_file.py') diff --git a/tests/test_file.py b/tests/test_file.py index 22a34aca..24a7fa9b 100755 --- a/tests/test_file.py +++ b/tests/test_file.py @@ -70,13 +70,15 @@ class BufferedFileTest (unittest.TestCase): def test_2_readline(self): f = LoopbackFile('r+U') - f.write(b'First line.\nSecond line.\r\nThird line.\nFinal line non-terminated.') + f.write(b'First line.\nSecond line.\r\nThird line.\nFourth line.\nFifth line.\nFinal line non-terminated.') self.assertEqual(f.readline(), 'First line.\n') # universal newline mode should convert this linefeed: self.assertEqual(f.readline(), 'Second line.\n') # truncated line: self.assertEqual(f.readline(7), 'Third l') self.assertEqual(f.readline(), 'ine.\n') + self.assertEqual(f.readline(25), 'Fourth line.\n') + self.assertEqual(f.readline(), 'Fifth line.\n') self.assertEqual(f.readline(), 'Final line non-terminated.') self.assertEqual(f.readline(), '') f.close() -- cgit v1.2.3 From 0a5485390d43f408b130011ae3452855e766786d Mon Sep 17 00:00:00 2001 From: achapp Date: Tue, 25 Nov 2014 12:30:32 -0600 Subject: new readline test passes Changed file.py readline() to always check for a newline. Had to make a few changes for what went into self._rbuffer in the case where buffer size was met or exceeded and we found a newline. --- paramiko/file.py | 12 ++++++------ tests/test_file.py | 1 + 2 files changed, 7 insertions(+), 6 deletions(-) (limited to 'tests/test_file.py') diff --git a/paramiko/file.py b/paramiko/file.py index 139c453b..f549cb99 100644 --- a/paramiko/file.py +++ b/paramiko/file.py @@ -204,6 +204,7 @@ class BufferedFile (object): if not (self._flags & self.FLAG_READ): raise IOError('File not open for reading') line = self._rbuffer + truncated = False; while True: if self._at_trailing_cr and (self._flags & self.FLAG_UNIVERSAL_NEWLINE) and (len(line) > 0): # edge case: the newline may be '\r\n' and we may have read @@ -218,11 +219,11 @@ class BufferedFile (object): # enough. if (size is not None) and (size >= 0): if len(line) >= size: - # truncate line and return + # truncate line self._rbuffer = line[size:] line = line[:size] - #self._pos += len(line) - break#return line if self._flags & self.FLAG_BINARY else u(line) + truncated = True + break n = size - len(line) else: n = self._bufsize @@ -245,14 +246,13 @@ class BufferedFile (object): if (rpos >= 0) and (rpos < pos or pos < 0): pos = rpos if pos == -1: - #self._rbuffer = line[size:] - #line = line[:size] self._pos += len(line) return line if self._flags & self.FLAG_BINARY else u(line) xpos = pos + 1 if (line[pos] == cr_byte_value) and (xpos < len(line)) and (line[xpos] == linefeed_byte_value): xpos += 1 - self._rbuffer = line[xpos:] + + self._rbuffer = line[xpos:] + self._rbuffer if truncated else line[xpos:] lf = line[pos:xpos] line = line[:pos] + linefeed_byte if (len(self._rbuffer) == 0) and (lf == cr_byte): diff --git a/tests/test_file.py b/tests/test_file.py index 24a7fa9b..044dc591 100755 --- a/tests/test_file.py +++ b/tests/test_file.py @@ -77,6 +77,7 @@ class BufferedFileTest (unittest.TestCase): # truncated line: self.assertEqual(f.readline(7), 'Third l') self.assertEqual(f.readline(), 'ine.\n') + # readline should not read past the fourth line self.assertEqual(f.readline(25), 'Fourth line.\n') self.assertEqual(f.readline(), 'Fifth line.\n') self.assertEqual(f.readline(), 'Final line non-terminated.') -- cgit v1.2.3 From b2f45f90350b46e69ebb677cb040b916c95fbc34 Mon Sep 17 00:00:00 2001 From: achapp Date: Tue, 25 Nov 2014 13:23:15 -0600 Subject: Refactoring Added comments. Removed fifth line from test because it was unnecessary since the final line could be used instead. --- paramiko/file.py | 7 +++++-- tests/test_file.py | 9 +++++---- 2 files changed, 10 insertions(+), 6 deletions(-) (limited to 'tests/test_file.py') diff --git a/paramiko/file.py b/paramiko/file.py index f549cb99..1c99abeb 100644 --- a/paramiko/file.py +++ b/paramiko/file.py @@ -204,7 +204,7 @@ class BufferedFile (object): if not (self._flags & self.FLAG_READ): raise IOError('File not open for reading') line = self._rbuffer - truncated = False; + truncated = False while True: if self._at_trailing_cr and (self._flags & self.FLAG_UNIVERSAL_NEWLINE) and (len(line) > 0): # edge case: the newline may be '\r\n' and we may have read @@ -246,12 +246,15 @@ class BufferedFile (object): if (rpos >= 0) and (rpos < pos or pos < 0): pos = rpos if pos == -1: + # we couldn't find a newline in the truncated string, return it self._pos += len(line) return line if self._flags & self.FLAG_BINARY else u(line) xpos = pos + 1 if (line[pos] == cr_byte_value) and (xpos < len(line)) and (line[xpos] == linefeed_byte_value): xpos += 1 - + # if the string was truncated, _rbuffer needs to have the string after + # the newline character plus the truncated part of the line we stored + # earlier in _rbuffer self._rbuffer = line[xpos:] + self._rbuffer if truncated else line[xpos:] lf = line[pos:xpos] line = line[:pos] + linefeed_byte diff --git a/tests/test_file.py b/tests/test_file.py index 044dc591..a6ff69e9 100755 --- a/tests/test_file.py +++ b/tests/test_file.py @@ -70,16 +70,17 @@ class BufferedFileTest (unittest.TestCase): def test_2_readline(self): f = LoopbackFile('r+U') - f.write(b'First line.\nSecond line.\r\nThird line.\nFourth line.\nFifth line.\nFinal line non-terminated.') + f.write(b'First line.\nSecond line.\r\nThird line.\n' + + b'Fourth line.\nFinal line non-terminated.') + self.assertEqual(f.readline(), 'First line.\n') # universal newline mode should convert this linefeed: self.assertEqual(f.readline(), 'Second line.\n') # truncated line: self.assertEqual(f.readline(7), 'Third l') self.assertEqual(f.readline(), 'ine.\n') - # readline should not read past the fourth line - self.assertEqual(f.readline(25), 'Fourth line.\n') - self.assertEqual(f.readline(), 'Fifth line.\n') + # newline should be detected and only the fourth line returned + self.assertEqual(f.readline(39), 'Fourth line.\n') self.assertEqual(f.readline(), 'Final line non-terminated.') self.assertEqual(f.readline(), '') f.close() -- cgit v1.2.3