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