How to use create_http_request method in localstack

Best Python code snippet using localstack_python

handler.py

Source:handler.py

...7# three important things are: vhost(domain/zone), path, and method/request type8class HttpHandler(BaseHTTPRequestHandler):9    def do_GET(self):10        self.server.logger.info("do_GET@handler.py - Handling request")11        self.create_http_request("GET")12        return self.dummy_response()13    def do_HEAD(self):14        self.server.logger.info("do_HEAD@handler.py - Handling request")15        self.create_http_request("HEAD")16        return self.dummy_response()17    def do_OPTIONS(self):18        self.server.logger.info("do_OPTIONS@handler.py - Handling request")19        self.create_http_request("OPTIONS")20        return self.dummy_response()21    def do_POST(self):22        self.server.logger.info("do_POST@handler.py - Handling request")23        self.create_http_request("POST")24        return self.dummy_response()25    def do_PATCH(self):26        self.server.logger.info("do_PATCH@handler.py - Handling request")27        self.create_http_request("PATCH")28        return self.dummy_response()29    def do_PUT(self):30        self.server.logger.info("do_PUT@handler.py - Handling request")31        self.create_http_request("PUT")32        return self.dummy_response()33    def do_DELETE(self):34        self.server.logger.info("do_DELETE@handler.py - Handling request")35        self.create_http_request("DELETE")36        return self.dummy_response()37    def send_error(self, code, message=None, explain=None):38        self.server.logger.info("send_error@handler.py - Sending error")39        type = "ERROR"40        if hasattr(self, "headers"):41            name = self.headers.get("HOST", self.client_address[0])42        else:43            name = self.client_address[0]44        if ":" in name:45            name = name.split(":")[0]46        path = self.path or "UNKNOWN"47        source_address = self.client_address[0]48        source_port = self.client_address[1]49        protocol = self.get_protocol()50        raw_request = (51            "ERROR: "52            + str(code)53            + "\n"54            + "MESSAGE: "55            + str(message)56            + "\n"57            + "EXPLAIN: "58            + str(explain)59        )60        self.server.api_client.create_http_request(61            name, path, source_address, source_port, type, protocol, raw_request,62        )63        return super().send_error(code, message=None, explain=None)64    def dummy_response(self):65        f = io.BytesIO()66        f.write(b'<!DOCTYPE html PUBLIC "-//W3C//DTD HTML 3.2 Final//EN">')67        f.write(b"<html>\n<title></title>\n")68        f.write(b"<body>\n")69        f.write(b"<h2>Message: Found</h2>\n")70        f.write(b"</body>\n")71        f.write(b"</html>\n")72        length = f.tell()73        f.seek(0)74        self.send_response(200)75        self.send_header("Content-type", "text/html")76        self.send_header("Content-Length", str(length))77        self.end_headers()78        if f:79            self.copyfile(f, self.wfile)80            f.close()81    def create_http_request(self, method):82        self.server.logger.info(83            "create_http_request@handler.py - Creating HTTP request"84        )85        type = method.upper()86        if hasattr(self, "headers"):87            name = self.headers.get("HOST", self.client_address[0])88        else:89            name = self.client_address[0]90        path = self.path91        if ":" in name:92            name = name.split(":")[0]93        source_address = self.client_address[0]94        source_port = self.client_address[1]95        protocol = self.get_protocol()96        raw_request = str(self.build_raw_request(method))97        self.server.api_client.create_http_request(98            name, path, source_address, source_port, type, protocol, raw_request,99        )100    def build_raw_request(self, method):101        self.server.logger.info("build_raw_request@handler.py - Building Raw Request")102        request = self.raw_requestline.decode().rstrip("\n\r")103        self.server.logger.info(104            "build_raw_request@handler.py - Attaching headers to Raw Request"105        )106        if hasattr(self, "headers"):107            for h, v in self.headers.items():108                request = request + "\n" + str(h) + ": " + str(v)109        if method in ["PUT", "POST", "PATCH", "DELETE"]:110            self.server.logger.info(111                "build_raw_request@handler.py - Parsing data for Raw Request"...

hpp_lib.py

Source:hpp_lib.py

