xref: /unit/test/test_settings.py (revision 853:a8359dce963e)
1import time
2import socket
3import unittest
4import unit
5
6class TestUnitSettings(unit.TestUnitApplicationPython):
7
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(b"""GET / HTTP/1.1
17""", start=True, raw=True)
18
19        time.sleep(3)
20
21        resp = self.http(b"""Host: localhost
22Connection: close
23
24""", sock=sock, raw=True)
25
26        self.assertEqual(resp['status'], 408, 'status header read timeout')
27
28    def test_settings_header_read_timeout_update(self):
29        self.load('empty')
30
31        r = None
32
33        self.conf({'http': { 'header_read_timeout': 4 }}, 'settings')
34
35        (resp, sock) = self.http(b"""GET / HTTP/1.1
36""", start=True, raw=True, no_recv=True)
37
38        time.sleep(2)
39
40        (resp, sock) = self.http(b"""Host: localhost
41""", start=True, sock=sock, raw=True, no_recv=True)
42
43        time.sleep(2)
44
45        (resp, sock) = self.http(b"""X-Blah: blah
46""", start=True, sock=sock, raw=True)
47
48        if len(resp) != 0:
49            sock.close()
50
51        else:
52            time.sleep(2)
53
54            resp = self.http(b"""Connection: close
55
56""", sock=sock, raw=True)
57
58        self.assertEqual(resp['status'], 408,
59            'status header read timeout update')
60
61    def test_settings_body_read_timeout(self):
62        self.load('empty')
63
64        self.conf({'http': { 'body_read_timeout': 2 }}, 'settings')
65
66        (resp, sock) = self.http(b"""POST / HTTP/1.1
67Host: localhost
68Content-Length: 10
69Connection: close
70
71""", start=True, raw_resp=True, raw=True)
72
73        time.sleep(3)
74
75        resp = self.http(b"""0123456789""", sock=sock, raw=True)
76
77        self.assertEqual(resp['status'], 408, 'status body read timeout')
78
79    def test_settings_body_read_timeout_update(self):
80        self.load('empty')
81
82        self.conf({'http': { 'body_read_timeout': 4 }}, 'settings')
83
84        (resp, sock) = self.http(b"""POST / HTTP/1.1
85Host: localhost
86Content-Length: 10
87Connection: close
88
89""", start=True, raw=True)
90
91        time.sleep(2)
92
93        (resp, sock) = self.http(b"""012""", start=True, sock=sock, raw=True)
94
95        time.sleep(2)
96
97        (resp, sock) = self.http(b"""345""", start=True, sock=sock, raw=True)
98
99        time.sleep(2)
100
101        resp = self.http(b"""6789""", sock=sock, raw=True)
102
103        self.assertEqual(resp['status'], 200, 'status body read timeout update')
104
105    def test_settings_send_timeout(self):
106        self.load('mirror')
107
108        data_len = 1048576
109
110        self.conf({'http': { 'send_timeout': 1 }}, 'settings')
111
112        addr = self.testdir + '/sock'
113
114        self.conf({"unix:" + addr: {'application': 'mirror'}}, 'listeners')
115
116        sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
117        sock.connect(addr)
118
119        req = """POST / HTTP/1.1
120Host: localhost
121Content-Type: text/html
122Content-Length: %d
123
124""" % data_len + ('X' * data_len)
125
126        sock.sendall(req.encode())
127
128        data = sock.recv(16).decode()
129
130        time.sleep(3)
131
132        data += self.recvall(sock).decode()
133
134        sock.close()
135
136        self.assertRegex(data, r'200 OK', 'status send timeout')
137        self.assertLess(len(data), data_len, 'data send timeout')
138
139    def test_settings_idle_timeout(self):
140        self.load('empty')
141
142        self.conf({'http': { 'idle_timeout': 2 }}, 'settings')
143
144        (resp, sock) = self.get(headers={
145            'Connection': 'keep-alive',
146            'Host': 'localhost'
147        }, start=True)
148
149        time.sleep(3)
150
151        resp = self.get(headers={
152            'Connection': 'close',
153            'Host': 'localhost'
154        }, sock=sock)
155
156        self.assertEqual(resp['status'], 408, 'status idle timeout')
157
158    def test_settings_max_body_size(self):
159        self.load('empty')
160
161        self.conf({'http': { 'max_body_size': 5 }}, 'settings')
162
163        self.assertEqual(self.post(body='01234')['status'], 200, 'status size')
164        self.assertEqual(self.post(body='012345')['status'], 413,
165            'status size max')
166
167    @unittest.expectedFailure
168    def test_settings_negative_value(self):
169        self.assertIn('error', self.conf({'http': { 'max_body_size': -1 }},
170            'settings'), 'settings negative value')
171
172if __name__ == '__main__':
173    TestUnitSettings.main()
174