import binascii import io import json import os import re import select import socket import pytest from conftest import option from unit.main import TestUnit class TestHTTP(TestUnit): def http(self, start_str, **kwargs): sock_type = kwargs.get('sock_type', 'ipv4') port = kwargs.get('port', 7080) url = kwargs.get('url', '/') http = 'HTTP/1.0' if 'http_10' in kwargs else 'HTTP/1.1' headers = kwargs.get('headers', {'Host': 'localhost', 'Connection': 'close'}) body = kwargs.get('body', b'') crlf = '\r\n' if 'addr' not in kwargs: addr = '::1' if sock_type == 'ipv6' else '127.0.0.1' else: addr = kwargs['addr'] sock_types = { 'ipv4': socket.AF_INET, 'ipv6': socket.AF_INET6, 'unix': socket.AF_UNIX, } if 'sock' not in kwargs: sock = socket.socket(sock_types[sock_type], socket.SOCK_STREAM) if ( sock_type == sock_types['ipv4'] or sock_type == sock_types['ipv6'] ): sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1) if 'wrapper' in kwargs: sock = kwargs['wrapper'](sock) connect_args = addr if sock_type == 'unix' else (addr, port) try: sock.connect(connect_args) except ConnectionRefusedError: sock.close() pytest.fail('Client can\'t connect to the server.') else: sock = kwargs['sock'] if 'raw' not in kwargs: req = ' '.join([start_str, url, http]) + crlf if body != b'': if isinstance(body, str): body = body.encode() elif isinstance(body, dict): body, content_type = self.form_encode(body) headers['Content-Type'] = content_type if 'Content-Length' not in headers: headers['Content-Length'] = len(body) for header, value in headers.items(): if isinstance(value, list): for v in value: req += header + ': ' + str(v) + crlf else: req += header + ': ' + str(value) + crlf req = (req + crlf).encode() + body else: req = start_str sock.sendall(req) encoding = kwargs.get('encoding', 'utf-8') self.log_out(req, encoding) resp = '' if 'no_recv' not in kwargs: recvall_kwargs = {} if 'read_timeout' in kwargs: recvall_kwargs['read_timeout'] = kwargs['read_timeout'] if 'read_buffer_size' in kwargs: recvall_kwargs['buff_size'] = kwargs['read_buffer_size'] resp = self.recvall(sock, **recvall_kwargs).decode(encoding) self.log_in(resp) if 'raw_resp' not in kwargs: resp = self._resp_to_dict(resp) headers = resp.get('headers') if headers and headers.get('Transfer-Encoding') == 'chunked': resp['body'] = self._parse_chunked_body(resp['body']).decode( encoding ) if 'json' in kwargs: resp = self._parse_json(resp) if 'start' not in kwargs: sock.close() return resp return (resp, sock) def log_out(self, log, encoding): if option.detailed: print('>>>') log = self.log_truncate(log) try: print(log.decode(encoding, 'ignore')) except UnicodeEncodeError: print(log) def log_in(self, log): if option.detailed: print('<<<') log = self.log_truncate(log) try: print(log) except UnicodeEncodeError: print(log.encode()) def log_truncate(self, log, limit=1024): len_log = len(log) if len_log > limit: log = log[:limit] appendix = '(...logged %s of %s bytes)' % (limit, len_log) if isinstance(log, bytes): appendix = appendix.encode() log = log + appendix return log def delete(self, **kwargs): return self.http('DELETE', **kwargs) def get(self, **kwargs): return self.http('GET', **kwargs) def head(self, **kwargs): return self.http('HEAD', **kwargs) def post(self, **kwargs): return self.http('POST', **kwargs) def put(self, **kwargs): return self.http('PUT', **kwargs) def recvall(self, sock, **kwargs): timeout_default = 60 timeout = kwargs.get('read_timeout', timeout_default) buff_size = kwargs.get('buff_size', 4096) data = b'' while True: rlist = select.select([sock], [], [], timeout)[0] if not rlist: # For all current cases if the "read_timeout" was changed # than test do not expect to get a response from server. if timeout == timeout_default: pytest.fail('Can\'t read response from server.') break try: part = sock.recv(buff_size) except: break data += part if not len(part): break return data def _resp_to_dict(self, resp): m = re.search(r'(.*?\x0d\x0a?)\x0d\x0a?(.*)', resp, re.M | re.S) if not m: return {} headers_text, body = m.group(1), m.group(2) p = re.compile('(.*?)\x0d\x0a?', re.M | re.S) headers_lines = p.findall(headers_text) status = re.search( r'^HTTP\/\d\.\d\s(\d+)|$', headers_lines.pop(0) ).group(1) headers = {} for line in headers_lines: m = re.search(r'(.*)\:\s(.*)', line) if m.group(1) not in headers: headers[m.group(1)] = m.group(2) elif isinstance(headers[m.group(1)], list): headers[m.group(1)].append(m.group(2)) else: headers[m.group(1)] = [headers[m.group(1)], m.group(2)] return {'status': int(status), 'headers': headers, 'body': body} def _parse_chunked_body(self, raw_body): if isinstance(raw_body, str): raw_body = bytes(raw_body.encode()) crlf = b'\r\n' chunks = raw_body.split(crlf) if len(chunks) < 3: pytest.fail('Invalid chunked body') if chunks.pop() != b'': pytest.fail('No CRLF at the end of the body') try: last_size = int(chunks[-2], 16) except: pytest.fail('Invalid zero size chunk') if last_size != 0 or chunks[-1] != b'': pytest.fail('Incomplete body') body = b'' while len(chunks) >= 2: try: size = int(chunks.pop(0), 16) except: pytest.fail('Invalid chunk size %s' % str(size)) if size == 0: assert len(chunks) == 1, 'last zero size' break temp_body = crlf.join(chunks) body += temp_body[:size] temp_body = temp_body[size + len(crlf) :] chunks = temp_body.split(crlf) return body def _parse_json(self, resp): headers = resp['headers'] assert 'Content-Type' in headers assert headers['Content-Type'] == 'application/json' resp['body'] = json.loads(resp['body']) return resp def getjson(self, **kwargs): return self.get(json=True, **kwargs) def form_encode(self, fields): is_multipart = False for _, value in fields.items(): if isinstance(value, dict): is_multipart = True break if is_multipart: body, content_type = self.multipart_encode(fields) else: body, content_type = self.form_url_encode(fields) return body, content_type def form_url_encode(self, fields): data = "&".join("%s=%s" % (name, value) for name, value in fields.items()).encode() return data, 'application/x-www-form-urlencoded' def multipart_encode(self, fields): boundary = binascii.hexlify(os.urandom(16)).decode('ascii') body = '' for field, value in fields.items(): filename = '' datatype = '' if isinstance(value, dict): datatype = 'text/plain' filename = value['filename'] if value.get('type'): datatype = value['type'] if not isinstance(value['data'], io.IOBase): pytest.fail('multipart encoding of file requires a stream.') data = value['data'].read() elif isinstance(value, str): data = value else: pytest.fail('multipart requires a string or stream data') body += ( "--%s\r\nContent-Disposition: form-data; name=\"%s\"" ) % (boundary, field) if filename != '': body += "; filename=\"%s\"" % filename body += "\r\n" if datatype != '': body += "Content-Type: %s\r\n" % datatype body += "\r\n%s\r\n" % data body += "--%s--\r\n" % boundary return body.encode(), "multipart/form-data; boundary=%s" % boundary