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