diff --git a/test/test_filelike.py b/test/test_filelike.py deleted file mode 100644 index b3870a54f..000000000 --- a/test/test_filelike.py +++ /dev/null @@ -1,143 +0,0 @@ -import unittest - -from StringIO import StringIO -from util.registry.filelike import FilelikeStreamConcat, LimitingStream, StreamSlice - -class TestFilelikeStreamConcat(unittest.TestCase): - def somegenerator(self): - yield 'some' - yield 'cool' - yield 'file-contents' - - def test_parts(self): - gens = iter([StringIO(s) for s in self.somegenerator()]) - fileobj = FilelikeStreamConcat(gens) - - self.assertEquals('so', fileobj.read(2)) - self.assertEquals('mec', fileobj.read(3)) - self.assertEquals('oolfile', fileobj.read(7)) - self.assertEquals('-contents', fileobj.read(-1)) - - def test_entire(self): - gens = iter([StringIO(s) for s in self.somegenerator()]) - fileobj = FilelikeStreamConcat(gens) - self.assertEquals('somecoolfile-contents', fileobj.read(-1)) - - -class TestLimitingStream(unittest.TestCase): - def test_nolimit(self): - fileobj = StringIO('this is a cool test') - stream = LimitingStream(fileobj) - self.assertEquals('this is a cool test', stream.read(-1)) - self.assertEquals(stream.tell(), len('this is a cool test')) - - def test_simplelimit(self): - fileobj = StringIO('this is a cool test') - stream = LimitingStream(fileobj, 4) - self.assertEquals('this', stream.read(-1)) - self.assertEquals(stream.tell(), 4) - - def test_simplelimit_readdefined(self): - fileobj = StringIO('this is a cool test') - stream = LimitingStream(fileobj, 4) - self.assertEquals('th', stream.read(2)) - self.assertEquals(stream.tell(), 2) - - def test_nolimit_readdefined(self): - fileobj = StringIO('this is a cool test') - stream = LimitingStream(fileobj, -1) - self.assertEquals('th', stream.read(2)) - self.assertEquals(stream.tell(), 2) - - def test_limit_multiread(self): - fileobj = StringIO('this is a cool test') - stream = LimitingStream(fileobj, 7) - self.assertEquals('this', stream.read(4)) - self.assertEquals(' is', stream.read(3)) - self.assertEquals('', stream.read(2)) - self.assertEquals(stream.tell(), 7) - - def test_limit_multiread2(self): - fileobj = StringIO('this is a cool test') - stream = LimitingStream(fileobj, 7) - self.assertEquals('this', stream.read(4)) - self.assertEquals(' is', stream.read(-1)) - self.assertEquals(stream.tell(), 7) - - def test_seek(self): - fileobj = StringIO('this is a cool test') - stream = LimitingStream(fileobj) - stream.seek(2) - - self.assertEquals('is', stream.read(2)) - self.assertEquals(stream.tell(), 4) - - def test_seek_withlimit(self): - fileobj = StringIO('this is a cool test') - stream = LimitingStream(fileobj, 3) - stream.seek(2) - - self.assertEquals('i', stream.read(2)) - self.assertEquals(stream.tell(), 3) - - def test_seek_pastlimit(self): - fileobj = StringIO('this is a cool test') - stream = LimitingStream(fileobj, 3) - stream.seek(4) - - self.assertEquals('', stream.read(1)) - self.assertEquals(stream.tell(), 3) - - def test_seek_to_tell(self): - fileobj = StringIO('this is a cool test') - stream = LimitingStream(fileobj, 3) - stream.seek(stream.tell()) - - self.assertEquals('thi', stream.read(4)) - self.assertEquals(stream.tell(), 3) - - -class TestStreamSlice(unittest.TestCase): - def test_none_read(self): - class NoneReader(object): - def read(self, size=None): - return None - - stream = StreamSlice(NoneReader(), 0) - self.assertEquals(None, stream.read(-1)) - self.assertEquals(0, stream.tell()) - - def test_noslice(self): - fileobj = StringIO('this is a cool test') - stream = StreamSlice(fileobj, 0) - self.assertEquals('this is a cool test', stream.read(-1)) - self.assertEquals(stream.tell(), len('this is a cool test')) - - def test_startindex(self): - fileobj = StringIO('this is a cool test') - stream = StreamSlice(fileobj, 5) - self.assertEquals('is a cool test', stream.read(-1)) - self.assertEquals(stream.tell(), len('is a cool test')) - - def test_startindex_limitedread(self): - fileobj = StringIO('this is a cool test') - stream = StreamSlice(fileobj, 5) - self.assertEquals('is a', stream.read(4)) - self.assertEquals(stream.tell(), 4) - - def test_slice(self): - fileobj = StringIO('this is a cool test') - stream = StreamSlice(fileobj, 5, 9) - self.assertEquals('is a', stream.read(-1)) - self.assertEquals(stream.tell(), len('is a')) - - def test_slice_explictread(self): - fileobj = StringIO('this is a cool test') - stream = StreamSlice(fileobj, 5, 9) - self.assertEquals('is', stream.read(2)) - self.assertEquals(' a', stream.read(5)) - self.assertEquals(stream.tell(), len('is a')) - - -if __name__ == '__main__': - unittest.main() diff --git a/util/registry/test/test_filelike.py b/util/registry/test/test_filelike.py new file mode 100644 index 000000000..406aa0d3b --- /dev/null +++ b/util/registry/test/test_filelike.py @@ -0,0 +1,132 @@ +from StringIO import StringIO +from util.registry.filelike import FilelikeStreamConcat, LimitingStream, StreamSlice + +def somegenerator(): + yield 'some' + yield 'cool' + yield 'file-contents' + +def test_parts(): + gens = iter([StringIO(s) for s in somegenerator()]) + fileobj = FilelikeStreamConcat(gens) + + assert fileobj.read(2) == 'so' + assert fileobj.read(3) == 'mec' + assert fileobj.read(7) == 'oolfile' + assert fileobj.read(-1) == '-contents' + +def test_entire(): + gens = iter([StringIO(s) for s in somegenerator()]) + fileobj = FilelikeStreamConcat(gens) + assert fileobj.read(-1) == 'somecoolfile-contents' + +def test_nolimit(): + fileobj = StringIO('this is a cool test') + stream = LimitingStream(fileobj) + assert stream.read(-1) == 'this is a cool test' + assert len('this is a cool test') == stream.tell() + +def test_simplelimit(): + fileobj = StringIO('this is a cool test') + stream = LimitingStream(fileobj, 4) + assert stream.read(-1) == 'this' + assert 4 == stream.tell() + +def test_simplelimit_readdefined(): + fileobj = StringIO('this is a cool test') + stream = LimitingStream(fileobj, 4) + assert stream.read(2) == 'th' + assert 2 == stream.tell() + +def test_nolimit_readdefined(): + fileobj = StringIO('this is a cool test') + stream = LimitingStream(fileobj, -1) + assert stream.read(2) == 'th' + assert 2 == stream.tell() + +def test_limit_multiread(): + fileobj = StringIO('this is a cool test') + stream = LimitingStream(fileobj, 7) + assert stream.read(4) == 'this' + assert stream.read(3) == ' is' + assert stream.read(2) == '' + assert 7 == stream.tell() + +def test_limit_multiread2(): + fileobj = StringIO('this is a cool test') + stream = LimitingStream(fileobj, 7) + assert stream.read(4) == 'this' + assert stream.read(-1) == ' is' + assert 7 == stream.tell() + +def test_seek(): + fileobj = StringIO('this is a cool test') + stream = LimitingStream(fileobj) + stream.seek(2) + + assert stream.read(2) == 'is' + assert 4 == stream.tell() + +def test_seek_withlimit(): + fileobj = StringIO('this is a cool test') + stream = LimitingStream(fileobj, 3) + stream.seek(2) + + assert stream.read(2) == 'i' + assert 3 == stream.tell() + +def test_seek_pastlimit(): + fileobj = StringIO('this is a cool test') + stream = LimitingStream(fileobj, 3) + stream.seek(4) + + assert stream.read(1) == '' + assert 3 == stream.tell() + +def test_seek_to_tell(): + fileobj = StringIO('this is a cool test') + stream = LimitingStream(fileobj, 3) + stream.seek(stream.tell()) + + assert stream.read(4) == 'thi' + assert 3 == stream.tell() + +def test_none_read(): + class NoneReader(object): + def read(self, size=None): + return None + + stream = StreamSlice(NoneReader(), 0) + assert stream.read(-1) == None + assert stream.tell() == 0 + +def test_noslice(): + fileobj = StringIO('this is a cool test') + stream = StreamSlice(fileobj, 0) + assert stream.read(-1) == 'this is a cool test' + assert len('this is a cool test') == stream.tell() + +def test_startindex(): + fileobj = StringIO('this is a cool test') + stream = StreamSlice(fileobj, 5) + assert stream.read(-1) == 'is a cool test' + assert len('is a cool test') == stream.tell() + +def test_startindex_limitedread(): + fileobj = StringIO('this is a cool test') + stream = StreamSlice(fileobj, 5) + assert stream.read(4) == 'is a' + assert 4 == stream.tell() + +def test_slice(): + fileobj = StringIO('this is a cool test') + stream = StreamSlice(fileobj, 5, 9) + assert stream.read(-1) == 'is a' + assert len('is a') == stream.tell() + +def test_slice_explictread(): + fileobj = StringIO('this is a cool test') + stream = StreamSlice(fileobj, 5, 9) + assert stream.read(2) == 'is' + assert stream.read(5) == ' a' + assert len('is a') == stream.tell()