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