test_limitreader.py 1.4 KB

123456789101112131415161718192021222324252627282930313233343536373839
  1. from warcio.limitreader import LimitReader
  2. from contextlib import closing
  3. from io import BytesIO
  4. class TestLimitReader(object):
  5. def test_limit_reader_1(self):
  6. assert b'abcdefghji' == LimitReader(BytesIO(b'abcdefghjiklmnopqrstuvwxyz'), 10).read(26)
  7. def test_limit_reader_2(self):
  8. assert b'abcdefgh' == LimitReader(BytesIO(b'abcdefghjiklmnopqrstuvwxyz'), 8).readline(26)
  9. def test_limit_reader_3(self):
  10. reader = LimitReader(BytesIO(b'abcdefghjiklmnopqrstuvwxyz'), 8)
  11. new_reader = LimitReader.wrap_stream(reader, 4)
  12. assert reader == new_reader
  13. assert b'abcd' == new_reader.readline(26)
  14. #assert b'abcd' == LimitReader.wrap_stream(LimitReader(BytesIO(b'abcdefghjiklmnopqrstuvwxyz'), 8), 4).readline(26)
  15. def test_limit_reader_multiple_read(self):
  16. reader = LimitReader(BytesIO(b'abcdefghjiklmnopqrstuvwxyz'), 10)
  17. string = None
  18. for x in [2, 2, 20]:
  19. string = reader.read(x)
  20. assert b'efghji' == string
  21. def test_limit_reader_zero(self):
  22. assert b'' == LimitReader(BytesIO(b'a'), 0).readline(0)
  23. def test_limit_reader_invalid_wrap(self):
  24. b = BytesIO(b'some data')
  25. assert LimitReader.wrap_stream(b, 'abc') == b
  26. def test_limit_reader_close(self):
  27. reader = LimitReader(BytesIO(b'abcdefg'), 3)
  28. with closing(reader):
  29. assert b'abc' == reader.read(10)
  30. assert reader.tell() == 3