Best Python code snippet using django-test-plus_python
server.py
Source:server.py
1"""An example of a simple HTTP server."""2import json3import mimetypes4import pickle5import socket6from os.path import isdir7from urllib.parse import unquote_plus8# Pickle file for storing data9PICKLE_DB = "db.pkl"10# Directory containing www data11WWW_DATA = "www-data"12# Header template for a successful HTTP request13HEADER_RESPONSE_200 = """HTTP/1.1 200 OK\r14content-type: %s\r15content-length: %d\r16connection: Close\r17\r18"""19# Represents a table row that holds user data20TABLE_ROW = """21<tr>22 <td>%d</td>23 <td>%s</td>24 <td>%s</td>25</tr>26"""27# Template for a 404 (Not found) error28RESPONSE_404 = """HTTP/1.1 404 Not found\r29content-type: text/html\r30connection: Close\r31\r32<!doctype html>33<h1>404 Page not found</h1>34<p>Page cannot be found.</p>35"""36RESPONSE_301 = """HTTP/1.1 301 Moved Permanently\r37location: %s\r38connection: Close\r39\r40"""41RESPONSE_400 = """HTTP/1.1 400 Bad Request\r42content-type: text/html\r43connection: Close\r44\r45<!doctype html>46<h1>400 Bad Request</h1>47<p>Page cannot be found.</p>48"""49RESPONSE_405 = """HTTP/1.1 405 Method Not Allowed\r50content-type: text/html\r51connection: Close\r52\r53<!doctype html>54<h1>405 Page Not Allowed</h1>55<p>Method Not Allowed.</p>56"""57def save_to_db(first, last):58 """Create a new user with given first and last name and store it into59 file-based database.60 For instance, save_to_db("Mick", "Jagger"), will create a new user61 "Mick Jagger" and also assign him a unique number.62 Do not modify this method."""63 existing = read_from_db()64 existing.append({65 "number": 1 if len(existing) == 0 else existing[-1]["number"] + 1,66 "first": first,67 "last": last68 })69 with open(PICKLE_DB, "wb") as handle:70 pickle.dump(existing, handle)71def read_from_db(criteria=None):72 """Read entries from the file-based DB subject to provided criteria73 Use this method to get users from the DB. The criteria parameters should74 either be omitted (returns all users) or be a dict that represents a query75 filter. For instance:76 - read_from_db({"number": 1}) will return a list of users with number 177 - read_from_db({"first": "bob"}) will return a list of users whose first78 name is "bob".79 Do not modify this method."""80 if criteria is None:81 criteria = {}82 else:83 # remove empty criteria values84 for key in ("number", "first", "last"):85 if key in criteria and criteria[key] == "":86 del criteria[key]87 # cast number to int88 if "number" in criteria:89 criteria["number"] = int(criteria["number"])90 try:91 with open(PICKLE_DB, "rb") as handle:92 data = pickle.load(handle)93 filtered = []94 for entry in data:95 predicate = True96 for key, val in criteria.items():97 if val != entry[key]:98 predicate = False99 if predicate:100 filtered.append(entry)101 return filtered102 except (IOError, EOFError):103 return []104def parse_headers(client):105 headers = dict()106 while True:107 line = client.readline().decode("utf-8").strip()108 if not line:109 return headers110 key, value = line.split(":", 1)111 headers[key.strip().lower()] = value.strip()112def parse_get_parametars(p):113 if p == "":114 return None115 d = dict()116 parametars = p.split("&")117 for i in parametars:118 key, value = i.split("=", 1)119 d[key.strip()] = value.strip()120 return d121def process_request(connection, address):122 client = connection.makefile("wrb")123 line = client.readline().decode("utf-8").strip()124 try:125 method, uri, version = line.split(" ")126 headers = parse_headers(client)127 get_parameters = ""128 if method != "GET" and method != "POST":129 print("405 Error : Wrong Method")130 client.write(RESPONSE_405.encode("utf-8"))131 client.close()132 if uri[0] != "/" and not (len(uri) > 0):133 print("400 Error : Bad Request")134 client.write(RESPONSE_400.encode("utf-8"))135 client.close()136 if version != "HTTP/1.1":137 print("400 Error : Bad Request")138 client.write(RESPONSE_400.encode("utf-8"))139 client.close()140 if "host" not in headers.keys():141 print("400 Error : Bad Request")142 client.write(RESPONSE_400.encode("utf-8"))143 client.close()144 if method == "POST" and "content-length" not in headers.keys():145 print("400 Error : Bad Request, no Content-Length")146 client.write(RESPONSE_400.encode("utf-8"))147 client.close()148 if "?" in uri:149 uri, get_parameters = uri.split("?", 1)150 except Exception as e:151 print("[%s:] Exception parsing '%s': %s" % (address, line, e))152 client.write(RESPONSE_400.encode("utf-8"))153 client.close()154 uri_dynamic = ["/www-data/app-add", "/app-add", "/www-data/app-index", "/app-index", "/www-data/app-json", "/app-json"]155 dynamic = 0156 if uri in uri_dynamic:157 if uri == "/app-index" or uri == "/www-data/app-index":158 uri = "/app_list.html"159 dynamic = 1160 if method != "GET":161 print("405 Error : Wrong Method")162 client.write(RESPONSE_405.encode("utf-8"))163 client.close()164 elif uri == "/app-add" or uri == "/www-data/app-add":165 uri = "/app_add.html"166 dynamic = 2167 if method != "POST":168 print("405 Error : Wrong Method")169 client.write(RESPONSE_405.encode("utf-8"))170 client.close()171 elif uri == "/app-json" or "/www-data/app-json":172 dynamic = 3173 if method != "GET":174 print("405 Error : Wrong Method")175 client.write(RESPONSE_405.encode("utf-8"))176 client.close()177 try:178 get_parameters = unquote_plus(get_parameters, "utf-8")179 body = json.dumps(read_from_db(parse_get_parametars(get_parameters))).encode()180 head = HEADER_RESPONSE_200 % (181 "application/json",182 len(body)183 )184 client.write(head.encode("utf-8"))185 client.write(body)186 except Exception as e:187 print("[%s:] Exception parsing '%s': %s" % (address, line, e))188 client.write(RESPONSE_400.encode("utf-8"))189 client.close()190 if (uri[-1] == "/"):191 r_301 = RESPONSE_301 % ("http://" + headers["host"].split(":")[0] + ":" + headers["host"].split(":")[1] + uri + "index.html")192 uri += "index.html"193 client.write(r_301.encode("utf-8"))194 elif uri[-1] != "/" and "." not in uri: # if doesnt have trailing slash and if its a dicitonary195 if (isdir(WWW_DATA + uri)):196 r_301 = RESPONSE_301 % ("http://" + headers["host"].split(":")[0] + ":" + headers["host"].split(":")[1] + uri + "/index.html")197 uri += "/index.html"198 client.write(r_301.encode("utf-8"))199 if dynamic in [0, 1, 2]:200 try:201 with open(WWW_DATA + "/" + uri, "rb") as handle: # 200 ok202 body = handle.read()203 mime_type, _ = mimetypes.guess_type(WWW_DATA + "/" + uri)204 if mime_type == None:205 mime_type = "application/octet-stream"206 if dynamic == 0:207 head = HEADER_RESPONSE_200 % (208 mime_type,209 len(body)210 )211 client.write(head.encode("utf-8"))212 client.write(body)213 elif dynamic == 1:214 table = ""215 get_parameters = unquote_plus(get_parameters, "utf-8")216 students = read_from_db(parse_get_parametars(get_parameters))217 for student in students:218 table += TABLE_ROW % (219 student["number"],220 student["first"],221 student["last"],222 )223 body = body.replace(b"{{students}}", table.encode("utf-8"))224 head = HEADER_RESPONSE_200 % (225 mime_type,226 len(body)227 )228 client.write(head.encode("utf-8"))229 client.write(body)230 elif dynamic == 2:231 post_parameters = (client.read(int(headers["content-length"])))232 post_parameters = unquote_plus(post_parameters.decode("utf-8"), "utf-8")233 d = parse_get_parametars(post_parameters)234 if "first" not in d.keys() and "last" not in d.keys():235 print("400 Error : Bad Request, missing parameters")236 client.write(RESPONSE_400.encode("utf-8"))237 connection.close()238 save_to_db(str(d["first"]), str(d["last"]))239 head = HEADER_RESPONSE_200 % (240 mime_type,241 len(body)242 )243 client.write(head.encode("utf-8"))244 client.write(body)245 except IOError:246 client.write(RESPONSE_404.encode("utf-8"))247 client.close()248 finally:249 client.close()250 client.close()251def main(port):252 """Starts the server and waits for connections."""253 server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)254 server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)255 server.bind(("", port))256 server.listen(1)257 print("Listening on %d" % port)258 while True:259 connection, address = server.accept()260 print("[%s:%d] CONNECTED" % address)261 process_request(connection, address)262 connection.close()263 print("[%s:%d] DISCONNECTED" % address)264if __name__ == "__main__":...
post_user_job.py
Source:post_user_job.py
1from io import BytesIO2from typing import Any, Dict, Optional, Union3import httpx4from ...client import Client5from ...models.post_user_job_multipart_data import PostUserJobMultipartData6from ...types import File, Response7def _get_kwargs(8 *,9 client: Client,10 multipart_data: PostUserJobMultipartData,11) -> Dict[str, Any]:12 url = "{}/user/job".format(client.base_url)13 headers: Dict[str, Any] = client.get_headers()14 cookies: Dict[str, Any] = client.get_cookies()15 multipart_multipart_data = multipart_data.to_multipart()16 return {17 "url": url,18 "headers": headers,19 "cookies": cookies,20 "timeout": client.get_timeout(),21 "files": multipart_multipart_data,22 }23def _parse_response(*, response: httpx.Response) -> Optional[Union[Any, File]]:24 if response.status_code == 200:25 response_200 = File(payload=BytesIO(response.content))26 return response_20027 if response.status_code == 405:28 response_405 = None29 return response_40530 return None31def _build_response(*, response: httpx.Response) -> Response[Union[Any, File]]:32 return Response(33 status_code=response.status_code,34 content=response.content,35 headers=response.headers,36 parsed=_parse_response(response=response),37 )38def sync_detailed(39 *,40 client: Client,41 multipart_data: PostUserJobMultipartData,42) -> Response[Union[Any, File]]:43 kwargs = _get_kwargs(44 client=client,45 multipart_data=multipart_data,46 )47 response = httpx.post(48 verify=client.verify_ssl,49 **kwargs,50 )51 return _build_response(response=response)52def sync(53 *,54 client: Client,55 multipart_data: PostUserJobMultipartData,56) -> Optional[Union[Any, File]]:57 """A new job is submitted with a specific job script"""58 return sync_detailed(59 client=client,60 multipart_data=multipart_data,61 ).parsed62async def asyncio_detailed(63 *,64 client: Client,65 multipart_data: PostUserJobMultipartData,66) -> Response[Union[Any, File]]:67 kwargs = _get_kwargs(68 client=client,69 multipart_data=multipart_data,70 )71 async with httpx.AsyncClient(verify=client.verify_ssl) as _client:72 response = await _client.post(**kwargs)73 return _build_response(response=response)74async def asyncio(75 *,76 client: Client,77 multipart_data: PostUserJobMultipartData,78) -> Optional[Union[Any, File]]:79 """A new job is submitted with a specific job script"""80 return (81 await asyncio_detailed(82 client=client,83 multipart_data=multipart_data,84 )...
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.
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Get 100 minutes of automation test minutes FREE!!