Merge pull request #3145 from quay/move-filelike-test
Move filelike tests to pytest
This commit is contained in:
Joseph Schorr
GitHub
commit
3bb889eacf
2 changed files with 132 additions and 143 deletions
|
|
@ -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()
|
||||
132
util/registry/test/test_filelike.py
Normal file
132
util/registry/test/test_filelike.py
Normal file
|
|
@ -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()
|
||||
Reference in a new issue