KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
How to use raw_http_response method in autotest
Best Python code snippet using autotest_python
http.py
Source:http.py 
1# The MIT License (MIT)2#3# Copyright (C) 2021 Simon Meins4#5# Permission is hereby granted, free of charge, to any person obtaining a copy6# of this software and associated documentation files (the "Software"), to deal7# in the Software without restriction, including without limitation the rights8# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell9# copies of the Software, and to permit persons to whom the Software is10# furnished to do so, subject to the following conditions:11#12# The above copyright notice and this permission notice shall be included in13# all copies or substantial portions of the Software.14#15# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR16# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,17# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE18# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER19# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,20# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN21# THE SOFTWARE.22# HTTP request library to send and receive data over the HTTP protocol.23# Import modules24# socket : network communication25# json : JSON manipulation26# namedtuple: special tuple object types27from socket import socket, AF_INET, SOCK_STREAM28from json import dumps29from collections import namedtuple30# Create namedtuple instances31url = namedtuple("url", "scheme netloc path")32status_line = namedtuple("status_line", "http_version status_code reason_phrase")33# Define constants34HTTP_VERSION = "HTTP/1.1"35HTTP_POST = "POST"36HTTP_GET = "GET"37class HTTPResponse:38 39 """40 HTTPResponse class to parse an HTTP response.41 Constructor (__init__):42 parameter raw_http_response [string]: raw HTTP response string43 """44 def __init__(self, raw_http_response: str) -> None:45 self.raw_http_response = raw_http_response46 def __repr__(self) -> str:47 return f"<Response [{self.status_code}]>"48 @property49 def _status_line(self) -> status_line:50 """51 Parse the status line of the HTTP response (private method).52 returns: HTTP status line [namedtuple]53 """54 status_line_end_idx = self.raw_http_response.index("\r\n")55 status_line_raw = self.raw_http_response[:status_line_end_idx].split()56 return status_line(57 http_version=status_line_raw[0], 58 status_code=status_line_raw[1], 59 reason_phrase=status_line_raw[2]60 )61 @property62 def status_code(self) -> int:63 """64 Return the status code of the HTTP response.65 returns: HTTP response status code [integer]66 """67 return int(self._status_line.status_code)68 @property69 def text(self) -> str:70 """71 Parse the HTTP response text (body data).72 returns: HTTP response text [string]73 """74 text_start_idx = self.raw_http_response.index("\r\n\r\n")75 http_response_text = self.raw_http_response[text_start_idx + 4:]76 return http_response_text77 @property78 def headers(self) -> dict:79 """80 Parse the HTTP response headers.81 returns: HTTP response headers [dictionary]82 """83 84 header_start_idx = self.raw_http_response.index("\r\n")85 header_end_idx = self.raw_http_response.index("\r\n\r\n")86 http_headers = self.raw_http_response[header_start_idx + 2:header_end_idx].replace(" ", "")87 return {key_value[0]: key_value[1] for header in http_headers.split("\r\n") if (key_value := header.split(":", 1))}88class HTTPRequest:89 """90 HTTPRequest class to construct and send an HTTP request.91 This HTTPRequest class implements a subset of the HTTP protocol version 1.1 which is defined in RFC2616.92 Constructor (__init__):93 parameter method [string]: HTTP request method (e.g. GET, POST, etc.)94 parameter url [string]: Destination URL for the HTTP request95 parameter data [dictionary]: HTTP request parameters96 parameter json [dictionary]: HTTP request JSON data (POST request only)97 parameter headers [dictionary]: HTTP request headers98 parameter timeout [integer]: socket timeout99 """100 def __init__(self, method: str, url: str, data: dict = {}, json: dict = {}, headers: dict = {}, timeout: float = 3.0) -> None:101 self.method = method102 self.url = url103 self.data = data104 self.json = json105 self.headers = headers106 self.timeout = timeout107 self.json_body: bool = True if self.json else False108 self.socket = socket(AF_INET, SOCK_STREAM)109 self.socket.settimeout(timeout)110 dest_host = self._parse_url.netloc[1]111 dest_port = self._parse_url.netloc[2]112 self.socket.connect((dest_host, int(dest_port)))113 def content_length(self, json: bool = True) -> int:114 """115 Calculate the content length of the HTTP request data.116 returns: content length [integer]117 """118 119 if not json:120 return len(self._parse_http_data())121 return len(self._parse_http_json())122 def _parse_http_headers(self) -> str:123 """124 Parse the HTTP request headers. Include 'Host' and 'Content-Length' header by default.125 returns: formatted HTTP request headers [string]126 """127 128 http_headers: str = ""129 dest_host = self._parse_url.netloc[0]130 http_headers += f"Host: {dest_host}\r\n"131 if self.method == HTTP_POST:132 http_headers += f"Content-Length: {self.content_length(self.json_body)}\r\n"133 for header_key, header_value in self.headers.items():134 http_headers += f"{header_key}: {header_value}\r\n"135 return http_headers136 137 @property138 def _parse_url(self) -> url:139 """140 Parse the destination URL.141 returns: destination URL components [namedtuple]142 """143 144 if not self.url.startswith("http"):145 raise ValueError("Invalid URL")146 delimiter = "/"147 url_parts = self.url.split(delimiter)148 scheme = url_parts[0].replace(":", "")149 netloc_abs = url_parts[2]150 netloc_host: str = ""151 netloc_port: str = ""152 if ":" in url_parts[2]:153 netloc_parts = netloc_abs.split(":")154 netloc_host, netloc_port = netloc_parts[0], netloc_parts[1]155 else:156 netloc_host, netloc_port = url_parts[2], 80157 path = ''.join([f"/{path}" for path in url_parts[3:]])158 return url(159 scheme=scheme, 160 netloc=(netloc_abs, netloc_host, netloc_port), 161 path=path162 )163 def _parse_http_json(self) -> str:164 """165 Convert JSON data to string.166 returns: JSON data [string]167 """168 return dumps(self.json)169 def _parse_http_data(self) -> str:170 """171 Parse the HTTP request parameters.172 returns: formatted HTTP request parameters [string]173 """174 request_data: str = ""175 for i, (key, data) in enumerate(self.data.items()):176 fmt = f"{key}={data}"177 if i != len(self.data) - 1: fmt += "&"178 request_data += fmt179 return request_data180 181 def construct_http_request(self) -> str:182 """183 Construct the final HTTP request string.184 returns: constructed HTTP request string [string] 185 """186 parsed_url = self._parse_url187 http_headers = self._parse_http_headers()188 body_data = self._parse_http_json()189 if not self.json:190 body_data = self._parse_http_data()191 if self.method == HTTP_GET:192 return f"{self.method} {parsed_url.path}{f'?{body_data}' if self.data and not self.json else ''} {HTTP_VERSION}\r\n{http_headers}\r\n"193 return f"{self.method} {parsed_url.path} {HTTP_VERSION}\r\n{http_headers}\r\n{body_data}"194 195 def send_http_request(self) -> None:196 """197 Construct and send a new HTTP request.198 """199 http_request = str.encode(self.construct_http_request())200 self.socket.send(http_request)201 def get_http_response(self) -> HTTPResponse:202 """203 Receive and parse the HTTP response.204 returns: parsed HTTP response [HTTPResponse]205 """206 207 response_buffer = b""208 while True:209 data = self.socket.recv(1024)210 if not data:211 break212 response_buffer += data213 self.socket.close()214 return HTTPResponse(response_buffer.decode("utf-8"))215def get(url: str, data: dict = {}, headers: dict = {}, timeout: float = 3.0) -> HTTPResponse:216 """217 Wrapper function to make a new HTTP GET request.218 returns: HTTP response [HTTPResponse]219 """220 http_request_instance = HTTPRequest(method="GET", url=url, data=data, headers=headers, timeout=timeout)221 http_request_instance.send_http_request()222 return http_request_instance.get_http_response()223def post(url: str, data: dict = {}, json: dict = {}, headers: dict = {}, timeout: float = 3.0) -> HTTPResponse:224 """225 Wrapper function to make a new HTTP POST request.226 returns: HTTP response [HTTPResponse]227 """228 http_request_instance = HTTPRequest("POST", url=url, data=data, json=json, headers=headers, timeout=timeout)229 http_request_instance.send_http_request()...
pinpoint_censor.py
Source:pinpoint_censor.py
...124 dns_result = process_raw_dns_response(raw_dns_response, is_timeout)125 dns_result['device'] = addr126 return dns_result127############################################# HTTP Part ##########################################128def process_raw_http_response(raw_http_response, is_timeout):129 http_result = dict()130 http_result['timestamp'] = int(time.time())131 http_result['status'] = 'success'132 http_result['status_code'] = 0133 #http_result['url'] = ''134 http_result['text'] = ''135 http_result['headers'] = dict()136 http_result['is_timeout'] = is_timeout137 class FakeSocket():138 def __init__(self, response_bytes):139 self._file = BytesIO(response_bytes)140 def makefile(self, *args, **kwargs):141 return self._file142 if not is_timeout:143 if raw_http_response == b'':144 http_result['status'] = 'fail'145 else:146 source = FakeSocket(raw_http_response)147 response = HTTPResponse(source)148 response.begin()149 http_result['text'] = response.read(len(raw_http_response)).decode()150 #http_result['url'] = raw_http_response.url151 http_result['status_code'] = response.status152 http_result['headers'] = dict(response.getheaders())153 else:154 http_result['status'] = 'fail'155 return http_result156def recvall(sock):157 data = b''158 bufsize = 4096159 while True:160 packet = sock.recv(bufsize)161 data += packet162 if len(packet) < bufsize:163 break164 return data165def http_request(domain, server, ttl, timeout = 5):166 request = "GET / HTTP/1.1\r\nHost: %s\r\nUser-Agent: Mozilla/5.0\r\n\r\n" % domain167 request = request.encode()168 sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)169 sock.settimeout(timeout)170 icmp_sock = socket.socket(socket.AF_INET, socket.SOCK_RAW, socket.IPPROTO_ICMP)171 icmp_sock.settimeout(1)172 try:173 # tcp handshake174 sock.connect((server, 80))175 port = sock.getsockname()[1]176 #print('send port', port)177 178 sock.setsockopt(socket.IPPROTO_IP, socket.IP_TTL, struct.pack('I', ttl))179 sock.send(request)180 time.sleep(1)181 182 raw_http_response = recvall(sock)183 is_timeout = False184 addr = get_router_ip(icmp_sock, port)185 186 # close socket187 # sock.shutdown(socket.SHUT_RDWR)188 # sock.close()189 except socket.timeout:190 raw_http_response = b''191 is_timeout = True192 addr = get_router_ip(icmp_sock, port)193 194 except SocketError as e:195 raw_http_response = b''196 is_timeout = False197 addr = '!'198 except Exception as e:199 print(e)200 raw_http_response = b''201 is_timeout = False202 addr = '!'203 204 http_result = process_raw_http_response(raw_http_response, is_timeout)205 http_result['device'] = addr206 return http_result207############################################# SNI Part ##########################################208 209def sni_request(domain, server, ttl, timeout = 5):210 sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)211 sock.settimeout(timeout)212 icmp_sock = socket.socket(socket.AF_INET, socket.SOCK_RAW, socket.IPPROTO_ICMP)213 icmp_sock.settimeout(1)214 context = ssl.SSLContext(ssl.PROTOCOL_TLS)215 sni_result = dict()216 sni_result['timestamp'] = int(time.time())217 sni_result['cert'] = ''218 sni_result['cert_serial'] = '0'...
proxy_request.py
Source:proxy_request.py
...93 is_timeout = False94 dns_result = process_raw_dns_response(raw_dns_response, is_timeout)95 return domain, dns_result96############################################# HTTP Part ##########################################97def process_raw_http_response(raw_http_response, is_timeout):98 http_result = dict()99 http_result['timestamp'] = int(time.time())100 http_result['status'] = 'success'101 http_result['status_code'] = 0102 http_result['url'] = ''103 http_result['text'] = ''104 http_result['headers'] = dict()105 http_result['is_timeout'] = is_timeout106 if not is_timeout:107 if raw_http_response == '':108 http_result['status'] = 'fail'109 else:110 http_result['text'] = raw_http_response.text111 http_result['url'] = raw_http_response.url112 http_result['status_code'] = raw_http_response.status_code113 http_result['headers'] = dict(raw_http_response.headers)114 else:115 http_result['status'] = 'fail'116 return http_result117def proxy_http(domain, server, proxy = {}, timeout = 5):118 through_proxy, proxy_address, proxy_port, username, password = unpack_proxy_args(proxy)119 120 if through_proxy:121 proxy = {"http":"socks5h://{}:{}@{}".format(username, password, proxy_address + ':' + str(proxy_port))}122 else:123 proxy = {}124 headers = dict()125 headers['Host'] = domain126 headers['User-Agent'] = 'Mozilla/5.0'127 128 url = 'http://' + server129 try:130 raw_http_response = requests.get(url, proxies=proxy, headers = headers, timeout = timeout)131 is_timeout = False132 133 except requests.exceptions.Timeout:134 raw_http_response = ''135 is_timeout = True136 except:137 raw_http_response = ''138 is_timeout = False139 140 http_result = process_raw_http_response(raw_http_response, is_timeout)141 return domain, http_result142############################################# SNI Part ##########################################143def process_raw_http_response_from_sni(raw_http_response, is_timeout):144 http_result = dict()145 http_result['timestamp'] = int(time.time())146 http_result['status'] = 'success'147 http_result['status_code'] = 0148 #http_result['url'] = ''149 http_result['text'] = ''150 http_result['headers'] = dict()151 http_result['is_timeout'] = is_timeout152 class FakeSocket():153 def __init__(self, response_bytes):154 self._file = BytesIO(response_bytes)...
Automation Testing Tutorials
Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.
LambdaTest Learning Hubs:
- JUnit Tutorial
- TestNG Tutorial
- Webdriver Tutorial
- WebDriverIO Tutorial
- Protractor Tutorial
- Selenium 4 Tutorial
- Jenkins Tutorial
- NUnit Tutorial
- Jest Tutorial
- Playwright Tutorial
- Cypress Tutorial
- PyTest Tutorial
YouTube
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Run autotest automation tests on LambdaTest cloud grid
Perform automation testing on 3000+ real desktop and mobile devices online.
Try LambdaTest Now !!
Get 100 minutes of automation test minutes FREE!!
Was this article helpful?
