1484Szelenkov@nginx.comimport unittest
2484Szelenkov@nginx.comimport unit
3484Szelenkov@nginx.com
4552Szelenkov@nginx.comclass TestUnitPythonApplication(unit.TestUnitApplicationPython):
5484Szelenkov@nginx.com
6484Szelenkov@nginx.com    def setUpClass():
7550Szelenkov@nginx.com        unit.TestUnit().check_modules('python')
8484Szelenkov@nginx.com
9552Szelenkov@nginx.com    def test_python_application_variables(self):
10552Szelenkov@nginx.com        self.load('variables')
11484Szelenkov@nginx.com
12484Szelenkov@nginx.com        body = 'Test body string.'
13484Szelenkov@nginx.com
14505Szelenkov@nginx.com        resp = self.post(headers={
15484Szelenkov@nginx.com            'Host': 'localhost',
16484Szelenkov@nginx.com            'Content-Type': 'text/html',
17484Szelenkov@nginx.com            'Custom-Header': 'blah'
18505Szelenkov@nginx.com        }, body=body)
19484Szelenkov@nginx.com
20505Szelenkov@nginx.com        self.assertEqual(resp['status'], 200, 'status')
21505Szelenkov@nginx.com        headers = resp['headers']
22*674Szelenkov@nginx.com        header_server = headers.pop('Server')
23*674Szelenkov@nginx.com        self.assertRegex(header_server, r'Unit/[\d\.]+', 'server header')
24*674Szelenkov@nginx.com        self.assertEqual(headers.pop('Server-Software'), header_server,
25*674Szelenkov@nginx.com            'server software header')
26599Szelenkov@nginx.com
27599Szelenkov@nginx.com        date = headers.pop('Date')
28599Szelenkov@nginx.com        self.assertEqual(date[-4:], ' GMT', 'date header timezone')
29599Szelenkov@nginx.com        self.assertLess(abs(self.date_to_sec_epoch(date) - self.sec_epoch()), 5,
30599Szelenkov@nginx.com            'date header')
31599Szelenkov@nginx.com
32495Szelenkov@nginx.com        self.assertDictEqual(headers, {
33495Szelenkov@nginx.com            'Content-Length': str(len(body)),
34495Szelenkov@nginx.com            'Content-Type': 'text/html',
35495Szelenkov@nginx.com            'Request-Method': 'POST',
36495Szelenkov@nginx.com            'Request-Uri': '/',
37495Szelenkov@nginx.com            'Http-Host': 'localhost',
38495Szelenkov@nginx.com            'Server-Protocol': 'HTTP/1.1',
39603Szelenkov@nginx.com            'Custom-Header': 'blah',
40603Szelenkov@nginx.com            'Wsgi-Version': '(1, 0)',
41603Szelenkov@nginx.com            'Wsgi-Url-Scheme': 'http',
42603Szelenkov@nginx.com            'Wsgi-Multithread': 'False',
43603Szelenkov@nginx.com            'Wsgi-Multiprocess': 'True',
44603Szelenkov@nginx.com            'Wsgi-Run-Once': 'False'
45495Szelenkov@nginx.com        }, 'headers')
46505Szelenkov@nginx.com        self.assertEqual(resp['body'], body, 'body')
47484Szelenkov@nginx.com
48497Szelenkov@nginx.com    def test_python_application_query_string(self):
49552Szelenkov@nginx.com        self.load('query_string')
50497Szelenkov@nginx.com
51505Szelenkov@nginx.com        resp = self.get(url='/?var1=val1&var2=val2')
52497Szelenkov@nginx.com
53513Szelenkov@nginx.com        self.assertEqual(resp['headers']['Query-String'], 'var1=val1&var2=val2',
54513Szelenkov@nginx.com            'Query-String header')
55497Szelenkov@nginx.com
56495Szelenkov@nginx.com    @unittest.expectedFailure
57495Szelenkov@nginx.com    def test_python_application_server_port(self):
58552Szelenkov@nginx.com        self.load('server_port')
59495Szelenkov@nginx.com
60505Szelenkov@nginx.com        self.assertEqual(self.get()['headers']['Server-Port'], '7080',
61495Szelenkov@nginx.com            'Server-Port header')
62484Szelenkov@nginx.com
63496Szelenkov@nginx.com    @unittest.expectedFailure
64496Szelenkov@nginx.com    def test_python_application_204_transfer_encoding(self):
65552Szelenkov@nginx.com        self.load('204_no_content')
66496Szelenkov@nginx.com
67505Szelenkov@nginx.com        self.assertNotIn('Transfer-Encoding', self.get()['headers'],
68496Szelenkov@nginx.com            '204 header transfer encoding')
69484Szelenkov@nginx.com
70602Szelenkov@nginx.com    def test_python_application_ctx_iter_atexit(self):
71602Szelenkov@nginx.com        self.skip_alerts.append(r'sendmsg.+failed')
72602Szelenkov@nginx.com        self.load('ctx_iter_atexit')
73602Szelenkov@nginx.com
74602Szelenkov@nginx.com        resp = self.post(headers={
75602Szelenkov@nginx.com            'Connection': 'close',
76602Szelenkov@nginx.com            'Content-Type': 'text/html',
77602Szelenkov@nginx.com            'Host': 'localhost'
78602Szelenkov@nginx.com        }, body='0123456789')
79602Szelenkov@nginx.com
80602Szelenkov@nginx.com        self.assertEqual(resp['status'], 200, 'ctx iter status')
81602Szelenkov@nginx.com        self.assertEqual(resp['body'], '0123456789', 'ctx iter body')
82602Szelenkov@nginx.com
83602Szelenkov@nginx.com        self.conf({
84602Szelenkov@nginx.com            "listeners": {},
85602Szelenkov@nginx.com            "applications": {}
86602Szelenkov@nginx.com        })
87602Szelenkov@nginx.com
88602Szelenkov@nginx.com        self.stop()
89602Szelenkov@nginx.com
90602Szelenkov@nginx.com        self.assertIsNotNone(self.search_in_log(r'RuntimeError'),
91602Szelenkov@nginx.com            'ctx iter atexit')
92602Szelenkov@nginx.com
93603Szelenkov@nginx.com    def test_python_keepalive_body(self):
94603Szelenkov@nginx.com        self.load('mirror')
95603Szelenkov@nginx.com
96603Szelenkov@nginx.com        (resp, sock) = self.post(headers={
97603Szelenkov@nginx.com            'Connection': 'keep-alive',
98603Szelenkov@nginx.com            'Content-Type': 'text/html',
99603Szelenkov@nginx.com            'Host': 'localhost'
100603Szelenkov@nginx.com        }, start=True, body='0123456789' * 500)
101603Szelenkov@nginx.com
102603Szelenkov@nginx.com        self.assertEqual(resp['body'], '0123456789' * 500, 'keep-alive 1')
103603Szelenkov@nginx.com
104603Szelenkov@nginx.com        resp = self.post(headers={
105603Szelenkov@nginx.com            'Connection': 'close',
106603Szelenkov@nginx.com            'Content-Type': 'text/html',
107603Szelenkov@nginx.com            'Host': 'localhost'
108603Szelenkov@nginx.com        }, sock=sock, body='0123456789')
109603Szelenkov@nginx.com
110603Szelenkov@nginx.com        self.assertEqual(resp['body'], '0123456789', 'keep-alive 2')
111603Szelenkov@nginx.com
112603Szelenkov@nginx.com    def test_python_atexit(self):
113603Szelenkov@nginx.com        self.skip_alerts.append(r'sendmsg.+failed')
114603Szelenkov@nginx.com        self.load('atexit')
115603Szelenkov@nginx.com
116603Szelenkov@nginx.com        self.get()
117603Szelenkov@nginx.com
118603Szelenkov@nginx.com        self.conf({
119603Szelenkov@nginx.com            "listeners": {},
120603Szelenkov@nginx.com            "applications": {}
121603Szelenkov@nginx.com        })
122603Szelenkov@nginx.com
123603Szelenkov@nginx.com        self.stop()
124603Szelenkov@nginx.com
125603Szelenkov@nginx.com        self.assertIsNotNone(self.search_in_log(r'At exit called\.'), 'atexit')
126603Szelenkov@nginx.com
127603Szelenkov@nginx.com    @unittest.expectedFailure
128603Szelenkov@nginx.com    def test_python_application_start_response_exit(self):
129603Szelenkov@nginx.com        self.load('start_response_exit')
130603Szelenkov@nginx.com
131603Szelenkov@nginx.com        self.assertEqual(self.get()['status'], 500, 'start response exit')
132603Szelenkov@nginx.com
133603Szelenkov@nginx.com    @unittest.expectedFailure
134603Szelenkov@nginx.com    def test_python_application_input_iter(self):
135603Szelenkov@nginx.com        self.load('input_iter')
136603Szelenkov@nginx.com
137603Szelenkov@nginx.com        body = '0123456789'
138603Szelenkov@nginx.com
139603Szelenkov@nginx.com        self.assertEqual(self.post(body=body)['body'], body, 'input iter')
140603Szelenkov@nginx.com
141603Szelenkov@nginx.com    @unittest.expectedFailure
142603Szelenkov@nginx.com    def test_python_application_input_read_length(self):
143603Szelenkov@nginx.com        self.load('input_read_length')
144603Szelenkov@nginx.com
145603Szelenkov@nginx.com        body = '0123456789'
146603Szelenkov@nginx.com
147603Szelenkov@nginx.com        resp = self.post(headers={
148603Szelenkov@nginx.com            'Host': 'localhost',
149603Szelenkov@nginx.com            'Input-Length': '5',
150603Szelenkov@nginx.com            'Connection': 'close'
151603Szelenkov@nginx.com        }, body=body)
152603Szelenkov@nginx.com
153603Szelenkov@nginx.com        self.assertEqual(resp['body'], body[:5], 'input read length lt body')
154603Szelenkov@nginx.com
155603Szelenkov@nginx.com        resp = self.post(headers={
156603Szelenkov@nginx.com            'Host': 'localhost',
157603Szelenkov@nginx.com            'Input-Length': '15',
158603Szelenkov@nginx.com            'Connection': 'close'
159603Szelenkov@nginx.com        }, body=body)
160603Szelenkov@nginx.com
161603Szelenkov@nginx.com        self.assertEqual(resp['body'], body, 'input read length gt body')
162603Szelenkov@nginx.com
163603Szelenkov@nginx.com        resp = self.post(headers={
164603Szelenkov@nginx.com            'Host': 'localhost',
165603Szelenkov@nginx.com            'Input-Length': '0',
166603Szelenkov@nginx.com            'Connection': 'close'
167603Szelenkov@nginx.com        }, body=body)
168603Szelenkov@nginx.com
169603Szelenkov@nginx.com        self.assertEqual(resp['body'], '', 'input read length zero')
170603Szelenkov@nginx.com
171603Szelenkov@nginx.com        resp = self.post(headers={
172603Szelenkov@nginx.com            'Host': 'localhost',
173603Szelenkov@nginx.com            'Input-Length': '-1',
174603Szelenkov@nginx.com            'Connection': 'close'
175603Szelenkov@nginx.com        }, body=body)
176603Szelenkov@nginx.com
177603Szelenkov@nginx.com        self.assertEqual(resp['body'], body, 'input read length negative')
178603Szelenkov@nginx.com
179603Szelenkov@nginx.com    @unittest.expectedFailure
180603Szelenkov@nginx.com    def test_python_application_errors_write(self):
181603Szelenkov@nginx.com        self.load('errors_write')
182603Szelenkov@nginx.com
183603Szelenkov@nginx.com        self.get()
184603Szelenkov@nginx.com
185603Szelenkov@nginx.com        self.stop()
186603Szelenkov@nginx.com
187603Szelenkov@nginx.com        self.assertIsNotNone(
188603Szelenkov@nginx.com            self.search_in_log(r'\[error\].+Error in application\.'),
189603Szelenkov@nginx.com            'errors write')
190603Szelenkov@nginx.com
191603Szelenkov@nginx.com    def test_python_application_body_array(self):
192603Szelenkov@nginx.com        self.load('body_array')
193603Szelenkov@nginx.com
194603Szelenkov@nginx.com        self.assertEqual(self.get()['body'], '0123456789', 'body array')
195603Szelenkov@nginx.com
196603Szelenkov@nginx.com    def test_python_application_body_io(self):
197603Szelenkov@nginx.com        self.load('body_io')
198603Szelenkov@nginx.com
199603Szelenkov@nginx.com        self.assertEqual(self.get()['body'], '0123456789', 'body io')
200603Szelenkov@nginx.com
201603Szelenkov@nginx.com    def test_python_application_body_io_file(self):
202603Szelenkov@nginx.com        self.load('body_io_file')
203603Szelenkov@nginx.com
204603Szelenkov@nginx.com        self.assertEqual(self.get()['body'], 'body\n', 'body io file')
205603Szelenkov@nginx.com
206603Szelenkov@nginx.com    @unittest.expectedFailure
207603Szelenkov@nginx.com    def test_python_application_syntax_error(self):
208603Szelenkov@nginx.com        self.skip_alerts.append(r'Python failed to import module "wsgi"')
209603Szelenkov@nginx.com        self.load('syntax_error')
210603Szelenkov@nginx.com
211603Szelenkov@nginx.com        self.assertEqual(self.get()['status'], 500, 'syntax error')
212603Szelenkov@nginx.com
213603Szelenkov@nginx.com    def test_python_application_close(self):
214603Szelenkov@nginx.com        self.load('close')
215603Szelenkov@nginx.com
216603Szelenkov@nginx.com        self.get()
217603Szelenkov@nginx.com
218603Szelenkov@nginx.com        self.stop()
219603Szelenkov@nginx.com
220603Szelenkov@nginx.com        self.assertIsNotNone(self.search_in_log(r'Close called\.'), 'close')
221603Szelenkov@nginx.com
222603Szelenkov@nginx.com    def test_python_application_close_error(self):
223603Szelenkov@nginx.com        self.load('close_error')
224603Szelenkov@nginx.com
225603Szelenkov@nginx.com        self.get()
226603Szelenkov@nginx.com
227603Szelenkov@nginx.com        self.stop()
228603Szelenkov@nginx.com
229603Szelenkov@nginx.com        self.assertIsNotNone(self.search_in_log(r'Close called\.'),
230603Szelenkov@nginx.com            'close error')
231603Szelenkov@nginx.com
232617Szelenkov@nginx.com    def test_python_application_not_iterable(self):
233617Szelenkov@nginx.com        self.load('not_iterable')
234617Szelenkov@nginx.com
235665Szelenkov@nginx.com        self.get()
236617Szelenkov@nginx.com
237617Szelenkov@nginx.com        self.stop()
238617Szelenkov@nginx.com
239617Szelenkov@nginx.com        self.assertIsNotNone(self.search_in_log(
240617Szelenkov@nginx.com            r'\[error\].+the application returned not an iterable object'),
241617Szelenkov@nginx.com            'not iterable')
242617Szelenkov@nginx.com
243664Szelenkov@nginx.com    def test_python_application_write(self):
244664Szelenkov@nginx.com        self.load('write')
245664Szelenkov@nginx.com
246664Szelenkov@nginx.com        self.assertEqual(self.get()['body'], '0123456789', 'write')
247664Szelenkov@nginx.com
248484Szelenkov@nginx.comif __name__ == '__main__':
249484Szelenkov@nginx.com    unittest.main()
250