xref: /unit/test/test_settings.py (revision 1654:fc7d0578e124)
1import re
2import socket
3import time
4
5import pytest
6
7from unit.applications.lang.python import TestApplicationPython
8
9
10class TestSettings(TestApplicationPython):
11    prerequisites = {'modules': {'python': 'any'}}
12
13    def test_settings_header_read_timeout(self):
14        self.load('empty')
15
16        self.conf({'http': {'header_read_timeout': 2}}, 'settings')
17
18        (resp, sock) = self.http(
19            b"""GET / HTTP/1.1
20""",
21            start=True,
22            read_timeout=1,
23            raw=True,
24        )
25
26        time.sleep(3)
27
28        resp = self.http(
29            b"""Host: localhost
30Connection: close
31
32""",
33            sock=sock,
34            raw=True,
35        )
36
37        assert resp['status'] == 408, 'status header read timeout'
38
39    def test_settings_header_read_timeout_update(self):
40        self.load('empty')
41
42        self.conf({'http': {'header_read_timeout': 4}}, 'settings')
43
44        (resp, sock) = self.http(
45            b"""GET / HTTP/1.1
46""",
47            start=True,
48            raw=True,
49            no_recv=True,
50        )
51
52        time.sleep(2)
53
54        (resp, sock) = self.http(
55            b"""Host: localhost
56""",
57            start=True,
58            sock=sock,
59            raw=True,
60            no_recv=True,
61        )
62
63        time.sleep(2)
64
65        (resp, sock) = self.http(
66            b"""X-Blah: blah
67""",
68            start=True,
69            sock=sock,
70            read_timeout=1,
71            raw=True,
72        )
73
74        if len(resp) != 0:
75            sock.close()
76
77        else:
78            time.sleep(2)
79
80            resp = self.http(
81                b"""Connection: close
82
83""",
84                sock=sock,
85                raw=True,
86            )
87
88        assert resp['status'] == 408, 'status header read timeout update'
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        assert 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        assert resp['status'] == 200, 'status body read timeout update'
148
149    def test_settings_send_timeout(self, temp_dir):
150        self.load('mirror')
151
152        data_len = 1048576
153
154        self.conf({'http': {'send_timeout': 1}}, 'settings')
155
156        addr = temp_dir + '/sock'
157
158        self.conf({"unix:" + addr: {'application': 'mirror'}}, 'listeners')
159
160        sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
161        sock.connect(addr)
162
163        req = """POST / HTTP/1.1
164Host: localhost
165Content-Type: text/html
166Content-Length: %d
167Connection: close
168
169""" % data_len + (
170            'X' * data_len
171        )
172
173        sock.sendall(req.encode())
174
175        data = sock.recv(16).decode()
176
177        time.sleep(3)
178
179        data += self.recvall(sock).decode()
180
181        sock.close()
182
183        assert re.search(r'200 OK', data), 'status send timeout'
184        assert len(data) < data_len, 'data send timeout'
185
186    def test_settings_idle_timeout(self):
187        self.load('empty')
188
189        assert self.get()['status'] == 200, 'init'
190
191        self.conf({'http': {'idle_timeout': 2}}, 'settings')
192
193        (resp, sock) = self.get(
194            headers={'Host': 'localhost', 'Connection': 'keep-alive'},
195            start=True,
196            read_timeout=1,
197        )
198
199        time.sleep(3)
200
201        resp = self.get(
202            headers={'Host': 'localhost', 'Connection': 'close'}, sock=sock
203        )
204
205        assert resp['status'] == 408, 'status idle timeout'
206
207    def test_settings_idle_timeout_2(self):
208        self.load('empty')
209
210        assert self.get()['status'] == 200, 'init'
211
212        self.conf({'http': {'idle_timeout': 1}}, 'settings')
213
214        _, sock = self.http(b'', start=True, raw=True, no_recv=True)
215
216        time.sleep(3)
217
218        assert (
219            self.get(
220                headers={'Host': 'localhost', 'Connection': 'close'}, sock=sock
221            )['status']
222            == 408
223        ), 'status idle timeout'
224
225    def test_settings_max_body_size(self):
226        self.load('empty')
227
228        self.conf({'http': {'max_body_size': 5}}, 'settings')
229
230        assert self.post(body='01234')['status'] == 200, 'status size'
231        assert self.post(body='012345')['status'] == 413, 'status size max'
232
233    def test_settings_max_body_size_large(self):
234        self.load('mirror')
235
236        self.conf({'http': {'max_body_size': 32 * 1024 * 1024}}, 'settings')
237
238        body = '0123456789abcdef' * 4 * 64 * 1024
239        resp = self.post(body=body, read_buffer_size=1024 * 1024)
240        assert resp['status'] == 200, 'status size 4'
241        assert resp['body'] == body, 'status body 4'
242
243        body = '0123456789abcdef' * 8 * 64 * 1024
244        resp = self.post(body=body, read_buffer_size=1024 * 1024)
245        assert resp['status'] == 200, 'status size 8'
246        assert resp['body'] == body, 'status body 8'
247
248        body = '0123456789abcdef' * 16 * 64 * 1024
249        resp = self.post(body=body, read_buffer_size=1024 * 1024)
250        assert resp['status'] == 200, 'status size 16'
251        assert resp['body'] == body, 'status body 16'
252
253        body = '0123456789abcdef' * 32 * 64 * 1024
254        resp = self.post(body=body, read_buffer_size=1024 * 1024)
255        assert resp['status'] == 200, 'status size 32'
256        assert resp['body'] == body, 'status body 32'
257
258    @pytest.mark.skip('not yet')
259    def test_settings_negative_value(self):
260        assert 'error' in self.conf(
261            {'http': {'max_body_size': -1}}, 'settings'
262        ), 'settings negative value'
263