xref: /unit/test/test_settings.py (revision 1477:b93d1acf81bd) 
1import socket
2import time
3import unittest
4
5from unit.applications.lang.python import TestApplicationPython
6
7
8class TestSettings(TestApplicationPython):
9    prerequisites = {'modules': {'python': 'any'}}
10
11    def test_settings_header_read_timeout(self):
12        self.load('empty')
13
14        self.conf({'http': {'header_read_timeout': 2}}, 'settings')
15
16        (resp, sock) = self.http(
17            b"""GET / HTTP/1.1
18""",
19            start=True,
20            read_timeout=1,
21            raw=True,
22        )
23
24        time.sleep(3)
25
26        resp = self.http(
27            b"""Host: localhost
28Connection: close
29
30""",
31            sock=sock,
32            raw=True,
33        )
34
35        self.assertEqual(resp['status'], 408, 'status header read timeout')
36
37    def test_settings_header_read_timeout_update(self):
38        self.load('empty')
39
40        self.conf({'http': {'header_read_timeout': 4}}, 'settings')
41
42        (resp, sock) = self.http(
43            b"""GET / HTTP/1.1
44""",
45            start=True,
46            raw=True,
47            no_recv=True,
48        )
49
50        time.sleep(2)
51
52        (resp, sock) = self.http(
53            b"""Host: localhost
54""",
55            start=True,
56            sock=sock,
57            raw=True,
58            no_recv=True,
59        )
60
61        time.sleep(2)
62
63        (resp, sock) = self.http(
64            b"""X-Blah: blah
65""",
66            start=True,
67            sock=sock,
68            read_timeout=1,
69            raw=True,
70        )
71
72        if len(resp) != 0:
73            sock.close()
74
75        else:
76            time.sleep(2)
77
78            resp = self.http(
79                b"""Connection: close
80
81""",
82                sock=sock,
83                raw=True,
84            )
85
86        self.assertEqual(
87            resp['status'], 408, 'status header read timeout update'
88        )
89
90    def test_settings_body_read_timeout(self):
91        self.load('empty')
92
93        self.conf({'http': {'body_read_timeout': 2}}, 'settings')
94
95        (resp, sock) = self.http(
96            b"""POST / HTTP/1.1
97Host: localhost
98Content-Length: 10
99Connection: close
100
101""",
102            start=True,
103            raw_resp=True,
104            read_timeout=1,
105            raw=True,
106        )
107
108        time.sleep(3)
109
110        resp = self.http(b"""0123456789""", sock=sock, raw=True)
111
112        self.assertEqual(resp['status'], 408, 'status body read timeout')
113
114    def test_settings_body_read_timeout_update(self):
115        self.load('empty')
116
117        self.conf({'http': {'body_read_timeout': 4}}, 'settings')
118
119        (resp, sock) = self.http(
120            b"""POST / HTTP/1.1
121Host: localhost
122Content-Length: 10
123Connection: close
124
125""",
126            start=True,
127            read_timeout=1,
128            raw=True,
129        )
130
131        time.sleep(2)
132
133        (resp, sock) = self.http(
134            b"""012""", start=True, sock=sock, read_timeout=1, raw=True
135        )
136
137        time.sleep(2)
138
139        (resp, sock) = self.http(
140            b"""345""", start=True, sock=sock, read_timeout=1, raw=True
141        )
142
143        time.sleep(2)
144
145        resp = self.http(b"""6789""", sock=sock, raw=True)
146
147        self.assertEqual(
148            resp['status'], 200, 'status body read timeout update'
149        )
150
151    def test_settings_send_timeout(self):
152        self.load('mirror')
153
154        data_len = 1048576
155
156        self.conf({'http': {'send_timeout': 1}}, 'settings')
157
158        addr = self.testdir + '/sock'
159
160        self.conf({"unix:" + addr: {'application': 'mirror'}}, 'listeners')
161
162        sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
163        sock.connect(addr)
164
165        req = """POST / HTTP/1.1
166Host: localhost
167Content-Type: text/html
168Content-Length: %d
169Connection: close
170
171""" % data_len + (
172            'X' * data_len
173        )
174
175        sock.sendall(req.encode())
176
177        data = sock.recv(16).decode()
178
179        time.sleep(3)
180
181        data += self.recvall(sock).decode()
182
183        sock.close()
184
185        self.assertRegex(data, r'200 OK', 'status send timeout')
186        self.assertLess(len(data), data_len, 'data send timeout')
187
188    def test_settings_idle_timeout(self):
189        self.load('empty')
190
191        self.assertEqual(self.get()['status'], 200, 'init')
192
193        self.conf({'http': {'idle_timeout': 2}}, 'settings')
194
195        (resp, sock) = self.get(
196            headers={'Host': 'localhost', 'Connection': 'keep-alive'},
197            start=True,
198            read_timeout=1,
199        )
200
201        time.sleep(3)
202
203        resp = self.get(
204            headers={'Host': 'localhost', 'Connection': 'close'}, sock=sock
205        )
206
207        self.assertEqual(resp['status'], 408, 'status idle timeout')
208
209    def test_settings_max_body_size(self):
210        self.load('empty')
211
212        self.conf({'http': {'max_body_size': 5}}, 'settings')
213
214        self.assertEqual(self.post(body='01234')['status'], 200, 'status size')
215        self.assertEqual(
216            self.post(body='012345')['status'], 413, 'status size max'
217        )
218
219    def test_settings_max_body_size_large(self):
220        self.load('mirror')
221
222        self.conf({'http': {'max_body_size': 32 * 1024 * 1024}}, 'settings')
223
224        body = '0123456789abcdef' * 4 * 64 * 1024
225        resp = self.post(body=body, read_buffer_size=1024 * 1024)
226        self.assertEqual(resp['status'], 200, 'status size 4')
227        self.assertEqual(resp['body'], body, 'status body 4')
228
229        body = '0123456789abcdef' * 8 * 64 * 1024
230        resp = self.post(body=body, read_buffer_size=1024 * 1024)
231        self.assertEqual(resp['status'], 200, 'status size 8')
232        self.assertEqual(resp['body'], body, 'status body 8')
233
234        body = '0123456789abcdef' * 16 * 64 * 1024
235        resp = self.post(body=body, read_buffer_size=1024 * 1024)
236        self.assertEqual(resp['status'], 200, 'status size 16')
237        self.assertEqual(resp['body'], body, 'status body 16')
238
239        body = '0123456789abcdef' * 32 * 64 * 1024
240        resp = self.post(body=body, read_buffer_size=1024 * 1024)
241        self.assertEqual(resp['status'], 200, 'status size 32')
242        self.assertEqual(resp['body'], body, 'status body 32')
243
244    @unittest.skip('not yet')
245    def test_settings_negative_value(self):
246        self.assertIn(
247            'error',
248            self.conf({'http': {'max_body_size': -1}}, 'settings'),
249            'settings negative value',
250        )
251
252
253if __name__ == '__main__':
254    TestSettings.main()
255