Add tests for reader classes (#1241)

* Fix FakeSocket.recv method

* Add tests for Unreader and its subclasses

* Add tests for EOFReader and LengthReader

THANKS

@@ -160,3 +160,4 @@ Hebert J <hebert@mail.ru>
 Kevin Littlejohn <kevin@littlejohn.id.au>
 Wolfgang Schnerring <wosc@wosc.de>
 Jason Madden <jason@nextthought.com>
+Eugene Obukhov <irvind25@gmail.com>

tests/t.py

@@ -49,7 +49,7 @@ class FakeSocket(object):
         return self.tmp.len
 
     def recv(self, length=None):
-        return self.tmp.read()
+        return self.tmp.read(length)
 
     def recv_into(self, buf, length):
         tmp_buffer = self.tmp.read(length)

tests/test_http.py

@@ -2,9 +2,11 @@
 
 import t
 import pytest
+
 from gunicorn import util
-from gunicorn.http.body import Body
+from gunicorn.http.body import Body, LengthReader, EOFReader
 from gunicorn.http.wsgi import Response
+from gunicorn.http.unreader import Unreader, IterUnreader, SocketUnreader
 from gunicorn.six import BytesIO
 from gunicorn.http.errors import InvalidHeader, InvalidHeaderName
 
@@ -108,3 +110,119 @@ def test_http_invalid_response_header():
     response = Response(mocked_request, mocked_socket, None)
     with pytest.raises(InvalidHeaderName):
         response.start_response("200 OK", [('foo\r\n', 'essai')])
+
+
+def test_unreader_read_when_size_is_none():
+    unreader = Unreader()
+    unreader.chunk = mock.MagicMock(side_effect=[b'qwerty', b'123456', b''])
+
+    assert unreader.read(size=None) == b'qwerty'
+    assert unreader.read(size=None) == b'123456'
+    assert unreader.read(size=None) == b''
+
+
+def test_unreader_unread():
+    unreader = Unreader()
+    unreader.unread(b'hi there')
+    assert b'hi there' in unreader.read()
+
+
+def test_unreader_read_zero_size():
+    unreader = Unreader()
+    unreader.chunk = mock.MagicMock(side_effect=[b'qwerty', b'asdfgh'])
+
+    assert unreader.read(size=0) == b''
+
+
+def test_unreader_read_with_nonzero_size():
+    unreader = Unreader()
+    unreader.chunk = mock.MagicMock(side_effect=[
+        b'qwerty', b'asdfgh', b'zxcvbn', b'123456', b'', b''
+    ])
+
+    assert unreader.read(size=5) == b'qwert'
+    assert unreader.read(size=5) == b'yasdf'
+    assert unreader.read(size=5) == b'ghzxc'
+    assert unreader.read(size=5) == b'vbn12'
+    assert unreader.read(size=5) == b'3456'
+    assert unreader.read(size=5) == b''
+
+
+def test_unreader_raises_excpetion_on_invalid_size():
+    unreader = Unreader()
+    with pytest.raises(TypeError):
+        unreader.read(size='foobar')
+    with pytest.raises(TypeError):
+        unreader.read(size=3.14)
+    with pytest.raises(TypeError):
+        unreader.read(size=[])
+
+
+def test_iter_unreader_chunk():
+    iter_unreader = IterUnreader((b'ab', b'cd', b'ef'))
+
+    assert iter_unreader.chunk() == b'ab'
+    assert iter_unreader.chunk() == b'cd'
+    assert iter_unreader.chunk() == b'ef'
+    assert iter_unreader.chunk() == b''
+    assert iter_unreader.chunk() == b''
+
+
+def test_socket_unreader_chunk():
+    fake_sock = t.FakeSocket(BytesIO(b'Lorem ipsum dolor'))
+    sock_unreader = SocketUnreader(fake_sock, max_chunk=5)
+
+    assert sock_unreader.chunk() == b'Lorem'
+    assert sock_unreader.chunk() == b' ipsu'
+    assert sock_unreader.chunk() == b'm dol'
+    assert sock_unreader.chunk() == b'or'
+    assert sock_unreader.chunk() == b''
+
+
+def test_length_reader_read():
+    unreader = IterUnreader((b'Lorem', b'ipsum', b'dolor', b'sit', b'amet'))
+    reader = LengthReader(unreader, 13)
+    assert reader.read(0) == b''
+    assert reader.read(5) == b'Lorem'
+    assert reader.read(6) == b'ipsumd'
+    assert reader.read(4) == b'ol'
+    assert reader.read(100) == b''
+
+    reader = LengthReader(unreader, 10)
+    assert reader.read(0) == b''
+    assert reader.read(5) == b'orsit'
+    assert reader.read(5) == b'amet'
+    assert reader.read(100) == b''
+
+
+def test_length_reader_read_invalid_size():
+    reader = LengthReader(None, 5)
+    with pytest.raises(TypeError):
+        reader.read('100')
+    with pytest.raises(TypeError):
+        reader.read([100])
+    with pytest.raises(ValueError):
+        reader.read(-100)
+
+
+def test_eof_reader_read():
+    unreader = IterUnreader((b'Lorem', b'ipsum', b'dolor', b'sit', b'amet'))
+    reader = EOFReader(unreader)
+
+    assert reader.read(0) == b''
+    assert reader.read(5) == b'Lorem'
+    assert reader.read(5) == b'ipsum'
+    assert reader.read(3) == b'dol'
+    assert reader.read(3) == b'ors'
+    assert reader.read(100) == b'itamet'
+    assert reader.read(100) == b''
+
+
+def test_eof_reader_read_invalid_size():
+    reader = EOFReader(None)
+    with pytest.raises(TypeError):
+        reader.read('100')
+    with pytest.raises(TypeError):
+        reader.read([100])
+    with pytest.raises(ValueError):
+        reader.read(-100)