7028eae083
* Streaming request by async for.
* Make all requests streaming and preload body for non-streaming handlers.
* Cleanup of code and avoid mixing streaming responses.
* Async http protocol loop.
* Change of test: don't require early bad request error but only after CRLF-CRLF.
* Add back streaming requests.
* Rewritten request body parser.
* Misc. cleanup, down to 4 failing tests.
* All tests OK.
* Entirely remove request body queue.
* Let black f*ckup the layout
* Better testing error messages on protocol errors.
* Remove StreamBuffer tests because the type is about to be removed.
* Remove tests using the deprecated get_headers function that can no longer be supported. Chunked mode is now autodetected, so do not put content-length header if chunked mode is preferred.
* Major refactoring of HTTP protocol handling (new module http.py added), all requests made streaming. A few compatibility issues and a lot of cleanup to be done remain, 16 tests failing.
* Terminate check_timeouts once connection_task finishes.
* Code cleanup, 14 tests failing.
* Much cleanup, 12 failing...
* Even more cleanup and error checking, 8 failing tests.
* Remove keep-alive header from responses. First of all, it should say timeout=<value> which wasn't the case with existing implementation, and secondly none of the other web servers I tried include this header.
* Everything but CustomServer OK.
* Linter
* Disable custom protocol test
* Remove unnecessary variables, optimise performance.
* A test was missing that body_init/body_push/body_finish are never called. Rewritten using receive_body and case switching to make it fail if bypassed.
* Minor fixes.
* Remove unused code.
* Py 3.8 check for deprecated loop argument.
* Fix a middleware cancellation handling test with py38.
* Linter 'n fixes
* Typing
* Stricter handling of request header size
* More specific error messages on Payload Too Large.
* Init http.response = None
* Messages further tuned.
* Always try to consume request body, plus minor cleanup.
* Add a missing check in case of close_if_idle on a dead connection.
* Avoid error messages on PayloadTooLarge.
* Add test for new API.
* json takes str, not bytes
* Default to no maximum request size for streaming handlers.
* Fix chunked mode crash.
* Header values should be strictly ASCII but both UTF-8 and Latin-1 exist. Use UTF-8B to
cope with all.
* Refactoring and cleanup.
* Unify response header processing of ASGI and asyncio modes.
* Avoid special handling of StreamingHTTPResponse.
* 35 % speedup in HTTP/1.1 response formatting (not so much overall effect).
* Duplicate set-cookie headers were being produced.
* Cleanup processed_headers some more.
* Linting
* Import ordering
* Response middleware ran by async request.respond().
* Need to check if transport is closing to avoid getting stuck in sending loops after peer has disconnected.
* Middleware and error handling refactoring.
* Linter
* Fix tracking of HTTP stage when writing to transport fails.
* Add clarifying comment
* Add a check for request body functions and a test for NotImplementedError.
* Linter and typing
* These must be tuples + hack mypy warnings away.
* New streaming test and minor fixes.
* Constant receive buffer size.
* 256 KiB send and receive buffers.
* Revert "256 KiB send and receive buffers."
This reverts commit abc1e3edb2.
* app.handle_exception already sends the response.
* Improved handling of errors during request.
* An odd hack to avoid an httpx limitation that causes test failures.
* Limit request header size to 8 KiB at most.
* Remove unnecessary use of format string.
* Cleanup tests
* Remove artifact
* Fix type checking
* Mark test for skipping
* Cleanup some edge cases
* Add ignore_body flag to safe methods
* Add unit tests for timeout logic
* Add unit tests for timeout logic
* Fix Mock usage in timeout test
* Change logging test to only logger in handler
* Windows py3.8 logging issue with current testing client
* Add test_header_size_exceeded
* Resolve merge conflicts
* Add request middleware to hard exception handling
* Add request middleware to hard exception handling
* Request middleware on exception handlers
* Linting
* Cleanup deprecations
Co-authored-by: L. Kärkkäinen <tronic@users.noreply.github.com>
Co-authored-by: Adam Hopkins <admhpkns@gmail.com>
895 lines
23 KiB
Python
895 lines
23 KiB
Python
import asyncio
|
|
|
|
import pytest
|
|
|
|
from sanic import Sanic
|
|
from sanic.constants import HTTP_METHODS
|
|
from sanic.response import json, text
|
|
from sanic.router import ParameterNameConflicts, RouteDoesNotExist, RouteExists
|
|
from sanic.testing import SanicTestClient
|
|
|
|
|
|
# ------------------------------------------------------------ #
|
|
# UTF-8
|
|
# ------------------------------------------------------------ #
|
|
|
|
|
|
@pytest.mark.parametrize("method", HTTP_METHODS)
|
|
def test_versioned_routes_get(app, method):
|
|
method = method.lower()
|
|
|
|
func = getattr(app, method)
|
|
if callable(func):
|
|
|
|
@func(f"/{method}", version=1)
|
|
def handler(request):
|
|
return text("OK")
|
|
|
|
else:
|
|
print(func)
|
|
raise Exception(f"Method: {method} is not callable")
|
|
|
|
client_method = getattr(app.test_client, method)
|
|
|
|
request, response = client_method(f"/v1/{method}")
|
|
assert response.status == 200
|
|
|
|
|
|
def test_shorthand_routes_get(app):
|
|
@app.get("/get")
|
|
def handler(request):
|
|
return text("OK")
|
|
|
|
request, response = app.test_client.get("/get")
|
|
assert response.text == "OK"
|
|
|
|
request, response = app.test_client.post("/get")
|
|
assert response.status == 405
|
|
|
|
|
|
def test_shorthand_routes_multiple(app):
|
|
@app.get("/get")
|
|
def get_handler(request):
|
|
return text("OK")
|
|
|
|
@app.options("/get")
|
|
def options_handler(request):
|
|
return text("")
|
|
|
|
request, response = app.test_client.get("/get/")
|
|
assert response.status == 200
|
|
assert response.text == "OK"
|
|
|
|
request, response = app.test_client.options("/get/")
|
|
assert response.status == 200
|
|
|
|
|
|
def test_route_strict_slash(app):
|
|
@app.get("/get", strict_slashes=True)
|
|
def handler1(request):
|
|
return text("OK")
|
|
|
|
@app.post("/post/", strict_slashes=True)
|
|
def handler2(request):
|
|
return text("OK")
|
|
|
|
request, response = app.test_client.get("/get")
|
|
assert response.text == "OK"
|
|
|
|
request, response = app.test_client.get("/get/")
|
|
assert response.status == 404
|
|
|
|
request, response = app.test_client.post("/post/")
|
|
assert response.text == "OK"
|
|
|
|
request, response = app.test_client.post("/post")
|
|
assert response.status == 404
|
|
|
|
|
|
def test_route_invalid_parameter_syntax(app):
|
|
with pytest.raises(ValueError):
|
|
|
|
@app.get("/get/<:string>", strict_slashes=True)
|
|
def handler(request):
|
|
return text("OK")
|
|
|
|
request, response = app.test_client.get("/get")
|
|
|
|
|
|
def test_route_strict_slash_default_value():
|
|
app = Sanic("test_route_strict_slash", strict_slashes=True)
|
|
|
|
@app.get("/get")
|
|
def handler(request):
|
|
return text("OK")
|
|
|
|
request, response = app.test_client.get("/get/")
|
|
assert response.status == 404
|
|
|
|
|
|
def test_route_strict_slash_without_passing_default_value(app):
|
|
@app.get("/get")
|
|
def handler(request):
|
|
return text("OK")
|
|
|
|
request, response = app.test_client.get("/get/")
|
|
assert response.text == "OK"
|
|
|
|
|
|
def test_route_strict_slash_default_value_can_be_overwritten():
|
|
app = Sanic("test_route_strict_slash", strict_slashes=True)
|
|
|
|
@app.get("/get", strict_slashes=False)
|
|
def handler(request):
|
|
return text("OK")
|
|
|
|
request, response = app.test_client.get("/get/")
|
|
assert response.text == "OK"
|
|
|
|
|
|
def test_route_slashes_overload(app):
|
|
@app.get("/hello/")
|
|
def handler_get(request):
|
|
return text("OK")
|
|
|
|
@app.post("/hello/")
|
|
def handler_post(request):
|
|
return text("OK")
|
|
|
|
request, response = app.test_client.get("/hello")
|
|
assert response.text == "OK"
|
|
|
|
request, response = app.test_client.get("/hello/")
|
|
assert response.text == "OK"
|
|
|
|
request, response = app.test_client.post("/hello")
|
|
assert response.text == "OK"
|
|
|
|
request, response = app.test_client.post("/hello/")
|
|
assert response.text == "OK"
|
|
|
|
|
|
def test_route_optional_slash(app):
|
|
@app.get("/get")
|
|
def handler(request):
|
|
return text("OK")
|
|
|
|
request, response = app.test_client.get("/get")
|
|
assert response.text == "OK"
|
|
|
|
request, response = app.test_client.get("/get/")
|
|
assert response.text == "OK"
|
|
|
|
|
|
def test_route_strict_slashes_set_to_false_and_host_is_a_list(app):
|
|
# Part of regression test for issue #1120
|
|
|
|
test_client = SanicTestClient(app, port=42101)
|
|
site1 = f"127.0.0.1:{test_client.port}"
|
|
|
|
# before fix, this raises a RouteExists error
|
|
@app.get("/get", host=[site1, "site2.com"], strict_slashes=False)
|
|
def get_handler(request):
|
|
return text("OK")
|
|
|
|
request, response = test_client.get("http://" + site1 + "/get")
|
|
assert response.text == "OK"
|
|
|
|
@app.post("/post", host=[site1, "site2.com"], strict_slashes=False)
|
|
def post_handler(request):
|
|
return text("OK")
|
|
|
|
request, response = test_client.post("http://" + site1 + "/post")
|
|
assert response.text == "OK"
|
|
|
|
@app.put("/put", host=[site1, "site2.com"], strict_slashes=False)
|
|
def put_handler(request):
|
|
return text("OK")
|
|
|
|
request, response = test_client.put("http://" + site1 + "/put")
|
|
assert response.text == "OK"
|
|
|
|
@app.delete("/delete", host=[site1, "site2.com"], strict_slashes=False)
|
|
def delete_handler(request):
|
|
return text("OK")
|
|
|
|
request, response = test_client.delete("http://" + site1 + "/delete")
|
|
assert response.text == "OK"
|
|
|
|
|
|
def test_shorthand_routes_post(app):
|
|
@app.post("/post")
|
|
def handler(request):
|
|
return text("OK")
|
|
|
|
request, response = app.test_client.post("/post")
|
|
assert response.text == "OK"
|
|
|
|
request, response = app.test_client.get("/post")
|
|
assert response.status == 405
|
|
|
|
|
|
def test_shorthand_routes_put(app):
|
|
@app.put("/put")
|
|
def handler(request):
|
|
return text("OK")
|
|
|
|
request, response = app.test_client.put("/put")
|
|
assert response.text == "OK"
|
|
|
|
request, response = app.test_client.get("/put")
|
|
assert response.status == 405
|
|
|
|
|
|
def test_shorthand_routes_delete(app):
|
|
@app.delete("/delete")
|
|
def handler(request):
|
|
return text("OK")
|
|
|
|
request, response = app.test_client.delete("/delete")
|
|
assert response.text == "OK"
|
|
|
|
request, response = app.test_client.get("/delete")
|
|
assert response.status == 405
|
|
|
|
|
|
def test_shorthand_routes_patch(app):
|
|
@app.patch("/patch")
|
|
def handler(request):
|
|
return text("OK")
|
|
|
|
request, response = app.test_client.patch("/patch")
|
|
assert response.text == "OK"
|
|
|
|
request, response = app.test_client.get("/patch")
|
|
assert response.status == 405
|
|
|
|
|
|
def test_shorthand_routes_head(app):
|
|
@app.head("/head")
|
|
def handler(request):
|
|
return text("OK")
|
|
|
|
request, response = app.test_client.head("/head")
|
|
assert response.status == 200
|
|
|
|
request, response = app.test_client.get("/head")
|
|
assert response.status == 405
|
|
|
|
|
|
def test_shorthand_routes_options(app):
|
|
@app.options("/options")
|
|
def handler(request):
|
|
return text("OK")
|
|
|
|
request, response = app.test_client.options("/options")
|
|
assert response.status == 200
|
|
|
|
request, response = app.test_client.get("/options")
|
|
assert response.status == 405
|
|
|
|
|
|
def test_static_routes(app):
|
|
@app.route("/test")
|
|
async def handler1(request):
|
|
return text("OK1")
|
|
|
|
@app.route("/pizazz")
|
|
async def handler2(request):
|
|
return text("OK2")
|
|
|
|
request, response = app.test_client.get("/test")
|
|
assert response.text == "OK1"
|
|
|
|
request, response = app.test_client.get("/pizazz")
|
|
assert response.text == "OK2"
|
|
|
|
|
|
def test_dynamic_route(app):
|
|
results = []
|
|
|
|
@app.route("/folder/<name>")
|
|
async def handler(request, name):
|
|
results.append(name)
|
|
return text("OK")
|
|
|
|
request, response = app.test_client.get("/folder/test123")
|
|
|
|
assert response.text == "OK"
|
|
assert results[0] == "test123"
|
|
|
|
|
|
def test_dynamic_route_string(app):
|
|
results = []
|
|
|
|
@app.route("/folder/<name:string>")
|
|
async def handler(request, name):
|
|
results.append(name)
|
|
return text("OK")
|
|
|
|
request, response = app.test_client.get("/folder/test123")
|
|
|
|
assert response.text == "OK"
|
|
assert results[0] == "test123"
|
|
|
|
request, response = app.test_client.get("/folder/favicon.ico")
|
|
|
|
assert response.text == "OK"
|
|
assert results[1] == "favicon.ico"
|
|
|
|
|
|
def test_dynamic_route_int(app):
|
|
results = []
|
|
|
|
@app.route("/folder/<folder_id:int>")
|
|
async def handler(request, folder_id):
|
|
results.append(folder_id)
|
|
return text("OK")
|
|
|
|
request, response = app.test_client.get("/folder/12345")
|
|
assert response.text == "OK"
|
|
assert type(results[0]) is int
|
|
|
|
request, response = app.test_client.get("/folder/asdf")
|
|
assert response.status == 404
|
|
|
|
|
|
def test_dynamic_route_number(app):
|
|
results = []
|
|
|
|
@app.route("/weight/<weight:number>")
|
|
async def handler(request, weight):
|
|
results.append(weight)
|
|
return text("OK")
|
|
|
|
request, response = app.test_client.get("/weight/12345")
|
|
assert response.text == "OK"
|
|
assert type(results[0]) is float
|
|
|
|
request, response = app.test_client.get("/weight/1234.56")
|
|
assert response.status == 200
|
|
|
|
request, response = app.test_client.get("/weight/.12")
|
|
assert response.status == 200
|
|
|
|
request, response = app.test_client.get("/weight/12.")
|
|
assert response.status == 200
|
|
|
|
request, response = app.test_client.get("/weight/1234-56")
|
|
assert response.status == 404
|
|
|
|
request, response = app.test_client.get("/weight/12.34.56")
|
|
assert response.status == 404
|
|
|
|
|
|
def test_dynamic_route_regex(app):
|
|
@app.route("/folder/<folder_id:[A-Za-z0-9]{0,4}>")
|
|
async def handler(request, folder_id):
|
|
return text("OK")
|
|
|
|
request, response = app.test_client.get("/folder/test")
|
|
assert response.status == 200
|
|
|
|
request, response = app.test_client.get("/folder/test1")
|
|
assert response.status == 404
|
|
|
|
request, response = app.test_client.get("/folder/test-123")
|
|
assert response.status == 404
|
|
|
|
request, response = app.test_client.get("/folder/")
|
|
assert response.status == 200
|
|
|
|
|
|
def test_dynamic_route_uuid(app):
|
|
import uuid
|
|
|
|
results = []
|
|
|
|
@app.route("/quirky/<unique_id:uuid>")
|
|
async def handler(request, unique_id):
|
|
results.append(unique_id)
|
|
return text("OK")
|
|
|
|
url = "/quirky/123e4567-e89b-12d3-a456-426655440000"
|
|
request, response = app.test_client.get(url)
|
|
assert response.text == "OK"
|
|
assert type(results[0]) is uuid.UUID
|
|
|
|
generated_uuid = uuid.uuid4()
|
|
request, response = app.test_client.get(f"/quirky/{generated_uuid}")
|
|
assert response.status == 200
|
|
|
|
request, response = app.test_client.get("/quirky/non-existing")
|
|
assert response.status == 404
|
|
|
|
|
|
def test_dynamic_route_path(app):
|
|
@app.route("/<path:path>/info")
|
|
async def handler(request, path):
|
|
return text("OK")
|
|
|
|
request, response = app.test_client.get("/path/1/info")
|
|
assert response.status == 200
|
|
|
|
request, response = app.test_client.get("/info")
|
|
assert response.status == 404
|
|
|
|
@app.route("/<path:path>")
|
|
async def handler1(request, path):
|
|
return text("OK")
|
|
|
|
request, response = app.test_client.get("/info")
|
|
assert response.status == 200
|
|
|
|
request, response = app.test_client.get("/whatever/you/set")
|
|
assert response.status == 200
|
|
|
|
|
|
def test_dynamic_route_unhashable(app):
|
|
@app.route("/folder/<unhashable:[A-Za-z0-9/]+>/end/")
|
|
async def handler(request, unhashable):
|
|
return text("OK")
|
|
|
|
request, response = app.test_client.get("/folder/test/asdf/end/")
|
|
assert response.status == 200
|
|
|
|
request, response = app.test_client.get("/folder/test///////end/")
|
|
assert response.status == 200
|
|
|
|
request, response = app.test_client.get("/folder/test/end/")
|
|
assert response.status == 200
|
|
|
|
request, response = app.test_client.get("/folder/test/nope/")
|
|
assert response.status == 404
|
|
|
|
|
|
@pytest.mark.parametrize("url", ["/ws", "ws"])
|
|
def test_websocket_route(app, url):
|
|
ev = asyncio.Event()
|
|
|
|
@app.websocket(url)
|
|
async def handler(request, ws):
|
|
assert request.scheme == "ws"
|
|
assert ws.subprotocol is None
|
|
ev.set()
|
|
|
|
request, response = app.test_client.websocket(url)
|
|
assert response.opened is True
|
|
assert ev.is_set()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize("url", ["/ws", "ws"])
|
|
async def test_websocket_route_asgi(app, url):
|
|
ev = asyncio.Event()
|
|
|
|
@app.websocket(url)
|
|
async def handler(request, ws):
|
|
ev.set()
|
|
|
|
request, response = await app.asgi_client.websocket(url)
|
|
assert ev.is_set()
|
|
|
|
|
|
def test_websocket_route_with_subprotocols(app):
|
|
results = []
|
|
|
|
@app.websocket("/ws", subprotocols=["foo", "bar"])
|
|
async def handler(request, ws):
|
|
results.append(ws.subprotocol)
|
|
assert ws.subprotocol is not None
|
|
|
|
request, response = app.test_client.websocket("/ws", subprotocols=["bar"])
|
|
assert response.opened is True
|
|
assert results == ["bar"]
|
|
|
|
request, response = app.test_client.websocket(
|
|
"/ws", subprotocols=["bar", "foo"]
|
|
)
|
|
assert response.opened is True
|
|
assert results == ["bar", "bar"]
|
|
|
|
request, response = app.test_client.websocket("/ws", subprotocols=["baz"])
|
|
assert response.opened is True
|
|
assert results == ["bar", "bar", None]
|
|
|
|
request, response = app.test_client.websocket("/ws")
|
|
assert response.opened is True
|
|
assert results == ["bar", "bar", None, None]
|
|
|
|
|
|
@pytest.mark.parametrize("strict_slashes", [True, False, None])
|
|
def test_add_webscoket_route(app, strict_slashes):
|
|
ev = asyncio.Event()
|
|
|
|
async def handler(request, ws):
|
|
assert ws.subprotocol is None
|
|
ev.set()
|
|
|
|
app.add_websocket_route(handler, "/ws", strict_slashes=strict_slashes)
|
|
request, response = app.test_client.websocket("/ws")
|
|
assert response.opened is True
|
|
assert ev.is_set()
|
|
|
|
|
|
def test_add_webscoket_route_with_version(app):
|
|
ev = asyncio.Event()
|
|
|
|
async def handler(request, ws):
|
|
assert ws.subprotocol is None
|
|
ev.set()
|
|
|
|
app.add_websocket_route(handler, "/ws", version=1)
|
|
request, response = app.test_client.websocket("/v1/ws")
|
|
assert response.opened is True
|
|
assert ev.is_set()
|
|
|
|
|
|
def test_route_duplicate(app):
|
|
|
|
with pytest.raises(RouteExists):
|
|
|
|
@app.route("/test")
|
|
async def handler1(request):
|
|
pass
|
|
|
|
@app.route("/test")
|
|
async def handler2(request):
|
|
pass
|
|
|
|
with pytest.raises(RouteExists):
|
|
|
|
@app.route("/test/<dynamic>/")
|
|
async def handler3(request, dynamic):
|
|
pass
|
|
|
|
@app.route("/test/<dynamic>/")
|
|
async def handler4(request, dynamic):
|
|
pass
|
|
|
|
|
|
def test_double_stack_route(app):
|
|
@app.route("/test/1")
|
|
@app.route("/test/2")
|
|
async def handler1(request):
|
|
return text("OK")
|
|
|
|
request, response = app.test_client.get("/test/1")
|
|
assert response.status == 200
|
|
request, response = app.test_client.get("/test/2")
|
|
assert response.status == 200
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_websocket_route_asgi(app):
|
|
ev = asyncio.Event()
|
|
|
|
@app.websocket("/test/1")
|
|
@app.websocket("/test/2")
|
|
async def handler(request, ws):
|
|
ev.set()
|
|
|
|
request, response = await app.asgi_client.websocket("/test/1")
|
|
first_set = ev.is_set()
|
|
ev.clear()
|
|
request, response = await app.asgi_client.websocket("/test/1")
|
|
second_set = ev.is_set()
|
|
assert first_set and second_set
|
|
|
|
|
|
def test_method_not_allowed(app):
|
|
@app.route("/test", methods=["GET"])
|
|
async def handler(request):
|
|
return text("OK")
|
|
|
|
request, response = app.test_client.get("/test")
|
|
assert response.status == 200
|
|
|
|
request, response = app.test_client.post("/test")
|
|
assert response.status == 405
|
|
|
|
|
|
@pytest.mark.parametrize("strict_slashes", [True, False, None])
|
|
def test_static_add_route(app, strict_slashes):
|
|
async def handler1(request):
|
|
return text("OK1")
|
|
|
|
async def handler2(request):
|
|
return text("OK2")
|
|
|
|
app.add_route(handler1, "/test", strict_slashes=strict_slashes)
|
|
app.add_route(handler2, "/test2", strict_slashes=strict_slashes)
|
|
|
|
request, response = app.test_client.get("/test")
|
|
assert response.text == "OK1"
|
|
|
|
request, response = app.test_client.get("/test2")
|
|
assert response.text == "OK2"
|
|
|
|
|
|
def test_dynamic_add_route(app):
|
|
|
|
results = []
|
|
|
|
async def handler(request, name):
|
|
results.append(name)
|
|
return text("OK")
|
|
|
|
app.add_route(handler, "/folder/<name>")
|
|
request, response = app.test_client.get("/folder/test123")
|
|
|
|
assert response.text == "OK"
|
|
assert results[0] == "test123"
|
|
|
|
|
|
def test_dynamic_add_route_string(app):
|
|
|
|
results = []
|
|
|
|
async def handler(request, name):
|
|
results.append(name)
|
|
return text("OK")
|
|
|
|
app.add_route(handler, "/folder/<name:string>")
|
|
request, response = app.test_client.get("/folder/test123")
|
|
|
|
assert response.text == "OK"
|
|
assert results[0] == "test123"
|
|
|
|
request, response = app.test_client.get("/folder/favicon.ico")
|
|
|
|
assert response.text == "OK"
|
|
assert results[1] == "favicon.ico"
|
|
|
|
|
|
def test_dynamic_add_route_int(app):
|
|
results = []
|
|
|
|
async def handler(request, folder_id):
|
|
results.append(folder_id)
|
|
return text("OK")
|
|
|
|
app.add_route(handler, "/folder/<folder_id:int>")
|
|
|
|
request, response = app.test_client.get("/folder/12345")
|
|
assert response.text == "OK"
|
|
assert type(results[0]) is int
|
|
|
|
request, response = app.test_client.get("/folder/asdf")
|
|
assert response.status == 404
|
|
|
|
|
|
def test_dynamic_add_route_number(app):
|
|
results = []
|
|
|
|
async def handler(request, weight):
|
|
results.append(weight)
|
|
return text("OK")
|
|
|
|
app.add_route(handler, "/weight/<weight:number>")
|
|
|
|
request, response = app.test_client.get("/weight/12345")
|
|
assert response.text == "OK"
|
|
assert type(results[0]) is float
|
|
|
|
request, response = app.test_client.get("/weight/1234.56")
|
|
assert response.status == 200
|
|
|
|
request, response = app.test_client.get("/weight/.12")
|
|
assert response.status == 200
|
|
|
|
request, response = app.test_client.get("/weight/12.")
|
|
assert response.status == 200
|
|
|
|
request, response = app.test_client.get("/weight/1234-56")
|
|
assert response.status == 404
|
|
|
|
request, response = app.test_client.get("/weight/12.34.56")
|
|
assert response.status == 404
|
|
|
|
|
|
def test_dynamic_add_route_regex(app):
|
|
async def handler(request, folder_id):
|
|
return text("OK")
|
|
|
|
app.add_route(handler, "/folder/<folder_id:[A-Za-z0-9]{0,4}>")
|
|
|
|
request, response = app.test_client.get("/folder/test")
|
|
assert response.status == 200
|
|
|
|
request, response = app.test_client.get("/folder/test1")
|
|
assert response.status == 404
|
|
|
|
request, response = app.test_client.get("/folder/test-123")
|
|
assert response.status == 404
|
|
|
|
request, response = app.test_client.get("/folder/")
|
|
assert response.status == 200
|
|
|
|
|
|
def test_dynamic_add_route_unhashable(app):
|
|
async def handler(request, unhashable):
|
|
return text("OK")
|
|
|
|
app.add_route(handler, "/folder/<unhashable:[A-Za-z0-9/]+>/end/")
|
|
|
|
request, response = app.test_client.get("/folder/test/asdf/end/")
|
|
assert response.status == 200
|
|
|
|
request, response = app.test_client.get("/folder/test///////end/")
|
|
assert response.status == 200
|
|
|
|
request, response = app.test_client.get("/folder/test/end/")
|
|
assert response.status == 200
|
|
|
|
request, response = app.test_client.get("/folder/test/nope/")
|
|
assert response.status == 404
|
|
|
|
|
|
def test_add_route_duplicate(app):
|
|
|
|
with pytest.raises(RouteExists):
|
|
|
|
async def handler1(request):
|
|
pass
|
|
|
|
async def handler2(request):
|
|
pass
|
|
|
|
app.add_route(handler1, "/test")
|
|
app.add_route(handler2, "/test")
|
|
|
|
with pytest.raises(RouteExists):
|
|
|
|
async def handler1(request, dynamic):
|
|
pass
|
|
|
|
async def handler2(request, dynamic):
|
|
pass
|
|
|
|
app.add_route(handler1, "/test/<dynamic>/")
|
|
app.add_route(handler2, "/test/<dynamic>/")
|
|
|
|
|
|
def test_add_route_method_not_allowed(app):
|
|
async def handler(request):
|
|
return text("OK")
|
|
|
|
app.add_route(handler, "/test", methods=["GET"])
|
|
|
|
request, response = app.test_client.get("/test")
|
|
assert response.status == 200
|
|
|
|
request, response = app.test_client.post("/test")
|
|
assert response.status == 405
|
|
|
|
|
|
def test_removing_slash(app):
|
|
@app.get("/rest/<resource>")
|
|
def get(_):
|
|
pass
|
|
|
|
@app.post("/rest/<resource>")
|
|
def post(_):
|
|
pass
|
|
|
|
assert len(app.router.routes_all.keys()) == 2
|
|
|
|
|
|
def test_overload_routes(app):
|
|
@app.route("/overload", methods=["GET"])
|
|
async def handler1(request):
|
|
return text("OK1")
|
|
|
|
@app.route("/overload", methods=["POST", "PUT"])
|
|
async def handler2(request):
|
|
return text("OK2")
|
|
|
|
request, response = app.test_client.get("/overload")
|
|
assert response.text == "OK1"
|
|
|
|
request, response = app.test_client.post("/overload")
|
|
assert response.text == "OK2"
|
|
|
|
request, response = app.test_client.put("/overload")
|
|
assert response.text == "OK2"
|
|
|
|
request, response = app.test_client.delete("/overload")
|
|
assert response.status == 405
|
|
|
|
with pytest.raises(RouteExists):
|
|
|
|
@app.route("/overload", methods=["PUT", "DELETE"])
|
|
async def handler3(request):
|
|
return text("Duplicated")
|
|
|
|
|
|
def test_unmergeable_overload_routes(app):
|
|
@app.route("/overload_whole", methods=None)
|
|
async def handler1(request):
|
|
return text("OK1")
|
|
|
|
with pytest.raises(RouteExists):
|
|
|
|
@app.route("/overload_whole", methods=["POST", "PUT"])
|
|
async def handler2(request):
|
|
return text("Duplicated")
|
|
|
|
request, response = app.test_client.get("/overload_whole")
|
|
assert response.text == "OK1"
|
|
|
|
request, response = app.test_client.post("/overload_whole")
|
|
assert response.text == "OK1"
|
|
|
|
@app.route("/overload_part", methods=["GET"])
|
|
async def handler3(request):
|
|
return text("OK1")
|
|
|
|
with pytest.raises(RouteExists):
|
|
|
|
@app.route("/overload_part")
|
|
async def handler4(request):
|
|
return text("Duplicated")
|
|
|
|
request, response = app.test_client.get("/overload_part")
|
|
assert response.text == "OK1"
|
|
|
|
request, response = app.test_client.post("/overload_part")
|
|
assert response.status == 405
|
|
|
|
|
|
def test_unicode_routes(app):
|
|
@app.get("/你好")
|
|
def handler1(request):
|
|
return text("OK1")
|
|
|
|
request, response = app.test_client.get("/你好")
|
|
assert response.text == "OK1"
|
|
|
|
@app.route("/overload/<param>", methods=["GET"])
|
|
async def handler2(request, param):
|
|
return text("OK2 " + param)
|
|
|
|
request, response = app.test_client.get("/overload/你好")
|
|
assert response.text == "OK2 你好"
|
|
|
|
|
|
def test_uri_with_different_method_and_different_params(app):
|
|
@app.route("/ads/<ad_id>", methods=["GET"])
|
|
async def ad_get(request, ad_id):
|
|
return json({"ad_id": ad_id})
|
|
|
|
@app.route("/ads/<action>", methods=["POST"])
|
|
async def ad_post(request, action):
|
|
return json({"action": action})
|
|
|
|
request, response = app.test_client.get("/ads/1234")
|
|
assert response.status == 200
|
|
assert response.json == {"ad_id": "1234"}
|
|
|
|
request, response = app.test_client.post("/ads/post")
|
|
assert response.status == 200
|
|
assert response.json == {"action": "post"}
|
|
|
|
|
|
def test_route_raise_ParameterNameConflicts(app):
|
|
with pytest.raises(ParameterNameConflicts):
|
|
|
|
@app.get("/api/v1/<user>/<user>/")
|
|
def handler(request, user):
|
|
return text("OK")
|
|
|
|
|
|
def test_route_invalid_host(app):
|
|
|
|
host = 321
|
|
with pytest.raises(ValueError) as excinfo:
|
|
|
|
@app.get("/test", host=host)
|
|
def handler(request):
|
|
return text("pass")
|
|
|
|
assert str(excinfo.value) == (
|
|
"Expected either string or Iterable of " "host strings, not {!r}"
|
|
).format(host)
|