From 39f64abf52669a32d2d58a7a056b89e6aa5feae7 Mon Sep 17 00:00:00 2001 From: bescoto Date: Fri, 21 Feb 2003 05:00:21 +0000 Subject: Iterfiles and iterrorps now can contain exceptions. git-svn-id: http://svn.savannah.nongnu.org/svn/rdiff-backup/trunk@283 2b77aa54-bcbc-44c9-a7ec-4f6cf2b41109 --- rdiff-backup/testing/iterfiletest.py | 38 ++++++++++++++++++++++++++++++++++++ 1 file changed, 38 insertions(+) (limited to 'rdiff-backup/testing/iterfiletest.py') diff --git a/rdiff-backup/testing/iterfiletest.py b/rdiff-backup/testing/iterfiletest.py index 63975d0..c8c77c6 100644 --- a/rdiff-backup/testing/iterfiletest.py +++ b/rdiff-backup/testing/iterfiletest.py @@ -3,6 +3,17 @@ from commontest import * from rdiff_backup.iterfile import * from rdiff_backup import lazy +class FileException: + """Like a file, but raise exception after certain # bytes read""" + def __init__(self, max): + self.count = 0 + self.max = max + def read(self, l): + self.count += l + if self.count > self.max: raise IOError(13, "Permission Denied") + return "a"*l + def close(self): return None + class testIterFile(unittest.TestCase): def setUp(self): @@ -15,6 +26,33 @@ class testIterFile(unittest.TestCase): assert lazy.Iter.equal(itm(), IterWrappingFile(FileWrappingIter(itm()))) + def testFile(self): + """Test sending files through iters""" + buf1 = "hello"*10000 + file1 = StringIO.StringIO(buf1) + buf2 = "goodbye"*10000 + file2 = StringIO.StringIO(buf2) + file_iter = FileWrappingIter(iter([file1, file2])) + + new_iter = IterWrappingFile(file_iter) + assert new_iter.next().read() == buf1 + assert new_iter.next().read() == buf2 + self.assertRaises(StopIteration, new_iter.next) + + def testFileException(self): + """Test encoding a file which raises an exception""" + f = FileException(100*1024) + new_iter = IterWrappingFile(FileWrappingIter(iter([f, "foo"]))) + f_out = new_iter.next() + assert f_out.read(10000) == "a"*10000 + try: buf = f_out.read(100*1024) + except IOError: pass + else: assert 0, len(buf) + + assert new_iter.next() == "foo" + self.assertRaises(StopIteration, new_iter.next) + + class testBufferedRead(unittest.TestCase): def testBuffering(self): """Test buffering a StringIO""" -- cgit v1.2.1