diff --git a/test/test_chunked.py b/test/test_chunked.py new file mode 100644 index 000000000..caa26f7eb --- /dev/null +++ b/test/test_chunked.py @@ -0,0 +1,188 @@ +import re + +import pytest +from unit.applications.lang.python import ApplicationPython + +prerequisites = {'modules': {'python': 'any'}} + +client = ApplicationPython() + + +@pytest.fixture(autouse=True) +def setup_method_fixture(): + client.load('mirror') + + assert 'success' in client.conf( + {"http": {"chunked_transform": True}}, 'settings' + ) + + +def test_chunked(): + def chunks(chunks=[]): + body = '' + + for c in chunks: + body = f'{body}{len(c):x}\r\n{c}\r\n' + + resp = client.get( + headers={ + 'Host': 'localhost', + 'Connection': 'close', + 'Transfer-Encoding': 'chunked', + }, + body=f'{body}0\r\n\r\n', + ) + + expect_body = ''.join(chunks) + + assert resp['status'] == 200 + assert resp['headers']['Content-Length'] == str(len(expect_body)) + assert resp['body'] == expect_body + + chunks() + chunks(['1']) + chunks(['0123456789']) + chunks(['0123456789' * 128]) + chunks(['0123456789' * 512]) + chunks(['0123456789' * 128, '1', '1', '0123456789' * 128, '1']) + + +def test_chunked_pipeline(): + sock = client.get( + no_recv=True, + headers={ + 'Host': 'localhost', + 'Transfer-Encoding': 'chunked', + }, + body='1\r\n$\r\n0\r\n\r\n', + ) + + resp = client.get( + sock=sock, + headers={ + 'Host': 'localhost', + 'Transfer-Encoding': 'chunked', + 'Connection': 'close', + }, + body='1\r\n%\r\n0\r\n\r\n', + raw_resp=True, + ) + + assert len(re.findall('200 OK', resp)) == 2 + assert len(re.findall('Content-Length: 1', resp)) == 2 + assert len(re.findall('$', resp)) == 1 + assert len(re.findall('%', resp)) == 1 + + +def test_chunked_max_body_size(): + assert 'success' in client.conf( + {'max_body_size': 1024, 'chunked_transform': True}, 'settings/http' + ) + + body = f'{2048:x}\r\n{"x" * 2048}\r\n0\r\n\r\n' + + assert ( + client.get( + headers={ + 'Host': 'localhost', + 'Connection': 'close', + 'Transfer-Encoding': 'chunked', + }, + body=body, + )['status'] + == 413 + ) + + +def test_chunked_after_last(): + resp = client.get( + headers={ + 'Host': 'localhost', + 'Connection': 'close', + 'Transfer-Encoding': 'chunked', + }, + body='1\r\na\r\n0\r\n\r\n1\r\nb\r\n0\r\n\r\n', + ) + + assert resp['status'] == 200 + assert resp['headers']['Content-Length'] == '1' + assert resp['body'] == 'a' + + +def test_chunked_transform(): + assert 'success' in client.conf( + {"http": {"chunked_transform": False}}, 'settings' + ) + + assert ( + client.get( + headers={ + 'Host': 'localhost', + 'Connection': 'close', + 'Transfer-Encoding': 'chunked', + }, + body='0\r\n\r\n', + )['status'] + == 411 + ) + + +def test_chunked_invalid(): + # invalid chunkes + + def check_body(body): + assert ( + client.get( + headers={ + 'Host': 'localhost', + 'Connection': 'close', + 'Transfer-Encoding': 'chunked', + }, + body=body, + )['status'] + == 400 + ) + + check_body('1\r\nblah\r\n0\r\n\r\n') + check_body('1\r\n\r\n1\r\n0\r\n\r\n') + check_body('1\r\n1\r\n\r\n0\r\n\r\n') + + # invalid transfer encoding header + + assert ( + client.get( + headers={ + 'Host': 'localhost', + 'Connection': 'close', + 'Transfer-Encoding': ['chunked', 'chunked'], + }, + body='0\r\n\r\n', + )['status'] + == 400 + ), 'two Transfer-Encoding headers' + + assert ( + client.get( + headers={ + 'Host': 'localhost', + 'Connection': 'close', + 'Transfer-Encoding': 'chunked', + 'Content-Length': '5', + }, + body='0\r\n\r\n', + )['status'] + == 400 + ), 'Transfer-Encoding and Content-Length' + + assert ( + client.get( + http_10=True, + headers={ + 'Host': 'localhost', + 'Connection': 'close', + 'Transfer-Encoding': 'chunked', + }, + body='0\r\n\r\n', + )['status'] + == 400 + ), 'Transfer-Encoding HTTP/1.0' diff --git a/test/unit/http.py b/test/unit/http.py index 9401501b6..e449c8b54 100644 --- a/test/unit/http.py +++ b/test/unit/http.py @@ -67,7 +67,7 @@ def http(self, start_str, **kwargs): headers['Content-Type'] = content_type - if 'Content-Length' not in headers: + if 'Content-Length' not in headers and 'Transfer-Encoding' not in headers: headers['Content-Length'] = len(body) for header, value in headers.items():