xref: /unit/test/test_settings.py (revision 1025:3469f5185562) 
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            read_timeout=1,
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            read_timeout=1,
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        self.assertEqual(
88            resp['status'], 408, 'status header read timeout update'
89        )
90
91    def test_settings_body_read_timeout(self):
92        self.load('empty')
93
94        self.conf({'http': {'body_read_timeout': 2}}, 'settings')
95
96        (resp, sock) = self.http(
97            b"""POST / HTTP/1.1
98Host: localhost
99Content-Length: 10
100Connection: close
101
102""",
103            start=True,
104            raw_resp=True,
105            read_timeout=1,
106            raw=True,
107        )
108
109        time.sleep(3)
110
111        resp = self.http(b"""0123456789""", sock=sock, raw=True)
112
113        self.assertEqual(resp['status'], 408, 'status body read timeout')
114
115    def test_settings_body_read_timeout_update(self):
116        self.load('empty')
117
118        self.conf({'http': {'body_read_timeout': 4}}, 'settings')
119
120        (resp, sock) = self.http(
121            b"""POST / HTTP/1.1
122Host: localhost
123Content-Length: 10
124Connection: close
125
126""",
127            start=True,
128            read_timeout=1,
129            raw=True,
130        )
131
132        time.sleep(2)
133
134        (resp, sock) = self.http(
135            b"""012""", start=True, sock=sock, read_timeout=1, raw=True
136        )
137
138        time.sleep(2)
139
140        (resp, sock) = self.http(
141            b"""345""", start=True, sock=sock, read_timeout=1, raw=True
142        )
143
144        time.sleep(2)
145
146        resp = self.http(b"""6789""", sock=sock, raw=True)
147
148        self.assertEqual(
149            resp['status'], 200, 'status body read timeout update'
150        )
151
152    def test_settings_send_timeout(self):
153        self.load('mirror')
154
155        data_len = 1048576
156
157        self.conf({'http': {'send_timeout': 1}}, 'settings')
158
159        addr = self.testdir + '/sock'
160
161        self.conf({"unix:" + addr: {'application': 'mirror'}}, 'listeners')
162
163        sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
164        sock.connect(addr)
165
166        req = """POST / HTTP/1.1
167Host: localhost
168Content-Type: text/html
169Content-Length: %d
170Connection: close
171
172""" % data_len + (
173            'X' * data_len
174        )
175
176        sock.sendall(req.encode())
177
178        data = sock.recv(16).decode()
179
180        time.sleep(3)
181
182        data += self.recvall(sock).decode()
183
184        sock.close()
185
186        self.assertRegex(data, r'200 OK', 'status send timeout')
187        self.assertLess(len(data), data_len, 'data send timeout')
188
189    def test_settings_idle_timeout(self):
190        self.load('empty')
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