xref: /unit/test/test_settings.py (revision 971:7098b3ba25fd) 
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, read_timeout=1, 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, read_timeout=1, raw=True, no_recv=True)
35
36        time.sleep(2)
37
38        (resp, sock) = self.http(b"""Host: localhost
39""", start=True, sock=sock, read_timeout=1, 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, read_timeout=1, 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, read_timeout=1, 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, read_timeout=1, raw=True)
88
89        time.sleep(2)
90
91        (resp, sock) = self.http(b"""012""", start=True, sock=sock,
92            read_timeout=1, raw=True)
93
94        time.sleep(2)
95
96        (resp, sock) = self.http(b"""345""", start=True, sock=sock,
97            read_timeout=1, 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
123Connection: close
124
125""" % data_len + ('X' * data_len)
126
127        sock.sendall(req.encode())
128
129        data = sock.recv(16).decode()
130
131        time.sleep(3)
132
133        data += self.recvall(sock).decode()
134
135        sock.close()
136
137        self.assertRegex(data, r'200 OK', 'status send timeout')
138        self.assertLess(len(data), data_len, 'data send timeout')
139
140    def test_settings_idle_timeout(self):
141        self.load('empty')
142
143        self.conf({'http': { 'idle_timeout': 2 }}, 'settings')
144
145        (resp, sock) = self.get(headers={
146            'Host': 'localhost',
147            'Connection': 'keep-alive'
148        }, start=True, read_timeout=1)
149
150        time.sleep(3)
151
152        resp = self.get(headers={
153            'Host': 'localhost',
154            'Connection': 'close'
155        }, sock=sock)
156
157        self.assertEqual(resp['status'], 408, 'status idle timeout')
158
159    def test_settings_max_body_size(self):
160        self.load('empty')
161
162        self.conf({'http': { 'max_body_size': 5 }}, 'settings')
163
164        self.assertEqual(self.post(body='01234')['status'], 200, 'status size')
165        self.assertEqual(self.post(body='012345')['status'], 413,
166            'status size max')
167
168    @unittest.expectedFailure
169    def test_settings_negative_value(self):
170        self.assertIn('error', self.conf({'http': { 'max_body_size': -1 }},
171            'settings'), 'settings negative value')
172
173if __name__ == '__main__':
174    TestUnitSettings.main()
175