import time import socket import unittest import unit class TestUnitSettings(unit.TestUnitApplicationPython): def setUpClass(): unit.TestUnit().check_modules('python') def test_settings_header_read_timeout(self): self.load('empty') self.conf({'http': { 'header_read_timeout': 2 }}, '/settings') (resp, sock) = self.http(b"""GET / HTTP/1.1 """, start=True, raw=True) time.sleep(3) resp = self.http(b"""Host: localhost Connection: close """, sock=sock, raw=True) self.assertEqual(resp['status'], 408, 'status header read timeout') def test_settings_body_read_timeout(self): self.load('empty') self.conf({'http': { 'body_read_timeout': 2 }}, '/settings') (resp, sock) = self.http(b"""POST / HTTP/1.1 Host: localhost Content-Length: 10 Connection: close """, start=True, raw_resp=True, raw=True) time.sleep(3) resp = self.http(b"""0123456789""", sock=sock, raw=True) self.assertEqual(resp['status'], 408, 'status body read timeout') def test_settings_send_timeout(self): self.load('mirror') data_len = 1048576 self.conf({'http': { 'send_timeout': 1 }}, '/settings') addr = self.testdir + '/sock' self.conf({"unix:" + addr: {'application': 'mirror'}}, '/listeners') sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) sock.connect(addr) req = """POST / HTTP/1.1 Host: localhost Content-Type: text/html Content-Length: %d """ % data_len + ('X' * data_len) sock.sendall(req.encode()) data = sock.recv(16).decode() time.sleep(3) data += self.recvall(sock) sock.close() self.assertRegex(data, r'200 OK', 'status send timeout') self.assertLess(len(data), data_len, 'data send timeout') def test_settings_idle_timeout(self): self.load('empty') self.conf({'http': { 'idle_timeout': 2 }}, '/settings') (resp, sock) = self.get(headers={ 'Connection': 'keep-alive', 'Host': 'localhost' }, start=True) time.sleep(3) resp = self.get(headers={ 'Connection': 'close', 'Host': 'localhost' }, sock=sock) self.assertEqual(resp, {}, 'idle timeout') def test_settings_max_body_size(self): self.load('empty') self.conf({'http': { 'max_body_size': 5 }}, '/settings') self.assertEqual(self.post(body='01234')['status'], 200, 'status size') self.assertEqual(self.post(body='012345')['status'], 413, 'status size max') @unittest.expectedFailure def test_settings_negative_value(self): self.assertIn('error', self.conf({'http': { 'max_body_size': -1 }}, '/settings'), 'settings negative value') if __name__ == '__main__': unittest.main()