From a04b5f09757f51ed1a6f308ad43929e8718b560b Mon Sep 17 00:00:00 2001 From: Eugene Obukhov Date: Fri, 15 Apr 2016 00:49:00 +0300 Subject: [PATCH] Add tests for reader classes (#1241) * Fix FakeSocket.recv method * Add tests for Unreader and its subclasses * Add tests for EOFReader and LengthReader --- THANKS | 1 + tests/t.py | 2 +- tests/test_http.py | 120 ++++++++++++++++++++++++++++++++++++++++++++- 3 files changed, 121 insertions(+), 2 deletions(-) diff --git a/THANKS b/THANKS index 9a69da18..e147d9b1 100644 --- a/THANKS +++ b/THANKS @@ -160,3 +160,4 @@ Hebert J Kevin Littlejohn Wolfgang Schnerring Jason Madden +Eugene Obukhov \ No newline at end of file diff --git a/tests/t.py b/tests/t.py index f9476c91..d9e627cf 100644 --- a/tests/t.py +++ b/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) diff --git a/tests/test_http.py b/tests/test_http.py index 1a559acf..ddea58ca 100644 --- a/tests/test_http.py +++ b/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)