diff options
author | Andrei Zeliankou <zelenkov@nginx.com> | 2020-09-16 21:31:15 +0100 |
---|---|---|
committer | Andrei Zeliankou <zelenkov@nginx.com> | 2020-09-16 21:31:15 +0100 |
commit | d5e915934066c77a59d211efafca10c117b73d05 (patch) | |
tree | f894a3c09bd8aa43e87276eed377eb09f97e46fe /test/test_proxy_chunked.py | |
parent | 77ecb6ab49257dd662aa9c461fed3dc1d74e5092 (diff) | |
download | unit-d5e915934066c77a59d211efafca10c117b73d05.tar.gz unit-d5e915934066c77a59d211efafca10c117b73d05.tar.bz2 |
Tests: migrated to the pytest.
Diffstat (limited to 'test/test_proxy_chunked.py')
-rw-r--r-- | test/test_proxy_chunked.py | 135 |
1 files changed, 62 insertions, 73 deletions
diff --git a/test/test_proxy_chunked.py b/test/test_proxy_chunked.py index f344b69a..93746703 100644 --- a/test/test_proxy_chunked.py +++ b/test/test_proxy_chunked.py @@ -12,7 +12,7 @@ class TestProxyChunked(TestApplicationPython): SERVER_PORT = 7999 @staticmethod - def run_server(server_port, testdir): + def run_server(server_port, temp_dir): sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1) @@ -81,68 +81,62 @@ class TestProxyChunked(TestApplicationPython): def get_http10(self, *args, **kwargs): return self.get(*args, http_10=True, **kwargs) - def setUp(self): - super().setUp() + def setup_method(self): + super().setup_method() - self.run_process(self.run_server, self.SERVER_PORT, self.testdir) + self.run_process(self.run_server, self.SERVER_PORT, self.temp_dir) self.waitforsocket(self.SERVER_PORT) - self.assertIn( - 'success', - self.conf( - { - "listeners": {"*:7080": {"pass": "routes"},}, - "routes": [ - { - "action": { - "proxy": "http://127.0.0.1:" - + str(self.SERVER_PORT) - } + assert 'success' in self.conf( + { + "listeners": {"*:7080": {"pass": "routes"},}, + "routes": [ + { + "action": { + "proxy": "http://127.0.0.1:" + + str(self.SERVER_PORT) } - ], - } - ), - 'proxy initial configuration', - ) + } + ], + } + ), 'proxy initial configuration' def test_proxy_chunked(self): for _ in range(10): - self.assertEqual( - self.get_http10(body='\r\n\r\n0\r\n\r\n')['status'], 200 - ) + assert self.get_http10(body='\r\n\r\n0\r\n\r\n')['status'] == 200 def test_proxy_chunked_body(self): part = '0123456789abcdef' - self.assertEqual( + assert ( self.get_http10(body=self.chunks([('1000', part + ' X 256')]))[ 'body' - ], - part * 256, + ] + == part * 256 ) - self.assertEqual( + assert ( self.get_http10(body=self.chunks([('100000', part + ' X 65536')]))[ 'body' - ], - part * 65536, + ] + == part * 65536 ) - self.assertEqual( + assert ( self.get_http10( body=self.chunks([('1000000', part + ' X 1048576')]), read_buffer_size=4096 * 4096, - )['body'], - part * 1048576, + )['body'] + == part * 1048576 ) - self.assertEqual( + assert ( self.get_http10( body=self.chunks( [('1000', part + ' X 256'), ('1000', part + ' X 256')] ) - )['body'], - part * 256 * 2, + )['body'] + == part * 256 * 2 ) - self.assertEqual( + assert ( self.get_http10( body=self.chunks( [ @@ -150,10 +144,10 @@ class TestProxyChunked(TestApplicationPython): ('100000', part + ' X 65536'), ] ) - )['body'], - part * 65536 * 2, + )['body'] + == part * 65536 * 2 ) - self.assertEqual( + assert ( self.get_http10( body=self.chunks( [ @@ -162,42 +156,40 @@ class TestProxyChunked(TestApplicationPython): ] ), read_buffer_size=4096 * 4096, - )['body'], - part * 1048576 * 2, + )['body'] + == part * 1048576 * 2 ) def test_proxy_chunked_fragmented(self): part = '0123456789abcdef' - self.assertEqual( + assert ( self.get_http10( body=self.chunks( [('1', hex(i % 16)[2:]) for i in range(4096)] ), - )['body'], - part * 256, + )['body'] + == part * 256 ) def test_proxy_chunked_send(self): - self.assertEqual( - self.get_http10(body='\r\n\r\n@0@\r\n\r\n')['status'], 200 - ) - self.assertEqual( + assert self.get_http10(body='\r\n\r\n@0@\r\n\r\n')['status'] == 200 + assert ( self.get_http10( body='\r@\n\r\n2\r@\na@b\r\n2\r\ncd@\r\n0\r@\n\r\n' - )['body'], - 'abcd', + )['body'] + == 'abcd' ) - self.assertEqual( + assert ( self.get_http10( body='\r\n\r\n2\r#\na#b\r\n##2\r\n#cd\r\n0\r\n#\r#\n' - )['body'], - 'abcd', + )['body'] + == 'abcd' ) def test_proxy_chunked_invalid(self): def check_invalid(body): - self.assertNotEqual(self.get_http10(body=body)['status'], 200) + assert self.get_http10(body=body)['status'] != 200 check_invalid('\r\n\r0') check_invalid('\r\n\r\n\r0') @@ -209,41 +201,38 @@ class TestProxyChunked(TestApplicationPython): check_invalid('\r\n\r\n0\r\nX') resp = self.get_http10(body='\r\n\r\n65#\r\nA X 100') - self.assertEqual(resp['status'], 200, 'incomplete chunk status') - self.assertNotEqual(resp['body'][-5:], '0\r\n\r\n', 'incomplete chunk') + assert resp['status'] == 200, 'incomplete chunk status' + assert resp['body'][-5:] != '0\r\n\r\n', 'incomplete chunk' resp = self.get_http10(body='\r\n\r\n64#\r\nA X 100') - self.assertEqual(resp['status'], 200, 'no zero chunk status') - self.assertNotEqual(resp['body'][-5:], '0\r\n\r\n', 'no zero chunk') + assert resp['status'] == 200, 'no zero chunk status' + assert resp['body'][-5:] != '0\r\n\r\n', 'no zero chunk' - self.assertEqual( - self.get_http10(body='\r\n\r\n80000000\r\nA X 100')['status'], 200, + assert ( + self.get_http10(body='\r\n\r\n80000000\r\nA X 100')['status'] + == 200 ) - self.assertEqual( + assert ( self.get_http10(body='\r\n\r\n10000000000000000\r\nA X 100')[ 'status' - ], - 502, + ] + == 502 ) - self.assertGreaterEqual( + assert ( len( self.get_http10( body='\r\n\r\n1000000\r\nA X 1048576\r\n1000000\r\nA X 100', read_buffer_size=4096 * 4096, )['body'] - ), - 1048576, + ) + >= 1048576 ) - self.assertGreaterEqual( + assert ( len( self.get_http10( body='\r\n\r\n1000000\r\nA X 1048576\r\nXXX\r\nA X 100', read_buffer_size=4096 * 4096, )['body'] - ), - 1048576, + ) + >= 1048576 ) - - -if __name__ == '__main__': - TestProxyChunked.main() |