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']
22674Szelenkov@nginx.com        header_server = headers.pop('Server')
23674Szelenkov@nginx.com        self.assertRegex(header_server, r'Unit/[\d\.]+', 'server header')
24674Szelenkov@nginx.com        self.assertEqual(headers.pop('Server-Software'), header_server,
25674Szelenkov@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
112684Szelenkov@nginx.com    def test_python_keepalive_reconfigure(self):
113684Szelenkov@nginx.com        self.skip_alerts.extend([
114684Szelenkov@nginx.com            r'sendmsg.+failed',
115684Szelenkov@nginx.com            r'recvmsg.+failed'
116684Szelenkov@nginx.com        ])
117684Szelenkov@nginx.com        self.load('mirror')
118684Szelenkov@nginx.com
119684Szelenkov@nginx.com        body = '0123456789'
120684Szelenkov@nginx.com        conns = 3
121684Szelenkov@nginx.com        socks = []
122684Szelenkov@nginx.com
123684Szelenkov@nginx.com        for i in range(conns):
124684Szelenkov@nginx.com            (resp, sock) = self.post(headers={
125684Szelenkov@nginx.com                'Connection': 'keep-alive',
126684Szelenkov@nginx.com                'Content-Type': 'text/html',
127684Szelenkov@nginx.com                'Host': 'localhost'
128684Szelenkov@nginx.com            }, start=True, body=body)
129684Szelenkov@nginx.com
130684Szelenkov@nginx.com            self.assertEqual(resp['body'], body, 'keep-alive open')
131684Szelenkov@nginx.com            self.assertIn('success', self.conf({
132684Szelenkov@nginx.com                "spare": i % 4,
133684Szelenkov@nginx.com                "max": (i % 4) + 1
134684Szelenkov@nginx.com            }, '/applications/mirror/processes'), 'reconfigure')
135684Szelenkov@nginx.com
136684Szelenkov@nginx.com            socks.append(sock)
137684Szelenkov@nginx.com
138684Szelenkov@nginx.com        for i in range(conns):
139684Szelenkov@nginx.com            (resp, sock) = self.post(headers={
140684Szelenkov@nginx.com                'Connection': 'keep-alive',
141684Szelenkov@nginx.com                'Content-Type': 'text/html',
142684Szelenkov@nginx.com                'Host': 'localhost'
143684Szelenkov@nginx.com            }, start=True, sock=socks[i], body=body)
144684Szelenkov@nginx.com
145684Szelenkov@nginx.com            self.assertEqual(resp['body'], body, 'keep-alive request')
146684Szelenkov@nginx.com            self.assertIn('success', self.conf({
147684Szelenkov@nginx.com                "spare": i % 4,
148684Szelenkov@nginx.com                "max": (i % 4) + 1
149684Szelenkov@nginx.com            }, '/applications/mirror/processes'), 'reconfigure 2')
150684Szelenkov@nginx.com
151684Szelenkov@nginx.com        for i in range(conns):
152684Szelenkov@nginx.com            resp = self.post(headers={
153684Szelenkov@nginx.com                'Connection': 'close',
154684Szelenkov@nginx.com                'Content-Type': 'text/html',
155684Szelenkov@nginx.com                'Host': 'localhost'
156684Szelenkov@nginx.com            }, sock=socks[i], body=body)
157684Szelenkov@nginx.com
158684Szelenkov@nginx.com            self.assertEqual(resp['body'], body, 'keep-alive close')
159684Szelenkov@nginx.com            self.assertIn('success', self.conf({
160684Szelenkov@nginx.com                "spare": i % 4,
161684Szelenkov@nginx.com                "max": (i % 4) + 1
162684Szelenkov@nginx.com            }, '/applications/mirror/processes'), 'reconfigure 3')
163684Szelenkov@nginx.com
164*750Szelenkov@nginx.com    def test_python_keepalive_reconfigure_2(self):
165*750Szelenkov@nginx.com        self.skip_alerts.append(r'sendmsg.+failed')
166*750Szelenkov@nginx.com        self.load('mirror')
167*750Szelenkov@nginx.com
168*750Szelenkov@nginx.com        body = '0123456789'
169*750Szelenkov@nginx.com
170*750Szelenkov@nginx.com        (resp, sock) = self.post(headers={
171*750Szelenkov@nginx.com            'Connection': 'keep-alive',
172*750Szelenkov@nginx.com            'Content-Type': 'text/html',
173*750Szelenkov@nginx.com            'Host': 'localhost'
174*750Szelenkov@nginx.com        }, start=True, body=body)
175*750Szelenkov@nginx.com
176*750Szelenkov@nginx.com        self.assertEqual(resp['body'], body, 'reconfigure 2 keep-alive 1')
177*750Szelenkov@nginx.com
178*750Szelenkov@nginx.com        self.load('empty')
179*750Szelenkov@nginx.com
180*750Szelenkov@nginx.com        (resp, sock) = self.post(headers={
181*750Szelenkov@nginx.com            'Connection': 'close',
182*750Szelenkov@nginx.com            'Content-Type': 'text/html',
183*750Szelenkov@nginx.com            'Host': 'localhost'
184*750Szelenkov@nginx.com        }, start=True, sock=sock, body=body)
185*750Szelenkov@nginx.com
186*750Szelenkov@nginx.com        self.assertEqual(resp['status'], 200, 'reconfigure 2 keep-alive 2')
187*750Szelenkov@nginx.com        self.assertEqual(resp['body'], '', 'reconfigure 2 keep-alive 2 body')
188*750Szelenkov@nginx.com
189*750Szelenkov@nginx.com        self.assertIn('success', self.conf({
190*750Szelenkov@nginx.com            "listeners": {},
191*750Szelenkov@nginx.com            "applications": {}
192*750Szelenkov@nginx.com        }), 'reconfigure 2 clear configuration')
193*750Szelenkov@nginx.com
194*750Szelenkov@nginx.com        resp = self.get(sock=sock)
195*750Szelenkov@nginx.com
196*750Szelenkov@nginx.com        self.assertEqual(resp, {}, 'reconfigure 2 keep-alive 3')
197*750Szelenkov@nginx.com
198*750Szelenkov@nginx.com    @unittest.expectedFailure
199*750Szelenkov@nginx.com    def test_python_keepalive_reconfigure_3(self):
200*750Szelenkov@nginx.com        self.load('empty')
201*750Szelenkov@nginx.com
202*750Szelenkov@nginx.com        (resp, sock) = self.http(b"""GET / HTTP/1.1
203*750Szelenkov@nginx.com""", start=True, raw=True)
204*750Szelenkov@nginx.com
205*750Szelenkov@nginx.com        self.assertIn('success', self.conf({
206*750Szelenkov@nginx.com            "listeners": {},
207*750Szelenkov@nginx.com            "applications": {}
208*750Szelenkov@nginx.com        }), 'reconfigure 3 clear configuration')
209*750Szelenkov@nginx.com
210*750Szelenkov@nginx.com        resp = self.http(b"""Host: localhost
211*750Szelenkov@nginx.comConnection: close
212*750Szelenkov@nginx.com
213*750Szelenkov@nginx.com""", sock=sock, raw=True)
214*750Szelenkov@nginx.com
215*750Szelenkov@nginx.com        self.assertEqual(resp['status'], 200, 'reconfigure 3')
216*750Szelenkov@nginx.com
217603Szelenkov@nginx.com    def test_python_atexit(self):
218603Szelenkov@nginx.com        self.skip_alerts.append(r'sendmsg.+failed')
219603Szelenkov@nginx.com        self.load('atexit')
220603Szelenkov@nginx.com
221603Szelenkov@nginx.com        self.get()
222603Szelenkov@nginx.com
223603Szelenkov@nginx.com        self.conf({
224603Szelenkov@nginx.com            "listeners": {},
225603Szelenkov@nginx.com            "applications": {}
226603Szelenkov@nginx.com        })
227603Szelenkov@nginx.com
228603Szelenkov@nginx.com        self.stop()
229603Szelenkov@nginx.com
230603Szelenkov@nginx.com        self.assertIsNotNone(self.search_in_log(r'At exit called\.'), 'atexit')
231603Szelenkov@nginx.com
232603Szelenkov@nginx.com    @unittest.expectedFailure
233603Szelenkov@nginx.com    def test_python_application_start_response_exit(self):
234603Szelenkov@nginx.com        self.load('start_response_exit')
235603Szelenkov@nginx.com
236603Szelenkov@nginx.com        self.assertEqual(self.get()['status'], 500, 'start response exit')
237603Szelenkov@nginx.com
238603Szelenkov@nginx.com    @unittest.expectedFailure
239603Szelenkov@nginx.com    def test_python_application_input_iter(self):
240603Szelenkov@nginx.com        self.load('input_iter')
241603Szelenkov@nginx.com
242603Szelenkov@nginx.com        body = '0123456789'
243603Szelenkov@nginx.com
244603Szelenkov@nginx.com        self.assertEqual(self.post(body=body)['body'], body, 'input iter')
245603Szelenkov@nginx.com
246603Szelenkov@nginx.com    @unittest.expectedFailure
247603Szelenkov@nginx.com    def test_python_application_input_read_length(self):
248603Szelenkov@nginx.com        self.load('input_read_length')
249603Szelenkov@nginx.com
250603Szelenkov@nginx.com        body = '0123456789'
251603Szelenkov@nginx.com
252603Szelenkov@nginx.com        resp = self.post(headers={
253603Szelenkov@nginx.com            'Host': 'localhost',
254603Szelenkov@nginx.com            'Input-Length': '5',
255603Szelenkov@nginx.com            'Connection': 'close'
256603Szelenkov@nginx.com        }, body=body)
257603Szelenkov@nginx.com
258603Szelenkov@nginx.com        self.assertEqual(resp['body'], body[:5], 'input read length lt body')
259603Szelenkov@nginx.com
260603Szelenkov@nginx.com        resp = self.post(headers={
261603Szelenkov@nginx.com            'Host': 'localhost',
262603Szelenkov@nginx.com            'Input-Length': '15',
263603Szelenkov@nginx.com            'Connection': 'close'
264603Szelenkov@nginx.com        }, body=body)
265603Szelenkov@nginx.com
266603Szelenkov@nginx.com        self.assertEqual(resp['body'], body, 'input read length gt body')
267603Szelenkov@nginx.com
268603Szelenkov@nginx.com        resp = self.post(headers={
269603Szelenkov@nginx.com            'Host': 'localhost',
270603Szelenkov@nginx.com            'Input-Length': '0',
271603Szelenkov@nginx.com            'Connection': 'close'
272603Szelenkov@nginx.com        }, body=body)
273603Szelenkov@nginx.com
274603Szelenkov@nginx.com        self.assertEqual(resp['body'], '', 'input read length zero')
275603Szelenkov@nginx.com
276603Szelenkov@nginx.com        resp = self.post(headers={
277603Szelenkov@nginx.com            'Host': 'localhost',
278603Szelenkov@nginx.com            'Input-Length': '-1',
279603Szelenkov@nginx.com            'Connection': 'close'
280603Szelenkov@nginx.com        }, body=body)
281603Szelenkov@nginx.com
282603Szelenkov@nginx.com        self.assertEqual(resp['body'], body, 'input read length negative')
283603Szelenkov@nginx.com
284603Szelenkov@nginx.com    @unittest.expectedFailure
285603Szelenkov@nginx.com    def test_python_application_errors_write(self):
286603Szelenkov@nginx.com        self.load('errors_write')
287603Szelenkov@nginx.com
288603Szelenkov@nginx.com        self.get()
289603Szelenkov@nginx.com
290603Szelenkov@nginx.com        self.stop()
291603Szelenkov@nginx.com
292603Szelenkov@nginx.com        self.assertIsNotNone(
293603Szelenkov@nginx.com            self.search_in_log(r'\[error\].+Error in application\.'),
294603Szelenkov@nginx.com            'errors write')
295603Szelenkov@nginx.com
296603Szelenkov@nginx.com    def test_python_application_body_array(self):
297603Szelenkov@nginx.com        self.load('body_array')
298603Szelenkov@nginx.com
299603Szelenkov@nginx.com        self.assertEqual(self.get()['body'], '0123456789', 'body array')
300603Szelenkov@nginx.com
301603Szelenkov@nginx.com    def test_python_application_body_io(self):
302603Szelenkov@nginx.com        self.load('body_io')
303603Szelenkov@nginx.com
304603Szelenkov@nginx.com        self.assertEqual(self.get()['body'], '0123456789', 'body io')
305603Szelenkov@nginx.com
306603Szelenkov@nginx.com    def test_python_application_body_io_file(self):
307603Szelenkov@nginx.com        self.load('body_io_file')
308603Szelenkov@nginx.com
309603Szelenkov@nginx.com        self.assertEqual(self.get()['body'], 'body\n', 'body io file')
310603Szelenkov@nginx.com
311603Szelenkov@nginx.com    @unittest.expectedFailure
312603Szelenkov@nginx.com    def test_python_application_syntax_error(self):
313603Szelenkov@nginx.com        self.skip_alerts.append(r'Python failed to import module "wsgi"')
314603Szelenkov@nginx.com        self.load('syntax_error')
315603Szelenkov@nginx.com
316603Szelenkov@nginx.com        self.assertEqual(self.get()['status'], 500, 'syntax error')
317603Szelenkov@nginx.com
318603Szelenkov@nginx.com    def test_python_application_close(self):
319603Szelenkov@nginx.com        self.load('close')
320603Szelenkov@nginx.com
321603Szelenkov@nginx.com        self.get()
322603Szelenkov@nginx.com
323603Szelenkov@nginx.com        self.stop()
324603Szelenkov@nginx.com
325603Szelenkov@nginx.com        self.assertIsNotNone(self.search_in_log(r'Close called\.'), 'close')
326603Szelenkov@nginx.com
327603Szelenkov@nginx.com    def test_python_application_close_error(self):
328603Szelenkov@nginx.com        self.load('close_error')
329603Szelenkov@nginx.com
330603Szelenkov@nginx.com        self.get()
331603Szelenkov@nginx.com
332603Szelenkov@nginx.com        self.stop()
333603Szelenkov@nginx.com
334603Szelenkov@nginx.com        self.assertIsNotNone(self.search_in_log(r'Close called\.'),
335603Szelenkov@nginx.com            'close error')
336603Szelenkov@nginx.com
337617Szelenkov@nginx.com    def test_python_application_not_iterable(self):
338617Szelenkov@nginx.com        self.load('not_iterable')
339617Szelenkov@nginx.com
340665Szelenkov@nginx.com        self.get()
341617Szelenkov@nginx.com
342617Szelenkov@nginx.com        self.stop()
343617Szelenkov@nginx.com
344617Szelenkov@nginx.com        self.assertIsNotNone(self.search_in_log(
345617Szelenkov@nginx.com            r'\[error\].+the application returned not an iterable object'),
346617Szelenkov@nginx.com            'not iterable')
347617Szelenkov@nginx.com
348664Szelenkov@nginx.com    def test_python_application_write(self):
349664Szelenkov@nginx.com        self.load('write')
350664Szelenkov@nginx.com
351664Szelenkov@nginx.com        self.assertEqual(self.get()['body'], '0123456789', 'write')
352664Szelenkov@nginx.com
353484Szelenkov@nginx.comif __name__ == '__main__':
354484Szelenkov@nginx.com    unittest.main()
355