1import unittest 2import unit 3 4class TestUnitPerlApplication(unit.TestUnitApplicationPerl): 5 6 def setUpClass(): 7 unit.TestUnit().check_modules('perl') 8 9 def test_perl_application(self): 10 self.load('variables') 11 12 body = 'Test body string.' 13 14 resp = self.post(headers={ 15 'Host': 'localhost', 16 'Content-Type': 'text/html', 17 'Custom-Header': 'blah', 18 'Connection': 'close' 19 }, body=body) 20 21 self.assertEqual(resp['status'], 200, 'status') 22 headers = resp['headers'] 23 header_server = headers.pop('Server') 24 self.assertRegex(header_server, r'Unit/[\d\.]+', 'server header') 25 self.assertEqual(headers.pop('Server-Software'), header_server, 26 'server software header') 27 28 date = headers.pop('Date') 29 self.assertEqual(date[-4:], ' GMT', 'date header timezone') 30 self.assertLess(abs(self.date_to_sec_epoch(date) - self.sec_epoch()), 5, 31 'date header') 32 33 self.assertDictEqual(headers, { 34 'Connection': 'close', 35 'Content-Length': str(len(body)), 36 'Content-Type': 'text/html', 37 'Request-Method': 'POST', 38 'Request-Uri': '/', 39 'Http-Host': 'localhost', 40 'Server-Protocol': 'HTTP/1.1', 41 'Custom-Header': 'blah', 42 'Psgi-Version': '11', 43 'Psgi-Url-Scheme': 'http', 44 'Psgi-Multithread': '', 45 'Psgi-Multiprocess': '1', 46 'Psgi-Run-Once': '', 47 'Psgi-Nonblocking': '', 48 'Psgi-Streaming': '1' 49 }, 'headers') 50 self.assertEqual(resp['body'], body, 'body') 51 52 def test_perl_application_query_string(self): 53 self.load('query_string') 54 55 resp = self.get(url='/?var1=val1&var2=val2') 56 57 self.assertEqual(resp['headers']['Query-String'], 'var1=val1&var2=val2', 58 'Query-String header') 59 60 def test_perl_application_query_string_empty(self): 61 self.load('query_string') 62 63 resp = self.get(url='/?') 64 65 self.assertEqual(resp['status'], 200, 'query string empty status') 66 self.assertEqual(resp['headers']['Query-String'], '', 67 'query string empty') 68 69 @unittest.expectedFailure 70 def test_perl_application_query_string_absent(self): 71 self.load('query_string') 72 73 resp = self.get() 74 75 self.assertEqual(resp['status'], 200, 'query string absent status') 76 self.assertEqual(resp['headers']['Query-String'], '', 77 'query string absent') 78 79 @unittest.expectedFailure 80 def test_perl_application_server_port(self): 81 self.load('server_port') 82 83 self.assertEqual(self.get()['headers']['Server-Port'], '7080', 84 'Server-Port header') 85 86 def test_perl_application_input_read_empty(self): 87 self.load('input_read_empty') 88 89 self.assertEqual(self.get()['body'], '', 'read empty') 90 91 def test_perl_application_input_read_parts(self): 92 self.load('input_read_parts') 93 94 self.assertEqual(self.post(body='0123456789')['body'], '0123456789', 95 'input read parts') 96 97 @unittest.expectedFailure 98 def test_perl_application_input_read_offset(self): 99 self.load('input_read_offset') 100 101 self.assertEqual(self.post(body='0123456789')['body'], '4567', 102 'read offset') 103 104 def test_perl_application_input_copy(self): 105 self.load('input_copy') 106 107 body = '0123456789' 108 self.assertEqual(self.post(body=body)['body'], body, 'input copy') 109 110 def test_perl_application_errors_print(self): 111 self.load('errors_print') 112 113 self.assertEqual(self.get()['body'], '1', 'errors result') 114 115 self.stop() 116 117 self.assertIsNotNone( 118 self.search_in_log(r'\[error\].+Error in application'), 119 'errors print') 120 121 def test_perl_application_header_equal_names(self): 122 self.load('header_equal_names') 123 124 self.assertListEqual(self.get()['headers']['Set-Cookie'], 125 ['tc=one,two,three', 'tc=four,five,six'], 'header equal names') 126 127 def test_perl_application_header_pairs(self): 128 self.load('header_pairs') 129 130 self.assertEqual(self.get()['headers']['blah'], 'blah', 'header pairs') 131 132 def test_perl_application_body_empty(self): 133 self.load('body_empty') 134 135 self.assertEqual(self.get()['body'], '0\r\n\r\n', 'body empty') 136 137 def test_perl_application_body_array(self): 138 self.load('body_array') 139 140 self.assertEqual(self.get()['body'], '0123456789', 'body array') 141 142 def test_perl_application_body_large(self): 143 self.load('variables') 144 145 body = '0123456789' * 1000 146 147 resp = self.post(body=body)['body'] 148 149 self.assertEqual(resp, body, 'body large') 150 151 def test_perl_application_body_io_empty(self): 152 self.load('body_io_empty') 153 154 self.assertEqual(self.get()['status'], 200, 'body io empty') 155 156 def test_perl_application_body_io_file(self): 157 self.load('body_io_file') 158 159 self.assertEqual(self.get()['body'], 'body\n', 'body io file') 160 161 @unittest.expectedFailure 162 def test_perl_application_syntax_error(self): 163 self.skip_alerts.extend([ 164 r'PSGI: Failed to parse script', 165 r'process \d+ exited on signal' 166 ]) 167 self.load('syntax_error') 168 169 self.assertEqual(self.get()['status'], 500, 'syntax error') 170 171 def test_perl_keepalive_body(self): 172 self.load('variables') 173 174 (resp, sock) = self.post(headers={ 175 'Host': 'localhost', 176 'Connection': 'keep-alive', 177 'Content-Type': 'text/html' 178 }, start=True, body='0123456789' * 500) 179 180 self.assertEqual(resp['body'], '0123456789' * 500, 'keep-alive 1') 181 182 resp = self.post(headers={ 183 'Host': 'localhost', 184 'Connection': 'close', 185 'Content-Type': 'text/html' 186 }, sock=sock, body='0123456789') 187 188 self.assertEqual(resp['body'], '0123456789', 'keep-alive 2') 189 190 def test_perl_body_io_fake(self): 191 self.load('body_io_fake') 192 193 self.assertEqual(self.get()['body'], '21', 'body io fake') 194 195 self.assertIsNotNone( 196 self.search_in_log(r'\[error\].+IOFake getline\(\) \$\/ is \d+'), 197 'body io fake $/ value') 198 199 self.assertIsNotNone( 200 self.search_in_log(r'\[error\].+IOFake close\(\) called'), 201 'body io fake close') 202 203 def test_perl_delayed_response(self): 204 self.load('delayed_response') 205 206 resp = self.get() 207 208 self.assertEqual(resp['status'], 200, 'status') 209 self.assertEqual(resp['body'], 'Hello World!', 'body') 210 211 def test_perl_streaming_body(self): 212 self.load('streaming_body') 213 214 resp = self.get() 215 216 self.assertEqual(resp['status'], 200, 'status') 217 self.assertEqual(resp['body'], 'Hello World!', 'body') 218 219if __name__ == '__main__': 220 TestUnitPerlApplication.main() 221