File manager - Edit - /opt/gsutil/third_party/urllib3/dummyserver/app.py
Back
from __future__ import annotations import collections import contextlib import datetime import email.utils import gzip import mimetypes import zlib from io import BytesIO from pathlib import Path from typing import Iterator import trio from quart import Response, make_response, request from quart.typing import ResponseReturnValue from quart_trio import QuartTrio hypercorn_app = QuartTrio(__name__) # Globals are not safe in Flask/Quart but work for our Hypercorn use case RETRY_TEST_NAMES: collections.Counter[str] = collections.Counter() LAST_RETRY_AFTER_REQ: datetime.datetime = datetime.datetime.min pyodide_testing_app = QuartTrio(__name__) DEFAULT_HEADERS = [ # Allow cross-origin requests for emscripten ("Access-Control-Allow-Origin", "*"), ("Cross-Origin-Opener-Policy", "same-origin"), ("Cross-Origin-Embedder-Policy", "require-corp"), ("Feature-Policy", "sync-xhr *;"), ("Access-Control-Allow-Headers", "*"), ] @hypercorn_app.route("/") @pyodide_testing_app.route("/") @pyodide_testing_app.route("/index") async def index() -> ResponseReturnValue: return await make_response("Dummy server!") @hypercorn_app.route("/alpn_protocol") async def alpn_protocol() -> ResponseReturnValue: """Return the requester's certificate.""" alpn_protocol = request.scope["extensions"]["tls"]["alpn_protocol"] return await make_response(alpn_protocol) @hypercorn_app.route("/certificate") async def certificate() -> ResponseReturnValue: """Return the requester's certificate.""" print("scope", request.scope) subject = request.scope["extensions"]["tls"]["client_cert_name"] subject_as_dict = dict(part.split("=") for part in subject.split(", ")) return await make_response(subject_as_dict) @hypercorn_app.route("/specific_method", methods=["GET", "POST", "PUT"]) @pyodide_testing_app.route("/specific_method", methods=["GET", "POST", "PUT"]) async def specific_method() -> ResponseReturnValue: "Confirm that the request matches the desired method type" method_param = (await request.values).get("method", "") if request.method.upper() == method_param.upper(): return await make_response("", 200) else: return await make_response( f"Wrong method: {method_param} != {request.method}", 400 ) @hypercorn_app.route("/upload", methods=["POST"]) async def upload() -> ResponseReturnValue: "Confirm that the uploaded file conforms to specification" params = await request.form param = params.get("upload_param") filename_param = params.get("upload_filename") size = int(params.get("upload_size", "0")) files_ = (await request.files).getlist(param) assert files_ is not None if len(files_) != 1: return await make_response( f"Expected 1 file for '{param}', not {len(files_)}", 400 ) file_ = files_[0] # data is short enough to read synchronously without blocking the event loop with contextlib.closing(file_.stream) as stream: data = stream.read() if int(size) != len(data): return await make_response(f"Wrong size: {int(size)} != {len(data)}", 400) if filename_param != file_.filename: return await make_response( f"Wrong filename: {filename_param} != {file_.filename}", 400 ) return await make_response() @hypercorn_app.route("/chunked") async def chunked() -> ResponseReturnValue: def generate() -> Iterator[str]: for _ in range(4): yield "123" return await make_response(generate()) @hypercorn_app.route("/chunked_gzip") async def chunked_gzip() -> ResponseReturnValue: def generate() -> Iterator[bytes]: compressor = zlib.compressobj(6, zlib.DEFLATED, 16 + zlib.MAX_WBITS) for uncompressed in [b"123"] * 4: yield compressor.compress(uncompressed) yield compressor.flush() return await make_response(generate(), 200, [("Content-Encoding", "gzip")]) @hypercorn_app.route("/keepalive") async def keepalive() -> ResponseReturnValue: if request.args.get("close", b"0") == b"1": headers = [("Connection", "close")] return await make_response("Closing", 200, headers) headers = [("Connection", "keep-alive")] return await make_response("Keeping alive", 200, headers) @hypercorn_app.route("/echo", methods=["GET", "POST", "PUT"]) async def echo() -> ResponseReturnValue: "Echo back the params" if request.method == "GET": return await make_response(request.query_string) return await make_response(await request.get_data()) @hypercorn_app.route("/echo_json", methods=["POST"]) @pyodide_testing_app.route("/echo_json", methods=["POST", "OPTIONS"]) async def echo_json() -> ResponseReturnValue: "Echo back the JSON" if request.method == "OPTIONS": return await make_response("", 200) data = await request.get_data() return await make_response(data, 200, request.headers) @hypercorn_app.route("/echo_uri/<path:rest>") @hypercorn_app.route("/echo_uri", defaults={"rest": ""}) async def echo_uri(rest: str) -> ResponseReturnValue: "Echo back the requested URI" assert request.full_path is not None return await make_response(request.full_path) @hypercorn_app.route("/echo_params") async def echo_params() -> ResponseReturnValue: "Echo back the query parameters" await request.get_data() echod = sorted((k, v) for k, v in request.args.items()) return await make_response(repr(echod)) @hypercorn_app.route("/headers", methods=["GET", "POST"]) async def headers() -> ResponseReturnValue: return await make_response(dict(request.headers.items())) @hypercorn_app.route("/headers_and_params") async def headers_and_params() -> ResponseReturnValue: return await make_response( { "headers": dict(request.headers), "params": request.args, } ) @hypercorn_app.route("/multi_headers", methods=["GET", "POST"]) async def multi_headers() -> ResponseReturnValue: return await make_response({"headers": list(request.headers)}) @hypercorn_app.route("/multi_redirect") async def multi_redirect() -> ResponseReturnValue: "Performs a redirect chain based on ``redirect_codes``" params = request.args codes = params.get("redirect_codes", "200") head, tail = codes.split(",", 1) if "," in codes else (codes, None) assert head is not None status = head if not tail: return await make_response("Done redirecting", status) headers = [("Location", f"/multi_redirect?redirect_codes={tail}")] return await make_response("", status, headers) @hypercorn_app.route("/encodingrequest") async def encodingrequest() -> ResponseReturnValue: "Check for UA accepting gzip/deflate encoding" data = b"hello, world!" encoding = request.headers.get("Accept-Encoding", "") headers = [] if encoding == "gzip": headers = [("Content-Encoding", "gzip")] file_ = BytesIO() with contextlib.closing(gzip.GzipFile("", mode="w", fileobj=file_)) as zipfile: zipfile.write(data) data = file_.getvalue() elif encoding == "deflate": headers = [("Content-Encoding", "deflate")] data = zlib.compress(data) elif encoding == "garbage-gzip": headers = [("Content-Encoding", "gzip")] data = b"garbage" elif encoding == "garbage-deflate": headers = [("Content-Encoding", "deflate")] data = b"garbage" return await make_response(data, 200, headers) @hypercorn_app.route("/redirect", methods=["GET", "POST", "PUT"]) async def redirect() -> ResponseReturnValue: "Perform a redirect to ``target``" values = await request.values target = values.get("target", "/") status = values.get("status", "303 See Other") status_code = status.split(" ")[0] headers = [("Location", target)] return await make_response("", status_code, headers) @hypercorn_app.route("/redirect_after") async def redirect_after() -> ResponseReturnValue: "Perform a redirect to ``target``" params = request.args date = params.get("date") if date: dt = datetime.datetime.fromtimestamp(float(date), tz=datetime.timezone.utc) http_dt = email.utils.format_datetime(dt, usegmt=True) retry_after = str(http_dt) else: retry_after = "1" target = params.get("target", "/") headers = [("Location", target), ("Retry-After", retry_after)] return await make_response("", 303, headers) @hypercorn_app.route("/retry_after") async def retry_after() -> ResponseReturnValue: global LAST_RETRY_AFTER_REQ params = request.args if datetime.datetime.now() - LAST_RETRY_AFTER_REQ < datetime.timedelta(seconds=1): status = params.get("status", "429 Too Many Requests") status_code = status.split(" ")[0] return await make_response("", status_code, [("Retry-After", "1")]) LAST_RETRY_AFTER_REQ = datetime.datetime.now() return await make_response("", 200) @hypercorn_app.route("/status") @pyodide_testing_app.route("/status") async def status() -> ResponseReturnValue: values = await request.values status = values.get("status", "200 OK") status_code = status.split(" ")[0] return await make_response("", status_code) @hypercorn_app.route("/source_address") async def source_address() -> ResponseReturnValue: """Return the requester's IP address.""" return await make_response(request.remote_addr) @hypercorn_app.route("/successful_retry", methods=["GET", "PUT"]) async def successful_retry() -> ResponseReturnValue: """First return an error and then success It's not currently very flexible as the number of retries is hard-coded. """ test_name = request.headers.get("test-name", None) if not test_name: return await make_response("test-name header not set", 400) RETRY_TEST_NAMES[test_name] += 1 if RETRY_TEST_NAMES[test_name] >= 2: return await make_response("Retry successful!", 200) else: return await make_response("need to keep retrying!", 418) @pyodide_testing_app.after_request def apply_caching(response: Response) -> ResponseReturnValue: for header, value in DEFAULT_HEADERS: response.headers[header] = value return response @pyodide_testing_app.route("/slow") async def slow() -> ResponseReturnValue: await trio.sleep(10) return await make_response("TEN SECONDS LATER", 200) @pyodide_testing_app.route("/bigfile") async def bigfile() -> ResponseReturnValue: # great big text file, should force streaming # if supported bigdata = 1048576 * b"WOOO YAY BOOYAKAH" return await make_response(bigdata, 200) @pyodide_testing_app.route("/mediumfile") async def mediumfile() -> ResponseReturnValue: # quite big file bigdata = 1024 * b"WOOO YAY BOOYAKAH" return await make_response(bigdata, 200) @pyodide_testing_app.route("/upload", methods=["POST", "OPTIONS"]) async def pyodide_upload() -> ResponseReturnValue: if request.method == "OPTIONS": return await make_response("", 200) spare_data = await request.get_data(parse_form_data=True) if len(spare_data) != 0: return await make_response("Bad upload data", 404) files = await request.files form = await request.form if form["upload_param"] != "filefield" or form["upload_filename"] != "lolcat.txt": return await make_response("Bad upload form values", 404) if len(files) != 1 or files.get("filefield") is None: return await make_response("Missing file in form", 404) file = files["filefield"] if file.filename != "lolcat.txt": return await make_response(f"File name incorrect {file.name}", 404) with contextlib.closing(file): data = file.read().decode("utf-8") if data != "I'm in ur multipart form-data, hazing a cheezburgr": return await make_response(f"File data incorrect {data}", 200) return await make_response("Uploaded file correct", 200) @pyodide_testing_app.route("/pyodide/<py_file>") async def pyodide(py_file: str) -> ResponseReturnValue: file_path = Path(pyodide_testing_app.config["pyodide_dist_dir"], py_file) if file_path.exists(): mime_type, encoding = mimetypes.guess_type(file_path) if not mime_type: mime_type = "text/plain" return await make_response( file_path.read_bytes(), 200, [("Content-Type", mime_type)] ) else: return await make_response("", 404) @pyodide_testing_app.route("/wheel/dist.whl") async def wheel() -> ResponseReturnValue: # serve our wheel wheel_folder = Path(__file__).parent.parent / "dist" wheels = list(wheel_folder.glob("*.whl")) if len(wheels) > 0: wheel = wheels[0] headers = [("Content-Disposition", f"inline; filename='{wheel.name}'")] resp = await make_response(wheel.read_bytes(), 200, headers) return resp else: return await make_response(f"NO WHEEL IN {wheel_folder}", 404)
| ver. 1.4 |
Github
|
.
| PHP 8.2.28 | Generation time: 0.01 |
proxy
|
phpinfo
|
Settings