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