1import time 2import socket 3import unittest 4from unit.applications.lang.python import TestApplicationPython 5 6 7class TestSettings(TestApplicationPython): 8 prerequisites = ['python'] 9 10 def test_settings_header_read_timeout(self): 11 self.load('empty') 12 13 self.conf({'http': {'header_read_timeout': 2}}, 'settings') 14 15 (resp, sock) = self.http( 16 b"""GET / HTTP/1.1 17""", 18 start=True, 19 read_timeout=1, 20 raw=True, 21 ) 22 23 time.sleep(3) 24 25 resp = self.http( 26 b"""Host: localhost 27Connection: close 28 29""", 30 sock=sock, 31 raw=True, 32 ) 33 34 self.assertEqual(resp['status'], 408, 'status header read timeout') 35 36 def test_settings_header_read_timeout_update(self): 37 self.load('empty') 38 39 self.conf({'http': {'header_read_timeout': 4}}, 'settings') 40 41 (resp, sock) = self.http( 42 b"""GET / HTTP/1.1 43""", 44 start=True, 45 raw=True, 46 no_recv=True, 47 ) 48 49 time.sleep(2) 50 51 (resp, sock) = self.http( 52 b"""Host: localhost 53""", 54 start=True, 55 sock=sock, 56 raw=True, 57 no_recv=True, 58 ) 59 60 time.sleep(2) 61 62 (resp, sock) = self.http( 63 b"""X-Blah: blah 64""", 65 start=True, 66 sock=sock, 67 read_timeout=1, 68 raw=True, 69 ) 70 71 if len(resp) != 0: 72 sock.close() 73 74 else: 75 time.sleep(2) 76 77 resp = self.http( 78 b"""Connection: close 79 80""", 81 sock=sock, 82 raw=True, 83 ) 84 85 self.assertEqual( 86 resp['status'], 408, 'status header read timeout update' 87 ) 88 89 def test_settings_body_read_timeout(self): 90 self.load('empty') 91 92 self.conf({'http': {'body_read_timeout': 2}}, 'settings') 93 94 (resp, sock) = self.http( 95 b"""POST / HTTP/1.1 96Host: localhost 97Content-Length: 10 98Connection: close 99 100""", 101 start=True, 102 raw_resp=True, 103 read_timeout=1, 104 raw=True, 105 ) 106 107 time.sleep(3) 108 109 resp = self.http(b"""0123456789""", sock=sock, raw=True) 110 111 self.assertEqual(resp['status'], 408, 'status body read timeout') 112 113 def test_settings_body_read_timeout_update(self): 114 self.load('empty') 115 116 self.conf({'http': {'body_read_timeout': 4}}, 'settings') 117 118 (resp, sock) = self.http( 119 b"""POST / HTTP/1.1 120Host: localhost 121Content-Length: 10 122Connection: close 123 124""", 125 start=True, 126 read_timeout=1, 127 raw=True, 128 ) 129 130 time.sleep(2) 131 132 (resp, sock) = self.http( 133 b"""012""", start=True, sock=sock, read_timeout=1, raw=True 134 ) 135 136 time.sleep(2) 137 138 (resp, sock) = self.http( 139 b"""345""", start=True, sock=sock, read_timeout=1, raw=True 140 ) 141 142 time.sleep(2) 143 144 resp = self.http(b"""6789""", sock=sock, raw=True) 145 146 self.assertEqual( 147 resp['status'], 200, 'status body read timeout update' 148 ) 149 150 def test_settings_send_timeout(self): 151 self.load('mirror') 152 153 data_len = 1048576 154 155 self.conf({'http': {'send_timeout': 1}}, 'settings') 156 157 addr = self.testdir + '/sock' 158 159 self.conf({"unix:" + addr: {'application': 'mirror'}}, 'listeners') 160 161 sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) 162 sock.connect(addr) 163 164 req = """POST / HTTP/1.1 165Host: localhost 166Content-Type: text/html 167Content-Length: %d 168Connection: close 169 170""" % data_len + ( 171 'X' * data_len 172 ) 173 174 sock.sendall(req.encode()) 175 176 data = sock.recv(16).decode() 177 178 time.sleep(3) 179 180 data += self.recvall(sock).decode() 181 182 sock.close() 183 184 self.assertRegex(data, r'200 OK', 'status send timeout') 185 self.assertLess(len(data), data_len, 'data send timeout') 186 187 def test_settings_idle_timeout(self): 188 self.load('empty') 189 190 self.assertEqual(self.get()['status'], 200, 'init') 191 192 self.conf({'http': {'idle_timeout': 2}}, 'settings') 193 194 (resp, sock) = self.get( 195 headers={'Host': 'localhost', 'Connection': 'keep-alive'}, 196 start=True, 197 read_timeout=1, 198 ) 199 200 time.sleep(3) 201 202 resp = self.get( 203 headers={'Host': 'localhost', 'Connection': 'close'}, sock=sock 204 ) 205 206 self.assertEqual(resp['status'], 408, 'status idle timeout') 207 208 def test_settings_max_body_size(self): 209 self.load('empty') 210 211 self.conf({'http': {'max_body_size': 5}}, 'settings') 212 213 self.assertEqual(self.post(body='01234')['status'], 200, 'status size') 214 self.assertEqual( 215 self.post(body='012345')['status'], 413, 'status size max' 216 ) 217 218 @unittest.expectedFailure 219 def test_settings_negative_value(self): 220 self.assertIn( 221 'error', 222 self.conf({'http': {'max_body_size': -1}}, 'settings'), 223 'settings negative value', 224 ) 225 226 227if __name__ == '__main__': 228 TestSettings.main() 229