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