1import re 2 3import pytest 4 5from unit.applications.lang.perl import ApplicationPerl 6 7prerequisites = {'modules': {'perl': 'all'}} 8 9client = ApplicationPerl() 10 11 12def test_perl_application(date_to_sec_epoch, sec_epoch): 13 client.load('variables') 14 15 body = 'Test body string.' 16 17 resp = client.post( 18 headers={ 19 'Host': 'localhost', 20 'Content-Type': 'text/html', 21 'Custom-Header': 'blah', 22 'Connection': 'close', 23 }, 24 body=body, 25 ) 26 27 assert resp['status'] == 200, 'status' 28 headers = resp['headers'] 29 header_server = headers.pop('Server') 30 assert re.search(r'Unit/[\d\.]+', header_server), 'server header' 31 assert ( 32 headers.pop('Server-Software') == header_server 33 ), 'server software header' 34 35 date = headers.pop('Date') 36 assert date[-4:] == ' GMT', 'date header timezone' 37 assert abs(date_to_sec_epoch(date) - sec_epoch) < 5, 'date header' 38 39 assert headers == { 40 'Connection': 'close', 41 'Content-Length': str(len(body)), 42 'Content-Type': 'text/html', 43 'Request-Method': 'POST', 44 'Request-Uri': '/', 45 'Http-Host': 'localhost', 46 'Server-Protocol': 'HTTP/1.1', 47 'Custom-Header': 'blah', 48 'Psgi-Version': '11', 49 'Psgi-Url-Scheme': 'http', 50 'Psgi-Multithread': '', 51 'Psgi-Multiprocess': '1', 52 'Psgi-Run-Once': '', 53 'Psgi-Nonblocking': '', 54 'Psgi-Streaming': '1', 55 }, 'headers' 56 assert resp['body'] == body, 'body' 57 58 59def test_perl_application_query_string(): 60 client.load('query_string') 61 62 resp = client.get(url='/?var1=val1&var2=val2') 63 64 assert ( 65 resp['headers']['Query-String'] == 'var1=val1&var2=val2' 66 ), 'Query-String header' 67 68 69def test_perl_application_query_string_empty(): 70 client.load('query_string') 71 72 resp = client.get(url='/?') 73 74 assert resp['status'] == 200, 'query string empty status' 75 assert resp['headers']['Query-String'] == '', 'query string empty' 76 77 78def test_perl_application_query_string_absent(): 79 client.load('query_string') 80 81 resp = client.get() 82 83 assert resp['status'] == 200, 'query string absent status' 84 assert resp['headers']['Query-String'] == '', 'query string absent' 85 86 87@pytest.mark.skip('not yet') 88def test_perl_application_server_port(): 89 client.load('server_port') 90 91 assert ( 92 client.get()['headers']['Server-Port'] == '8080' 93 ), 'Server-Port header' 94 95 96def test_perl_application_input_read_empty(): 97 client.load('input_read_empty') 98 99 assert client.get()['body'] == '', 'read empty' 100 101 102def test_perl_application_input_read_parts(): 103 client.load('input_read_parts') 104 105 assert ( 106 client.post(body='0123456789')['body'] == '0123456789' 107 ), 'input read parts' 108 109 110def test_perl_application_input_buffered_read(): 111 client.load('input_buffered_read') 112 113 assert client.post(body='012345')['body'] == '012345', 'buffered read #1' 114 assert ( 115 client.post(body='9876543210')['body'] == '9876543210' 116 ), 'buffered read #2' 117 118 119def test_perl_application_input_close(): 120 client.load('input_close') 121 122 assert client.post(body='012345')['body'] == '012345', 'input close #1' 123 assert ( 124 client.post(body='9876543210')['body'] == '9876543210' 125 ), 'input close #2' 126 127 128@pytest.mark.skip('not yet') 129def test_perl_application_input_read_offset(): 130 client.load('input_read_offset') 131 132 assert client.post(body='0123456789')['body'] == '4567', 'read offset' 133 134 135def test_perl_application_input_copy(): 136 client.load('input_copy') 137 138 body = '0123456789' 139 assert client.post(body=body)['body'] == body, 'input copy' 140 141 142def test_perl_application_errors_print(wait_for_record): 143 client.load('errors_print') 144 145 assert client.get()['body'] == '1', 'errors result' 146 147 assert ( 148 wait_for_record(r'\[error\].+Error in application') is not None 149 ), 'errors print' 150 151 152def test_perl_application_header_equal_names(): 153 client.load('header_equal_names') 154 155 assert client.get()['headers']['Set-Cookie'] == [ 156 'tc=one,two,three', 157 'tc=four,five,six', 158 ], 'header equal names' 159 160 161def test_perl_application_header_pairs(): 162 client.load('header_pairs') 163 164 assert client.get()['headers']['blah'] == 'blah', 'header pairs' 165 166 167def test_perl_application_body_empty(): 168 client.load('body_empty') 169 170 assert client.get()['body'] == '', 'body empty' 171 172 173def test_perl_application_body_array(): 174 client.load('body_array') 175 176 assert client.get()['body'] == '0123456789', 'body array' 177 178 179def test_perl_application_body_large(): 180 client.load('variables') 181 182 body = '0123456789' * 1000 183 184 resp = client.post(body=body)['body'] 185 186 assert resp == body, 'body large' 187 188 189def test_perl_application_body_io_empty(): 190 client.load('body_io_empty') 191 192 assert client.get()['status'] == 200, 'body io empty' 193 194 195def test_perl_application_body_io_file(): 196 client.load('body_io_file') 197 198 assert client.get()['body'] == 'body\n', 'body io file' 199 200 201def test_perl_streaming_body_multiple_responses(): 202 client.load('streaming_body_multiple_responses') 203 204 assert client.get()['status'] == 200 205 206 207@pytest.mark.skip('not yet') 208def test_perl_application_syntax_error(skip_alert): 209 skip_alert(r'PSGI: Failed to parse script') 210 client.load('syntax_error') 211 212 assert client.get()['status'] == 500, 'syntax error' 213 214 215def test_perl_keepalive_body(): 216 client.load('variables') 217 218 assert client.get()['status'] == 200, 'init' 219 220 body = '0123456789' * 500 221 (resp, sock) = client.post( 222 headers={ 223 'Host': 'localhost', 224 'Connection': 'keep-alive', 225 'Content-Type': 'text/html', 226 }, 227 start=True, 228 body=body, 229 read_timeout=1, 230 ) 231 232 assert resp['body'] == body, 'keep-alive 1' 233 234 body = '0123456789' 235 resp = client.post( 236 headers={ 237 'Host': 'localhost', 238 'Connection': 'close', 239 'Content-Type': 'text/html', 240 }, 241 sock=sock, 242 body=body, 243 ) 244 245 assert resp['body'] == body, 'keep-alive 2' 246 247 248def test_perl_body_io_fake(wait_for_record): 249 client.load('body_io_fake') 250 251 assert client.get()['body'] == '21', 'body io fake' 252 253 assert ( 254 wait_for_record(r'\[error\].+IOFake getline\(\) \$\/ is \d+') 255 is not None 256 ), 'body io fake $/ value' 257 258 assert ( 259 wait_for_record(r'\[error\].+IOFake close\(\) called') is not None 260 ), 'body io fake close' 261 262 263def test_perl_delayed_response(): 264 client.load('delayed_response') 265 266 resp = client.get() 267 268 assert resp['status'] == 200, 'status' 269 assert resp['body'] == 'Hello World!', 'body' 270 271 272def test_perl_streaming_body(): 273 client.load('streaming_body') 274 275 resp = client.get() 276 277 assert resp['status'] == 200, 'status' 278 assert resp['body'] == 'Hello World!', 'body' 279 280 281def test_perl_application_threads(): 282 client.load('threads') 283 284 assert 'success' in client.conf( 285 '4', 'applications/threads/threads' 286 ), 'configure 4 threads' 287 288 socks = [] 289 290 for _ in range(4): 291 sock = client.get( 292 headers={ 293 'Host': 'localhost', 294 'X-Delay': '2', 295 'Connection': 'close', 296 }, 297 no_recv=True, 298 ) 299 300 socks.append(sock) 301 302 threads = set() 303 304 for sock in socks: 305 resp = client.recvall(sock).decode('utf-8') 306 307 client.log_in(resp) 308 309 resp = client._resp_to_dict(resp) 310 311 assert resp['status'] == 200, 'status' 312 313 threads.add(resp['headers']['X-Thread']) 314 315 assert resp['headers']['Psgi-Multithread'] == '1', 'multithread' 316 317 sock.close() 318 319 assert len(socks) == len(threads), 'threads differs' 320