How to use response_405 method in Django Test Plus

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        )...

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.