From 3d2a7c40cca0a80c558c506bb7487613e9696131 Mon Sep 17 00:00:00 2001 From: JustAnotherArchivist Date: Wed, 25 Mar 2020 10:43:35 +0000 Subject: [PATCH] Add tests for parsing chunked data split before or in the final CRLF PR #4651 by @JustAnotherArchivist This change adds tests that demonstrate the failures described in #4630. They are marked as xfail so that they don't affect the CI status. Once the issue is fixed, they'll reported as XPASS and pytest will fail, which would be a signal that it's time to remove the xfail markers keeping the contents of the tests to prevent regressions. (ref: https://pganssle-talks.github.io/xfail-lightning) Co-Authored-By: Sviatoslav Sydorenko --- tests/test_http_parser.py | 49 +++++++++++++++++++++++++++++++++++++++ 1 file changed, 49 insertions(+) diff --git a/tests/test_http_parser.py b/tests/test_http_parser.py index 141eaba13ab..af2e8094706 100644 --- a/tests/test_http_parser.py +++ b/tests/test_http_parser.py @@ -826,6 +826,55 @@ async def test_parse_chunked_payload_size_error(self, stream) -> None: assert isinstance(out.exception(), http_exceptions.TransferEncodingError) + @pytest.mark.xfail( + reason="see https://github.com/aio-libs/aiohttp/issues/4630" + ) + async def test_parse_chunked_payload_split_end(self, protocol) -> None: + out = aiohttp.StreamReader(protocol, loop=None) + p = HttpPayloadParser(out, chunked=True) + p.feed_data(b'4\r\nasdf\r\n0\r\n') + p.feed_data(b'\r\n') + + assert out.is_eof() + assert b'asdf' == b''.join(out._buffer) + + @pytest.mark.xfail( + reason="see https://github.com/aio-libs/aiohttp/issues/4630" + ) + async def test_parse_chunked_payload_split_end2(self, protocol) -> None: + out = aiohttp.StreamReader(protocol, loop=None) + p = HttpPayloadParser(out, chunked=True) + p.feed_data(b'4\r\nasdf\r\n0\r\n\r') + p.feed_data(b'\n') + + assert out.is_eof() + assert b'asdf' == b''.join(out._buffer) + + async def test_parse_chunked_payload_split_end_trailers(self, + protocol) -> None: + out = aiohttp.StreamReader(protocol, loop=None) + p = HttpPayloadParser(out, chunked=True) + p.feed_data(b'4\r\nasdf\r\n0\r\n') + p.feed_data(b'Content-MD5: 912ec803b2ce49e4a541068d495ab570\r\n') + p.feed_data(b'\r\n') + + assert out.is_eof() + assert b'asdf' == b''.join(out._buffer) + + @pytest.mark.xfail( + reason="see https://github.com/aio-libs/aiohttp/issues/4630" + ) + async def test_parse_chunked_payload_split_end_trailers2(self, + protocol) -> None: + out = aiohttp.StreamReader(protocol, loop=None) + p = HttpPayloadParser(out, chunked=True) + p.feed_data(b'4\r\nasdf\r\n0\r\n') + p.feed_data(b'Content-MD5: 912ec803b2ce49e4a541068d495ab570\r\n\r') + p.feed_data(b'\n') + + assert out.is_eof() + assert b'asdf' == b''.join(out._buffer) + async def test_http_payload_parser_length(self, stream) -> None: out = aiohttp.FlowControlDataQueue(stream, loop=asyncio.get_event_loop())