KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
How to use handle_http method in localstack
Best Python code snippet using localstack_python
mock_service.py
Source:mock_service.py 
...21JSON_INDENT = 222ENCODING_TYPE = "UTF-8"23config = {}24class BasicServer(BaseHTTPRequestHandler):25 def handle_http(self, status, content_type, response):26 self.send_response(status)27 self.send_header(HEADER_CONTENT_TYPE, content_type)28 self.end_headers()29 self.wfile.write(bytes(response, ENCODING_TYPE))30 def do_GET(self):31 if ".css" in self.get_path():32 self.handle_http(STATUS_OK, CONTENT_TYPE_CSS, read_file(PATH_CSS_FILE))33 elif ".js" in self.get_path():34 self.handle_http(STATUS_OK, CONTENT_TYPE_JS, read_file(PATH_JS_FILE))35 elif ".png" in self.get_path():36 self.send_response(STATUS_OK)37 self.send_header(HEADER_CONTENT_TYPE, CONTENT_TYPE_PNG)38 self.end_headers()39 self.wfile.write(read_image(PATH_IMG_FILE))40 else:41 self.handle_path_request()42 def do_POST(self):43 if self.headers.get_content_type() == CONTENT_TYPE_JSON:44 length = int(self.headers.get(HEADER_CONTENT_LENGTH))45 body = self.rfile.read(length)46 if self.path == "/serviceNames":47 self.handle_http(STATUS_OK, CONTENT_TYPE_JSON, self.create_json_from_string(get_routes_as_json()))48 elif self.path == "/OPR_UPDATE":49 if self.is_data_json(body.decode(ENCODING_TYPE)):50 name = json.loads(body.decode(ENCODING_TYPE))["serviceNameSelect"]51 data = json.loads(body.decode(ENCODING_TYPE))["jsonTextArea"]52 if self.is_data_json(data):53 is_updated = update_file(os.path.join(PATH_RESPONSE_FOLDER, "{}.json".format(name)), data)54 if is_updated:55 self.handle_http(STATUS_OK, CONTENT_TYPE_TEXT, "JSON updated")56 else:57 self.handle_http(STATUS_ERROR, CONTENT_TYPE_TEXT, "JSON is NOT updated")58 else:59 self.handle_http(STATUS_ERROR, CONTENT_TYPE_TEXT, "Malformed JSON text")60 else:61 self.handle_http(STATUS_ERROR, CONTENT_TYPE_TEXT, "Malformed JSON")62 elif self.path == "/OPR_INSERT":63 if self.is_data_json(body.decode(ENCODING_TYPE)):64 name = json.loads(body.decode(ENCODING_TYPE))["serviceNameInput"]65 requested_route = self.get_route_name(name, get_route_list())66 if requested_route:67 self.handle_http(STATUS_ERROR, CONTENT_TYPE_TEXT, "\"{}\" service already exists".format(name))68 else:69 data = json.loads(body.decode(ENCODING_TYPE))["jsonTextArea"]70 if self.is_data_json(data):71 is_added = write_file(os.path.join(PATH_RESPONSE_FOLDER, "{}.json".format(name)), data)72 if is_added:73 self.handle_http(STATUS_OK, CONTENT_TYPE_TEXT, "JSON added")74 else:75 self.handle_http(STATUS_ERROR, CONTENT_TYPE_TEXT, "JSON is NOT added")76 else:77 self.handle_http(STATUS_ERROR, CONTENT_TYPE_TEXT, "Malformed JSON text")78 else:79 self.handle_http(STATUS_ERROR, CONTENT_TYPE_TEXT, "Malformed JSON")80 elif self.path == "/OPR_SELECT":81 if self.is_data_json(body.decode(ENCODING_TYPE)):82 name = json.loads(body.decode(ENCODING_TYPE))["serviceNameSelect"]83 requested_route = self.get_route_name(name, get_route_list())84 if requested_route:85 self.handle_http(STATUS_OK, CONTENT_TYPE_JSON, self.create_json_from_file(requested_route))86 else:87 self.handle_http(STATUS_ERROR, CONTENT_TYPE_TEXT, "\"{}\" service does not exist".format(name))88 else:89 self.handle_http(STATUS_ERROR, CONTENT_TYPE_TEXT, "Malformed JSON")90 elif self.path == "/OPR_DELETE":91 if self.is_data_json(body.decode(ENCODING_TYPE)):92 name = json.loads(body.decode(ENCODING_TYPE))["serviceNameSelect"]93 requested_route = self.get_route_name(name, get_route_list())94 if requested_route:95 is_deleted = delete_file(requested_route)96 if is_deleted:97 self.handle_http(STATUS_OK, CONTENT_TYPE_TEXT, "\"{}\" service JSON deleted.".format(name))98 else:99 self.handle_http(STATUS_ERROR, CONTENT_TYPE_TEXT,100 "Cannot delete \"{}\" service JSON.".format(name))101 else:102 self.handle_http(STATUS_ERROR, CONTENT_TYPE_TEXT, "\"{}\" service does not exist".format(name))103 else:104 self.handle_http(STATUS_ERROR, CONTENT_TYPE_TEXT, "Malformed JSON")105 else:106 self.handle_path_request()107 def handle_path_request(self):108 if self.path.count("/") == 1:109 if self.path.split("/")[1]:110 requested_route = self.get_route_name(self.get_path(), get_route_list())111 if requested_route:112 self.handle_http(STATUS_OK, CONTENT_TYPE_JSON, self.create_json_from_file(requested_route))113 else:114 self.handle_http(STATUS_ERROR, CONTENT_TYPE_TEXT,115 "\"{}\" service response does not exist".format(self.get_path()))116 else:117 self.handle_http(STATUS_OK, CONTENT_TYPE_TEXT, read_file(PATH_HTML_FILE))118 else:119 self.handle_http(STATUS_ERROR, CONTENT_TYPE_TEXT, "Wrong path")120 @staticmethod121 def get_route_name(requested_route, routes):122 for rt in routes:123 if os.path.basename(rt).split(".")[0] == requested_route:124 return rt125 return None126 def get_path(self):127 return os.path.basename(self.path)128 @staticmethod129 def is_data_json(data):130 try:131 json.loads(data)132 except ValueError:133 return False...
server.py
Source:server.py
...21 body = self.rfile.read(content_len)22 print(body)23 if body == b'1':24 Server.req_power_status = True25 self.wfile.write(self.handle_http(200, "text/html", "GG"))26 elif body == b'0':27 Server.req_power_status = False28 self.wfile.write(self.handle_http(200, "text/html", "GG"))29 else:30 self.wfile.write(self.handle_http(400, "text/html", "BAD REQUEST!"))31 def do_GET(self):32 if self.path == "/on":33 Server.req_power_status = True34 self.wfile.write(self.handle_http(200, "text/html", "GG"))35 elif self.path == "/off":36 Server.req_power_status = False37 self.wfile.write(self.handle_http(200, "text/html", "GG"))38 elif self.path == "/mute":39 Server.req_mute_status = True40 self.wfile.write(self.handle_http(200, "text/html", "GG"))41 elif self.path == "/unmute":42 Server.req_mute_status = False43 self.wfile.write(self.handle_http(200, "text/html", "GG"))44 elif self.path == "/status" or self.path == "/api":45 self.wait_power_status()46 self.wfile.write(self.handle_http(200, "text/html", '1' if Server.power_status else '0'))47 else:48 print(self.path)49 self.wfile.write(self.handle_http(400, "text/html", "BAD REQUEST!"))50 51 def handle_http(self, status, content_type, text):52 self.send_response(status)53 self.send_header('Content-type', content_type)54 self.end_headers()...
test_dos.py
Source:test_dos.py
...17 }18 return slow_body19 handle_http = build_server(lambda x, y: 1, max_receive_timeout_s=0.2)20 with pytest.raises(TimeoutError):21 await handle_http({"type": "http"})(slow_body_reader(0.1), noop_sender)22 with pytest.raises(TimeoutError):23 await handle_http({"type": "http"})(slow_body_reader(1.1), noop_sender)24@pytest.mark.asyncio25async def test_slow_body_receiver():26 async def noop_handler(request):27 return {"status": 200, "body": b"asdf"}28 def slow_body_receiver(timeout_s):29 async def slow_body(x):30 await asyncio.sleep(timeout_s)31 return slow_body32 handle_http = build_server(noop_handler, max_responde_timeout_s=0.1)33 with pytest.raises(TimeoutError):...
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 localstack 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?
