Deleted
Added
test_proxy_chunked.py (1527:89bc2fef480b) | test_proxy_chunked.py (1596:b7e2d4d92624) |
---|---|
1import re 2import select 3import socket 4import time 5 6from unit.applications.lang.python import TestApplicationPython 7 8 9class TestProxyChunked(TestApplicationPython): 10 prerequisites = {'modules': {'python': 'any'}} 11 12 SERVER_PORT = 7999 13 14 @staticmethod | 1import re 2import select 3import socket 4import time 5 6from unit.applications.lang.python import TestApplicationPython 7 8 9class TestProxyChunked(TestApplicationPython): 10 prerequisites = {'modules': {'python': 'any'}} 11 12 SERVER_PORT = 7999 13 14 @staticmethod |
15 def run_server(server_port, testdir): | 15 def run_server(server_port, temp_dir): |
16 sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) 17 sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) 18 sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1) 19 20 server_address = ('127.0.0.1', server_port) 21 sock.bind(server_address) 22 sock.listen(10) 23 --- 52 unchanged lines hidden (view full) --- 76 for l, c in chunks: 77 body = body + l + '\r\n' + c + '\r\n' 78 79 return body + '0\r\n\r\n' 80 81 def get_http10(self, *args, **kwargs): 82 return self.get(*args, http_10=True, **kwargs) 83 | 16 sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) 17 sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) 18 sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1) 19 20 server_address = ('127.0.0.1', server_port) 21 sock.bind(server_address) 22 sock.listen(10) 23 --- 52 unchanged lines hidden (view full) --- 76 for l, c in chunks: 77 body = body + l + '\r\n' + c + '\r\n' 78 79 return body + '0\r\n\r\n' 80 81 def get_http10(self, *args, **kwargs): 82 return self.get(*args, http_10=True, **kwargs) 83 |
84 def setUp(self): 85 super().setUp() | 84 def setup_method(self): 85 super().setup_method() |
86 | 86 |
87 self.run_process(self.run_server, self.SERVER_PORT, self.testdir) | 87 self.run_process(self.run_server, self.SERVER_PORT, self.temp_dir) |
88 self.waitforsocket(self.SERVER_PORT) 89 | 88 self.waitforsocket(self.SERVER_PORT) 89 |
90 self.assertIn( 91 'success', 92 self.conf( 93 { 94 "listeners": {"*:7080": {"pass": "routes"},}, 95 "routes": [ 96 { 97 "action": { 98 "proxy": "http://127.0.0.1:" 99 + str(self.SERVER_PORT) 100 } | 90 assert 'success' in self.conf( 91 { 92 "listeners": {"*:7080": {"pass": "routes"},}, 93 "routes": [ 94 { 95 "action": { 96 "proxy": "http://127.0.0.1:" 97 + str(self.SERVER_PORT) |
101 } | 98 } |
102 ], 103 } 104 ), 105 'proxy initial configuration', 106 ) | 99 } 100 ], 101 } 102 ), 'proxy initial configuration' |
107 108 def test_proxy_chunked(self): 109 for _ in range(10): | 103 104 def test_proxy_chunked(self): 105 for _ in range(10): |
110 self.assertEqual( 111 self.get_http10(body='\r\n\r\n0\r\n\r\n')['status'], 200 112 ) | 106 assert self.get_http10(body='\r\n\r\n0\r\n\r\n')['status'] == 200 |
113 114 def test_proxy_chunked_body(self): 115 part = '0123456789abcdef' 116 | 107 108 def test_proxy_chunked_body(self): 109 part = '0123456789abcdef' 110 |
117 self.assertEqual( | 111 assert ( |
118 self.get_http10(body=self.chunks([('1000', part + ' X 256')]))[ 119 'body' | 112 self.get_http10(body=self.chunks([('1000', part + ' X 256')]))[ 113 'body' |
120 ], 121 part * 256, | 114 ] 115 == part * 256 |
122 ) | 116 ) |
123 self.assertEqual( | 117 assert ( |
124 self.get_http10(body=self.chunks([('100000', part + ' X 65536')]))[ 125 'body' | 118 self.get_http10(body=self.chunks([('100000', part + ' X 65536')]))[ 119 'body' |
126 ], 127 part * 65536, | 120 ] 121 == part * 65536 |
128 ) | 122 ) |
129 self.assertEqual( | 123 assert ( |
130 self.get_http10( 131 body=self.chunks([('1000000', part + ' X 1048576')]), 132 read_buffer_size=4096 * 4096, | 124 self.get_http10( 125 body=self.chunks([('1000000', part + ' X 1048576')]), 126 read_buffer_size=4096 * 4096, |
133 )['body'], 134 part * 1048576, | 127 )['body'] 128 == part * 1048576 |
135 ) 136 | 129 ) 130 |
137 self.assertEqual( | 131 assert ( |
138 self.get_http10( 139 body=self.chunks( 140 [('1000', part + ' X 256'), ('1000', part + ' X 256')] 141 ) | 132 self.get_http10( 133 body=self.chunks( 134 [('1000', part + ' X 256'), ('1000', part + ' X 256')] 135 ) |
142 )['body'], 143 part * 256 * 2, | 136 )['body'] 137 == part * 256 * 2 |
144 ) | 138 ) |
145 self.assertEqual( | 139 assert ( |
146 self.get_http10( 147 body=self.chunks( 148 [ 149 ('100000', part + ' X 65536'), 150 ('100000', part + ' X 65536'), 151 ] 152 ) | 140 self.get_http10( 141 body=self.chunks( 142 [ 143 ('100000', part + ' X 65536'), 144 ('100000', part + ' X 65536'), 145 ] 146 ) |
153 )['body'], 154 part * 65536 * 2, | 147 )['body'] 148 == part * 65536 * 2 |
155 ) | 149 ) |
156 self.assertEqual( | 150 assert ( |
157 self.get_http10( 158 body=self.chunks( 159 [ 160 ('1000000', part + ' X 1048576'), 161 ('1000000', part + ' X 1048576'), 162 ] 163 ), 164 read_buffer_size=4096 * 4096, | 151 self.get_http10( 152 body=self.chunks( 153 [ 154 ('1000000', part + ' X 1048576'), 155 ('1000000', part + ' X 1048576'), 156 ] 157 ), 158 read_buffer_size=4096 * 4096, |
165 )['body'], 166 part * 1048576 * 2, | 159 )['body'] 160 == part * 1048576 * 2 |
167 ) 168 169 def test_proxy_chunked_fragmented(self): 170 part = '0123456789abcdef' 171 | 161 ) 162 163 def test_proxy_chunked_fragmented(self): 164 part = '0123456789abcdef' 165 |
172 self.assertEqual( | 166 assert ( |
173 self.get_http10( 174 body=self.chunks( 175 [('1', hex(i % 16)[2:]) for i in range(4096)] 176 ), | 167 self.get_http10( 168 body=self.chunks( 169 [('1', hex(i % 16)[2:]) for i in range(4096)] 170 ), |
177 )['body'], 178 part * 256, | 171 )['body'] 172 == part * 256 |
179 ) 180 181 def test_proxy_chunked_send(self): | 173 ) 174 175 def test_proxy_chunked_send(self): |
182 self.assertEqual( 183 self.get_http10(body='\r\n\r\n@0@\r\n\r\n')['status'], 200 184 ) 185 self.assertEqual( | 176 assert self.get_http10(body='\r\n\r\n@0@\r\n\r\n')['status'] == 200 177 assert ( |
186 self.get_http10( 187 body='\r@\n\r\n2\r@\na@b\r\n2\r\ncd@\r\n0\r@\n\r\n' | 178 self.get_http10( 179 body='\r@\n\r\n2\r@\na@b\r\n2\r\ncd@\r\n0\r@\n\r\n' |
188 )['body'], 189 'abcd', | 180 )['body'] 181 == 'abcd' |
190 ) | 182 ) |
191 self.assertEqual( | 183 assert ( |
192 self.get_http10( 193 body='\r\n\r\n2\r#\na#b\r\n##2\r\n#cd\r\n0\r\n#\r#\n' | 184 self.get_http10( 185 body='\r\n\r\n2\r#\na#b\r\n##2\r\n#cd\r\n0\r\n#\r#\n' |
194 )['body'], 195 'abcd', | 186 )['body'] 187 == 'abcd' |
196 ) 197 198 def test_proxy_chunked_invalid(self): 199 def check_invalid(body): | 188 ) 189 190 def test_proxy_chunked_invalid(self): 191 def check_invalid(body): |
200 self.assertNotEqual(self.get_http10(body=body)['status'], 200) | 192 assert self.get_http10(body=body)['status'] != 200 |
201 202 check_invalid('\r\n\r0') 203 check_invalid('\r\n\r\n\r0') 204 check_invalid('\r\n\r\n\r\n0') 205 check_invalid('\r\nContent-Length: 5\r\n\r\n0\r\n\r\n') 206 check_invalid('\r\n\r\n1\r\nXX\r\n0\r\n\r\n') 207 check_invalid('\r\n\r\n2\r\nX\r\n0\r\n\r\n') 208 check_invalid('\r\n\r\nH\r\nXX\r\n0\r\n\r\n') 209 check_invalid('\r\n\r\n0\r\nX') 210 211 resp = self.get_http10(body='\r\n\r\n65#\r\nA X 100') | 193 194 check_invalid('\r\n\r0') 195 check_invalid('\r\n\r\n\r0') 196 check_invalid('\r\n\r\n\r\n0') 197 check_invalid('\r\nContent-Length: 5\r\n\r\n0\r\n\r\n') 198 check_invalid('\r\n\r\n1\r\nXX\r\n0\r\n\r\n') 199 check_invalid('\r\n\r\n2\r\nX\r\n0\r\n\r\n') 200 check_invalid('\r\n\r\nH\r\nXX\r\n0\r\n\r\n') 201 check_invalid('\r\n\r\n0\r\nX') 202 203 resp = self.get_http10(body='\r\n\r\n65#\r\nA X 100') |
212 self.assertEqual(resp['status'], 200, 'incomplete chunk status') 213 self.assertNotEqual(resp['body'][-5:], '0\r\n\r\n', 'incomplete chunk') | 204 assert resp['status'] == 200, 'incomplete chunk status' 205 assert resp['body'][-5:] != '0\r\n\r\n', 'incomplete chunk' |
214 215 resp = self.get_http10(body='\r\n\r\n64#\r\nA X 100') | 206 207 resp = self.get_http10(body='\r\n\r\n64#\r\nA X 100') |
216 self.assertEqual(resp['status'], 200, 'no zero chunk status') 217 self.assertNotEqual(resp['body'][-5:], '0\r\n\r\n', 'no zero chunk') | 208 assert resp['status'] == 200, 'no zero chunk status' 209 assert resp['body'][-5:] != '0\r\n\r\n', 'no zero chunk' |
218 | 210 |
219 self.assertEqual( 220 self.get_http10(body='\r\n\r\n80000000\r\nA X 100')['status'], 200, | 211 assert ( 212 self.get_http10(body='\r\n\r\n80000000\r\nA X 100')['status'] 213 == 200 |
221 ) | 214 ) |
222 self.assertEqual( | 215 assert ( |
223 self.get_http10(body='\r\n\r\n10000000000000000\r\nA X 100')[ 224 'status' | 216 self.get_http10(body='\r\n\r\n10000000000000000\r\nA X 100')[ 217 'status' |
225 ], 226 502, | 218 ] 219 == 502 |
227 ) | 220 ) |
228 self.assertGreaterEqual( | 221 assert ( |
229 len( 230 self.get_http10( 231 body='\r\n\r\n1000000\r\nA X 1048576\r\n1000000\r\nA X 100', 232 read_buffer_size=4096 * 4096, 233 )['body'] | 222 len( 223 self.get_http10( 224 body='\r\n\r\n1000000\r\nA X 1048576\r\n1000000\r\nA X 100', 225 read_buffer_size=4096 * 4096, 226 )['body'] |
234 ), 235 1048576, | 227 ) 228 >= 1048576 |
236 ) | 229 ) |
237 self.assertGreaterEqual( | 230 assert ( |
238 len( 239 self.get_http10( 240 body='\r\n\r\n1000000\r\nA X 1048576\r\nXXX\r\nA X 100', 241 read_buffer_size=4096 * 4096, 242 )['body'] | 231 len( 232 self.get_http10( 233 body='\r\n\r\n1000000\r\nA X 1048576\r\nXXX\r\nA X 100', 234 read_buffer_size=4096 * 4096, 235 )['body'] |
243 ), 244 1048576, | 236 ) 237 >= 1048576 |
245 ) | 238 ) |
246 247 248if __name__ == '__main__': 249 TestProxyChunked.main() | |