...8        for payload in payloads:9            new_url = asp_url_hpp(url, param_name, payload)10            for method in methods:11                requests.append(12                    http_helper.create_http_request(13                        method,14                        new_url,15                        body,16                        headers,17                        payload18                    )19                )20    elif "DATA" in source.upper():21        for payload in payloads:22            new_body = asp_post_hpp(body, param_name, payload)23            for method in methods:24                requests.append(25                    http_helper.create_http_request(26                        method,27                        url,28                        new_body,29                        headers,30                        payload))31    elif "COOKIE" in source.upper():32        for payload in payloads:33            new_headers = asp_cookie_hpp(headers, param_name, payload)34            for method in methods:35                requests.append(36                    http_helper.create_http_request(37                        method,38                        url,39                        body,40                        new_headers,41                        payload))42    return requests43def asp_url_hpp(url, param_name, payload):44    if urlparse.urlparse(url)[4] == '':45        sep = "?"46    else:47        sep = '&'48    for pay_token in payload.split(","):49        url += sep + param_name + "=" + pay_token50        sep = '&'51    return url52def asp_post_hpp(body, param_name, payload):53    if body is None or body == '':54        sep = ""55    else:56        sep = '&'57    for pay_token in payload.split(","):58        body += sep + param_name + "=" + pay_token59        sep = '&'60    return body61def asp_cookie_hpp(headers, param_name, payload):62    new_headers = headers.copy()63    try:64        cookie_value = new_headers.pop('Cookie')65        sep = "&"66    except KeyError:67        cookie_value = ""68        sep = ""69    for pay_token in payload.split(","):70        cookie_value += sep + param_name + "=" + pay_token71        sep = '&'72    new_headers.add("Cookie", cookie_value)73    return new_headers74def param_overwrite(http_helper, param, accepted_source, payload, url, body,75                    headers):76    requests = []77    if "URL" == accepted_source:78        new_url = HTTPHelper.add_url_param(url, param, "")79        new_body = HTTPHelper.add_body_param(body, param, payload)80        new_headers = HTTPHelper.add_cookie_param(headers, param, payload)81        requests.append(http_helper.create_http_request(82                        "GET",83                        new_url,84                        new_body,85                        headers,86                        payload))87        requests.append(http_helper.create_http_request(88                        accepted_source,89                        new_url,90                        body,91                        new_headers,92                        payload))93        #Duplicate param , first is empty and the second contains the payload94        new_url = HTTPHelper.add_url_param(new_url, param, payload)95        requests.append(http_helper.create_http_request(96                        accepted_source,97                        new_url,98                        body,99                        headers,100                        payload))101        return requests102    elif "DATA" == accepted_source:103        new_url = HTTPHelper.add_url_param(url, param, payload)104        new_body = HTTPHelper.add_body_param(body, param, "")105        new_headers = HTTPHelper.add_cookie_param(headers, param, payload)106        #Overwrite body with cookie107        requests.append(http_helper.create_http_request(108                        accepted_source,109                        url,110                        new_body,111                        new_headers,112                        payload))113        #Overwrite body with url114        requests.append(http_helper.create_http_request(115                        accepted_source,116                        new_url,117                        body,118                        headers,119                        payload))120        #Duplicate param , first is empty and the second contains the payload121        new_body = HTTPHelper.add_url_param(new_body, param, payload)122        requests.append(http_helper.create_http_request(123                        accepted_source,124                        url,125                        new_body,126                        headers,127                        payload))128        return requests129    elif "COOKIE" == accepted_source:130        new_url = HTTPHelper.add_url_param(url, param, payload)131        new_body = HTTPHelper.add_body_param(body, param, payload)132        new_headers = HTTPHelper.add_cookie_param(headers, param, "")133        requests.append(http_helper.create_http_request(134                        accepted_source,135                        new_url,136                        body,137                        new_headers,138                        payload))139        requests.append(http_helper.create_http_request(140                        accepted_source,141                        url,142                        new_body,143                        headers,144                        payload))145        #Duplicate param , first is empty and the second contains the payload146        new_headers = HTTPHelper.add_cookie_param(new_headers, param, payload)147        requests.append(http_helper.create_http_request(148                        accepted_source,149                        url,150                        body,151                        new_headers,152                        payload))...

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.