xref: /unit/test/test_python_application.py (revision 603:a11b80dedc6b)
1import unittest
2import unit
3
4class TestUnitPythonApplication(unit.TestUnitApplicationPython):
5
6    def setUpClass():
7        unit.TestUnit().check_modules('python')
8
9    def test_python_application_variables(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        }, body=body)
19
20        self.assertEqual(resp['status'], 200, 'status')
21        headers = resp['headers']
22        self.assertRegex(headers.pop('Server'), r'Unit/[\d\.]+',
23            'server header')
24
25        date = headers.pop('Date')
26        self.assertEqual(date[-4:], ' GMT', 'date header timezone')
27        self.assertLess(abs(self.date_to_sec_epoch(date) - self.sec_epoch()), 5,
28            'date header')
29
30        self.assertDictEqual(headers, {
31            'Content-Length': str(len(body)),
32            'Content-Type': 'text/html',
33            'Request-Method': 'POST',
34            'Request-Uri': '/',
35            'Http-Host': 'localhost',
36            'Server-Protocol': 'HTTP/1.1',
37            'Custom-Header': 'blah',
38            'Wsgi-Version': '(1, 0)',
39            'Wsgi-Url-Scheme': 'http',
40            'Wsgi-Multithread': 'False',
41            'Wsgi-Multiprocess': 'True',
42            'Wsgi-Run-Once': 'False'
43        }, 'headers')
44        self.assertEqual(resp['body'], body, 'body')
45
46    def test_python_application_query_string(self):
47        self.load('query_string')
48
49        resp = self.get(url='/?var1=val1&var2=val2')
50
51        self.assertEqual(resp['headers']['Query-String'], 'var1=val1&var2=val2',
52            'Query-String header')
53
54    @unittest.expectedFailure
55    def test_python_application_server_port(self):
56        self.load('server_port')
57
58        self.assertEqual(self.get()['headers']['Server-Port'], '7080',
59            'Server-Port header')
60
61    @unittest.expectedFailure
62    def test_python_application_204_transfer_encoding(self):
63        self.load('204_no_content')
64
65        self.assertNotIn('Transfer-Encoding', self.get()['headers'],
66            '204 header transfer encoding')
67
68    def test_python_application_ctx_iter_atexit(self):
69        self.skip_alerts.append(r'sendmsg.+failed')
70        self.load('ctx_iter_atexit')
71
72        resp = self.post(headers={
73            'Connection': 'close',
74            'Content-Type': 'text/html',
75            'Host': 'localhost'
76        }, body='0123456789')
77
78        self.assertEqual(resp['status'], 200, 'ctx iter status')
79        self.assertEqual(resp['body'], '0123456789', 'ctx iter body')
80
81        self.conf({
82            "listeners": {},
83            "applications": {}
84        })
85
86        self.stop()
87
88        self.assertIsNotNone(self.search_in_log(r'RuntimeError'),
89            'ctx iter atexit')
90
91    def test_python_keepalive_body(self):
92        self.load('mirror')
93
94        (resp, sock) = self.post(headers={
95            'Connection': 'keep-alive',
96            'Content-Type': 'text/html',
97            'Host': 'localhost'
98        }, start=True, body='0123456789' * 500)
99
100        self.assertEqual(resp['body'], '0123456789' * 500, 'keep-alive 1')
101
102        resp = self.post(headers={
103            'Connection': 'close',
104            'Content-Type': 'text/html',
105            'Host': 'localhost'
106        }, sock=sock, body='0123456789')
107
108        self.assertEqual(resp['body'], '0123456789', 'keep-alive 2')
109
110    def test_python_atexit(self):
111        self.skip_alerts.append(r'sendmsg.+failed')
112        self.load('atexit')
113
114        self.get()
115
116        self.conf({
117            "listeners": {},
118            "applications": {}
119        })
120
121        self.stop()
122
123        self.assertIsNotNone(self.search_in_log(r'At exit called\.'), 'atexit')
124
125    @unittest.expectedFailure
126    def test_python_application_start_response_exit(self):
127        self.load('start_response_exit')
128
129        self.assertEqual(self.get()['status'], 500, 'start response exit')
130
131    @unittest.expectedFailure
132    def test_python_application_input_iter(self):
133        self.load('input_iter')
134
135        body = '0123456789'
136
137        self.assertEqual(self.post(body=body)['body'], body, 'input iter')
138
139    @unittest.expectedFailure
140    def test_python_application_input_read_length(self):
141        self.load('input_read_length')
142
143        body = '0123456789'
144
145        resp = self.post(headers={
146            'Host': 'localhost',
147            'Input-Length': '5',
148            'Connection': 'close'
149        }, body=body)
150
151        self.assertEqual(resp['body'], body[:5], 'input read length lt body')
152
153        resp = self.post(headers={
154            'Host': 'localhost',
155            'Input-Length': '15',
156            'Connection': 'close'
157        }, body=body)
158
159        self.assertEqual(resp['body'], body, 'input read length gt body')
160
161        resp = self.post(headers={
162            'Host': 'localhost',
163            'Input-Length': '0',
164            'Connection': 'close'
165        }, body=body)
166
167        self.assertEqual(resp['body'], '', 'input read length zero')
168
169        resp = self.post(headers={
170            'Host': 'localhost',
171            'Input-Length': '-1',
172            'Connection': 'close'
173        }, body=body)
174
175        self.assertEqual(resp['body'], body, 'input read length negative')
176
177    @unittest.expectedFailure
178    def test_python_application_errors_write(self):
179        self.load('errors_write')
180
181        self.get()
182
183        self.stop()
184
185        self.assertIsNotNone(
186            self.search_in_log(r'\[error\].+Error in application\.'),
187            'errors write')
188
189    def test_python_application_body_array(self):
190        self.load('body_array')
191
192        self.assertEqual(self.get()['body'], '0123456789', 'body array')
193
194    def test_python_application_body_io(self):
195        self.load('body_io')
196
197        self.assertEqual(self.get()['body'], '0123456789', 'body io')
198
199    def test_python_application_body_io_file(self):
200        self.load('body_io_file')
201
202        self.assertEqual(self.get()['body'], 'body\n', 'body io file')
203
204    @unittest.expectedFailure
205    def test_python_application_syntax_error(self):
206        self.skip_alerts.append(r'Python failed to import module "wsgi"')
207        self.load('syntax_error')
208
209        self.assertEqual(self.get()['status'], 500, 'syntax error')
210
211    def test_python_application_close(self):
212        self.load('close')
213
214        self.get()
215
216        self.stop()
217
218        self.assertIsNotNone(self.search_in_log(r'Close called\.'), 'close')
219
220    def test_python_application_close_error(self):
221        self.load('close_error')
222
223        self.get()
224
225        self.stop()
226
227        self.assertIsNotNone(self.search_in_log(r'Close called\.'),
228            'close error')
229
230if __name__ == '__main__':
231    unittest.main()
232