quay/test/test_filelike.py

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)


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()