bc08383acd
* Remove unnecessary import in test_constants.py, which also fixes an error on win (#2180) Co-authored-by: Adam Hopkins <admhpkns@gmail.com> * Manually reset the buffer when streaming request body (#2183) * Remove Duplicated Dependencies and PEP 517 Support (#2173) * Remove duplicated dependencies * Specify setuptools as the tool for generating distribution (PEP 517) * Add `isort` to `dev_require` * manage all dependencies in setup.py * Execute `make pretty` * Set usedevelop to true (revert previous change) * Fix the handling of the end of a chunked request. (#2188) * Fix the handling of the end of a chunked request. * Avoid hardcoding final chunk header size. * Add some unit tests for pipeline body reading * Decode bytes for json serialization Co-authored-by: L. Kärkkäinen <tronic@users.noreply.github.com> Co-authored-by: Adam Hopkins <adam@amhopkins.com> * Resolve regressions in exceptions (#2181) * Update sanic-routing to fix path issues plus lookahead / lookbehind support (#2178) * Update sanic-routing to fix path issues plus lookahead / lookbehind support * Update setup.py Co-authored-by: Adam Hopkins <adam@amhopkins.com> Co-authored-by: Adam Hopkins <admhpkns@gmail.com> * style(app,blueprints): add some type hints (#2196) * style(app,blueprints): add some type hints * style(app): option is Any * style(blueprints): url prefix default value is ``""`` * style(app): backward compatible * style(app): backward compatible * style(blueprints): defult is None * style(app): apply code style (black) * Update some CC config (#2199) * Update README.rst * raise exception for `_static_request_handler` unknown exception; add test with custom error (#2195) Co-authored-by: n.feofanov <n.feofanov@visionlabs.ru> Co-authored-by: Adam Hopkins <admhpkns@gmail.com> * Change dumps to AnyStr (#2193) * HTTP tests (#2194) * Fix issues with after request handling in HTTP pipelining (#2201) * Clean up after a request is complete, before the next pipelined request. * Limit the size of request body consumed after handler has finished. * Linter error. * Add unit test re: bad headers Co-authored-by: L. Kärkkäinen <tronic@users.noreply.github.com> Co-authored-by: Adam Hopkins <admhpkns@gmail.com> Co-authored-by: Adam Hopkins <adam@amhopkins.com> * Update CHANGELOG * Log remote address if available (#2207) * Log remote address if available * Add tests * Fix testing version Co-authored-by: Adam Hopkins <adam@amhopkins.com> * Fixed for handling exceptions of asgi app call. (#2211) @cansarigol3megawatt Thanks for looking into this and getting the quick turnaround on this. I will 🍒 pick this into the 21.6 branch and get it out a little later tonight. * Signals Integration (#2160) * Update some tests * Resolve #2122 route decorator returning tuple * Use rc sanic-routing version * Update unit tests to <:str> * Minimal working version with some signals implemented * Add more http signals * Update ASGI and change listeners to signals * Allow for dynamic ODE signals * Allow signals to be stacked * Begin tests * Prioritize match_info on keyword argument injection * WIP on tests * Compat with signals * Work through some test coverage * Passing tests * Post linting * Setup proper resets * coverage reporting * Fixes from vltr comments * clear delayed tasks * Fix bad test * rm pycache * uncomment windows tests (#2214) * Add convenience methods to BP groups (#2209) * Fix bug where ws exceptions not being logged (#2213) * Fix bug where ws exceptions not being logged * Fix t\est * Style: add type hints (#2217) * style(routes): add_route argument, return typing * style(listeners): typing * style(views): typing as_view * style(routes): change type hint * style(listeners): change type hint * style(routes): change type hint * add some more types * Change as_view typing * Add some cleaner type annotations Co-authored-by: Adam Hopkins <adam@amhopkins.com> * Add default messages to SanicExceptions (#2216) * Add default messages to SanicExceptions * Cleaner exception message setting * Copy Blueprints Implementation (#2184) * Accept header parsing (#2200) * Add some tests * docstring * Add accept matching * Add some more tests on matching * Add matching flags for wildcards * Add mathing controls to accept * Limit uvicorn 14 in testing * Add convenience for annotated handlers (#2225) * Split HttpProtocol parts into base SanicProtocol and HTTPProtocol subclass (#2229) * Split HttpProtocol parts into base SanicProtocol and HTTPProtocol subclass. * lint fixes * re-black server.py * Move server.py into its own module (#2230) * Move server.py into its own module * Change monkeypatch path on test_logging.py * Blueprint specific exception handlers (#2208) * Call abort() on sockets after close() to prevent dangling sockets (#2231) * Add ability to return Falsey but not-None from handlers (#2236) * Adds Blueprint Group exception decorator (#2238) * Add exception decorator * Added tests * Fix line too long * Static DIR and FILE resource types (#2244) * Explicit static directive for serving file or dir Co-authored-by: anbuhckr <36891836+anbuhckr@users.noreply.github.com> Co-authored-by: anbuhckr <miki.suhendra@gmail.com> * Close HTTP loop when connection task cancelled (#2245) * Terminate loop when no transport exists * Add log when closing HTTP loop because of shutdown * Add unit test * New websockets (#2158) * First attempt at new Websockets implementation based on websockets >= 9.0, with sans-i/o features. Requires more work. * Update sanic/websocket.py Co-authored-by: Adam Hopkins <adam@amhopkins.com> * Update sanic/websocket.py Co-authored-by: Adam Hopkins <adam@amhopkins.com> * Update sanic/websocket.py Co-authored-by: Adam Hopkins <adam@amhopkins.com> * wip, update websockets code to new Sans/IO API * Refactored new websockets impl into own modules Incorporated other suggestions made by team * Another round of work on the new websockets impl * Added websocket_timeout support (matching previous/legacy support) * Lots more comments * Incorporated suggested changes from previous round of review * Changed RuntimeError usage to ServerError * Changed SanicException usage to ServerError * Removed some redundant asserts * Change remaining asserts to ServerErrors * Fixed some timeout handling issues * Fixed websocket.close() handling, and made it more robust * Made auto_close task smarter and more error-resilient * Made fail_connection routine smarter and more error-resilient * Further new websockets impl fixes * Update compatibility with Websockets v10 * Track server connection state in a more precise way * Try to handle the shutdown process more gracefully * Add a new end_connection() helper, to use as an alterative to close() or fail_connection() * Kill the auto-close task and keepalive-timeout task when sanic is shutdown * Deprecate WEBSOCKET_READ_LIMIT and WEBSOCKET_WRITE_LIMIT configs, they are not used in this implementation. * Change a warning message to debug level Remove default values for deprecated websocket parameters * Fix flake8 errors * Fix a couple of missed failing tests * remove websocket bench from examples * Integrate suggestions from code reviews Use Optional[T] instead of union[T,None] Fix mypy type logic errors change "is not None" to truthy checks where appropriate change "is None" to falsy checks were appropriate Add more debug logging when debug mode is on Change to using sanic.logger for debug logging rather than error_logger. * Fix long line lengths of debug messages Add some new debug messages when websocket IO is paused and unpaused for flow control Fix websocket example to use app.static() * remove unused import in websocket example app * re-run isort after Flake8 fixes Co-authored-by: Adam Hopkins <adam@amhopkins.com> Co-authored-by: Adam Hopkins <admhpkns@gmail.com> * Account for BP with exception handler but no routes (#2246) * Don't log "enabled" if auto-reload disabled (#2247) Fixes #2240 Co-authored-by: Adam Hopkins <admhpkns@gmail.com> * Smarter auto fallback (#2162) * Smarter auto fallback * remove config from blueprints * Add tests for error formatting * Add check for proper format * Fix some tests * Add some tests * docstring * Add accept matching * Add some more tests on matching * Fix contains bug, earlier return on MediaType eq * Add matching flags for wildcards * Add mathing controls to accept * Cleanup dev cruft * Add cleanup and resolve OSError relating to test implementation * Fix test * Fix some typos * Some fixes to the new Websockets impl (#2248) * First attempt at new Websockets implementation based on websockets >= 9.0, with sans-i/o features. Requires more work. * Update sanic/websocket.py Co-authored-by: Adam Hopkins <adam@amhopkins.com> * Update sanic/websocket.py Co-authored-by: Adam Hopkins <adam@amhopkins.com> * Update sanic/websocket.py Co-authored-by: Adam Hopkins <adam@amhopkins.com> * wip, update websockets code to new Sans/IO API * Refactored new websockets impl into own modules Incorporated other suggestions made by team * Another round of work on the new websockets impl * Added websocket_timeout support (matching previous/legacy support) * Lots more comments * Incorporated suggested changes from previous round of review * Changed RuntimeError usage to ServerError * Changed SanicException usage to ServerError * Removed some redundant asserts * Change remaining asserts to ServerErrors * Fixed some timeout handling issues * Fixed websocket.close() handling, and made it more robust * Made auto_close task smarter and more error-resilient * Made fail_connection routine smarter and more error-resilient * Further new websockets impl fixes * Update compatibility with Websockets v10 * Track server connection state in a more precise way * Try to handle the shutdown process more gracefully * Add a new end_connection() helper, to use as an alterative to close() or fail_connection() * Kill the auto-close task and keepalive-timeout task when sanic is shutdown * Deprecate WEBSOCKET_READ_LIMIT and WEBSOCKET_WRITE_LIMIT configs, they are not used in this implementation. * Change a warning message to debug level Remove default values for deprecated websocket parameters * Fix flake8 errors * Fix a couple of missed failing tests * remove websocket bench from examples * Integrate suggestions from code reviews Use Optional[T] instead of union[T,None] Fix mypy type logic errors change "is not None" to truthy checks where appropriate change "is None" to falsy checks were appropriate Add more debug logging when debug mode is on Change to using sanic.logger for debug logging rather than error_logger. * Fix long line lengths of debug messages Add some new debug messages when websocket IO is paused and unpaused for flow control Fix websocket example to use app.static() * remove unused import in websocket example app * re-run isort after Flake8 fixes * Some fixes to the new Websockets impl Will throw WebsocketClosed exception instead of ServerException now when attempting to read or write to closed websocket, this makes it easier to catch The various ws.recv() methods now have the ability to raise CancelledError into your websocket handler Fix a niche close-socket negotiation bug Fix bug where http protocol thought the websocket never sent any response. Allow data to still send in some cases after websocket enters CLOSING state. Fix some badly formatted and badly placed comments * allow eof_received to send back data too, if the connection is in CLOSING state Co-authored-by: Adam Hopkins <adam@amhopkins.com> Co-authored-by: Adam Hopkins <admhpkns@gmail.com> * 21.9 release docs (#2218) * Beging 21.9 release docs * Add PRs to changelog * Change deprecation version * Update logging tests * Bump version * Update changelog * Change dev install command (#2251) Co-authored-by: Zhiwei <zhi.wei.liang@outlook.com> Co-authored-by: L. Kärkkäinen <98187+Tronic@users.noreply.github.com> Co-authored-by: L. Kärkkäinen <tronic@users.noreply.github.com> Co-authored-by: Robert Palmer <robd003@users.noreply.github.com> Co-authored-by: Ryu JuHeon <saidbysolo@gmail.com> Co-authored-by: gluhar2006 <49654448+gluhar2006@users.noreply.github.com> Co-authored-by: n.feofanov <n.feofanov@visionlabs.ru> Co-authored-by: Néstor Pérez <25409753+prryplatypus@users.noreply.github.com> Co-authored-by: Can Sarigol <56863826+cansarigol3megawatt@users.noreply.github.com> Co-authored-by: Zhiwei <chihwei.public@outlook.com> Co-authored-by: YongChan Cho <h3236516@gmail.com> Co-authored-by: Zhiwei <zhiwei@sinatra.ai> Co-authored-by: Ashley Sommer <ashleysommer@gmail.com> Co-authored-by: anbuhckr <36891836+anbuhckr@users.noreply.github.com> Co-authored-by: anbuhckr <miki.suhendra@gmail.com>
1091 lines
28 KiB
Python
1091 lines
28 KiB
Python
import asyncio
|
|
import inspect
|
|
import os
|
|
|
|
import pytest
|
|
|
|
from sanic.app import Sanic
|
|
from sanic.blueprints import Blueprint
|
|
from sanic.constants import HTTP_METHODS
|
|
from sanic.exceptions import (
|
|
InvalidUsage,
|
|
NotFound,
|
|
SanicException,
|
|
ServerError,
|
|
)
|
|
from sanic.request import Request
|
|
from sanic.response import json, text
|
|
from sanic.views import CompositionView
|
|
|
|
|
|
# ------------------------------------------------------------ #
|
|
# GET
|
|
# ------------------------------------------------------------ #
|
|
|
|
|
|
def test_bp(app):
|
|
bp = Blueprint("test_text")
|
|
|
|
@bp.route("/")
|
|
def handler(request):
|
|
return text("Hello")
|
|
|
|
app.blueprint(bp)
|
|
request, response = app.test_client.get("/")
|
|
|
|
assert response.text == "Hello"
|
|
|
|
|
|
def test_bp_app_access(app):
|
|
bp = Blueprint("test")
|
|
|
|
with pytest.raises(
|
|
SanicException,
|
|
match="<Blueprint test> has not yet been registered to an app",
|
|
):
|
|
bp.apps
|
|
|
|
app.blueprint(bp)
|
|
|
|
assert app in bp.apps
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def static_file_directory():
|
|
"""The static directory to serve"""
|
|
current_file = inspect.getfile(inspect.currentframe())
|
|
current_directory = os.path.dirname(os.path.abspath(current_file))
|
|
static_directory = os.path.join(current_directory, "static")
|
|
return static_directory
|
|
|
|
|
|
def get_file_path(static_file_directory, file_name):
|
|
return os.path.join(static_file_directory, file_name)
|
|
|
|
|
|
def get_file_content(static_file_directory, file_name):
|
|
"""The content of the static file to check"""
|
|
with open(get_file_path(static_file_directory, file_name), "rb") as file:
|
|
return file.read()
|
|
|
|
|
|
@pytest.mark.parametrize("method", HTTP_METHODS)
|
|
def test_versioned_routes_get(app, method):
|
|
bp = Blueprint("test_text")
|
|
|
|
method = method.lower()
|
|
|
|
func = getattr(bp, method)
|
|
if callable(func):
|
|
|
|
@func(f"/{method}", version=1)
|
|
def handler(request):
|
|
return text("OK")
|
|
|
|
else:
|
|
raise Exception(f"{func} is not callable")
|
|
|
|
app.blueprint(bp)
|
|
|
|
client_method = getattr(app.test_client, method)
|
|
|
|
request, response = client_method(f"/v1/{method}")
|
|
assert response.status == 200
|
|
|
|
|
|
def test_bp_strict_slash(app):
|
|
bp = Blueprint("test_text")
|
|
|
|
@bp.get("/get", strict_slashes=True)
|
|
def get_handler(request):
|
|
return text("OK")
|
|
|
|
@bp.post("/post/", strict_slashes=True)
|
|
def post_handler(request):
|
|
return text("OK")
|
|
|
|
app.blueprint(bp)
|
|
|
|
request, response = app.test_client.get("/get")
|
|
assert response.text == "OK"
|
|
assert response.json is None
|
|
|
|
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_bp_strict_slash_default_value(app):
|
|
bp = Blueprint("test_text", strict_slashes=True)
|
|
|
|
@bp.get("/get")
|
|
def get_handler(request):
|
|
return text("OK")
|
|
|
|
@bp.post("/post/")
|
|
def post_handler(request):
|
|
return text("OK")
|
|
|
|
app.blueprint(bp)
|
|
|
|
request, response = app.test_client.get("/get/")
|
|
assert response.status == 404
|
|
|
|
request, response = app.test_client.post("/post")
|
|
assert response.status == 404
|
|
|
|
|
|
def test_bp_strict_slash_without_passing_default_value(app):
|
|
bp = Blueprint("test_text")
|
|
|
|
@bp.get("/get")
|
|
def get_handler(request):
|
|
return text("OK")
|
|
|
|
@bp.post("/post/")
|
|
def post_handler(request):
|
|
return text("OK")
|
|
|
|
app.blueprint(bp)
|
|
|
|
request, response = app.test_client.get("/get/")
|
|
assert response.text == "OK"
|
|
|
|
request, response = app.test_client.post("/post")
|
|
assert response.text == "OK"
|
|
|
|
|
|
def test_bp_strict_slash_default_value_can_be_overwritten(app):
|
|
bp = Blueprint("test_text", strict_slashes=True)
|
|
|
|
@bp.get("/get", strict_slashes=False)
|
|
def get_handler(request):
|
|
return text("OK")
|
|
|
|
@bp.post("/post/", strict_slashes=False)
|
|
def post_handler(request):
|
|
return text("OK")
|
|
|
|
app.blueprint(bp)
|
|
|
|
request, response = app.test_client.get("/get/")
|
|
assert response.text == "OK"
|
|
|
|
request, response = app.test_client.post("/post")
|
|
assert response.text == "OK"
|
|
|
|
|
|
def test_bp_with_url_prefix(app):
|
|
bp = Blueprint("test_text", url_prefix="/test1")
|
|
|
|
@bp.route("/")
|
|
def handler(request):
|
|
return text("Hello")
|
|
|
|
app.blueprint(bp)
|
|
request, response = app.test_client.get("/test1/")
|
|
|
|
assert response.text == "Hello"
|
|
|
|
|
|
def test_several_bp_with_url_prefix(app):
|
|
bp = Blueprint("test_text", url_prefix="/test1")
|
|
bp2 = Blueprint("test_text2", url_prefix="/test2")
|
|
|
|
@bp.route("/")
|
|
def handler(request):
|
|
return text("Hello")
|
|
|
|
@bp2.route("/")
|
|
def handler2(request):
|
|
return text("Hello2")
|
|
|
|
app.blueprint(bp)
|
|
app.blueprint(bp2)
|
|
request, response = app.test_client.get("/test1/")
|
|
assert response.text == "Hello"
|
|
|
|
request, response = app.test_client.get("/test2/")
|
|
assert response.text == "Hello2"
|
|
|
|
|
|
def test_bp_with_host(app):
|
|
bp = Blueprint("test_bp_host", url_prefix="/test1", host="example.com")
|
|
|
|
@bp.route("/")
|
|
def handler1(request):
|
|
return text("Hello")
|
|
|
|
@bp.route("/", host="sub.example.com")
|
|
def handler2(request):
|
|
return text("Hello subdomain!")
|
|
|
|
app.blueprint(bp)
|
|
headers = {"Host": "example.com"}
|
|
|
|
request, response = app.test_client.get("/test1/", headers=headers)
|
|
assert response.text == "Hello"
|
|
|
|
headers = {"Host": "sub.example.com"}
|
|
request, response = app.test_client.get("/test1/", headers=headers)
|
|
assert response.body == b"Hello subdomain!"
|
|
|
|
|
|
def test_several_bp_with_host(app):
|
|
bp = Blueprint(
|
|
"test_text",
|
|
url_prefix="/test",
|
|
host="example.com",
|
|
strict_slashes=True,
|
|
)
|
|
bp2 = Blueprint(
|
|
"test_text2",
|
|
url_prefix="/test",
|
|
host="sub.example.com",
|
|
strict_slashes=True,
|
|
)
|
|
|
|
@bp.route("/")
|
|
def handler(request):
|
|
return text("Hello")
|
|
|
|
@bp2.route("/")
|
|
def handler1(request):
|
|
return text("Hello2")
|
|
|
|
@bp2.route("/other/")
|
|
def handler2(request):
|
|
return text("Hello3")
|
|
|
|
app.blueprint(bp)
|
|
app.blueprint(bp2)
|
|
|
|
assert bp.host == "example.com"
|
|
headers = {"Host": "example.com"}
|
|
request, response = app.test_client.get("/test/", headers=headers)
|
|
|
|
assert response.text == "Hello"
|
|
|
|
assert bp2.host == "sub.example.com"
|
|
headers = {"Host": "sub.example.com"}
|
|
request, response = app.test_client.get("/test/", headers=headers)
|
|
|
|
assert response.text == "Hello2"
|
|
request, response = app.test_client.get("/test/other/", headers=headers)
|
|
assert response.text == "Hello3"
|
|
|
|
|
|
def test_bp_with_host_list(app):
|
|
bp = Blueprint(
|
|
"test_bp_host",
|
|
url_prefix="/test1",
|
|
host=["example.com", "sub.example.com"],
|
|
)
|
|
|
|
@bp.route("/")
|
|
def handler1(request):
|
|
return text("Hello")
|
|
|
|
@bp.route("/", host=["sub1.example.com"])
|
|
def handler2(request):
|
|
return text("Hello subdomain!")
|
|
|
|
app.blueprint(bp)
|
|
headers = {"Host": "example.com"}
|
|
request, response = app.test_client.get("/test1/", headers=headers)
|
|
assert response.text == "Hello"
|
|
|
|
headers = {"Host": "sub.example.com"}
|
|
request, response = app.test_client.get("/test1/", headers=headers)
|
|
assert response.text == "Hello"
|
|
|
|
headers = {"Host": "sub1.example.com"}
|
|
request, response = app.test_client.get("/test1/", headers=headers)
|
|
|
|
assert response.text == "Hello subdomain!"
|
|
|
|
|
|
def test_several_bp_with_host_list(app):
|
|
bp = Blueprint(
|
|
"test_text",
|
|
url_prefix="/test",
|
|
host=["example.com", "sub.example.com"],
|
|
)
|
|
bp2 = Blueprint(
|
|
"test_text2",
|
|
url_prefix="/test",
|
|
host=["sub1.example.com", "sub2.example.com"],
|
|
)
|
|
|
|
@bp.route("/")
|
|
def handler(request):
|
|
return text("Hello")
|
|
|
|
@bp2.route("/")
|
|
def handler1(request):
|
|
return text("Hello2")
|
|
|
|
@bp2.route("/other/")
|
|
def handler2(request):
|
|
return text("Hello3")
|
|
|
|
app.blueprint(bp)
|
|
app.blueprint(bp2)
|
|
|
|
assert bp.host == ["example.com", "sub.example.com"]
|
|
headers = {"Host": "example.com"}
|
|
request, response = app.test_client.get("/test/", headers=headers)
|
|
assert response.text == "Hello"
|
|
|
|
assert bp.host == ["example.com", "sub.example.com"]
|
|
headers = {"Host": "sub.example.com"}
|
|
request, response = app.test_client.get("/test/", headers=headers)
|
|
assert response.text == "Hello"
|
|
|
|
assert bp2.host == ["sub1.example.com", "sub2.example.com"]
|
|
headers = {"Host": "sub1.example.com"}
|
|
request, response = app.test_client.get("/test/", headers=headers)
|
|
assert response.text == "Hello2"
|
|
request, response = app.test_client.get("/test/other/", headers=headers)
|
|
assert response.text == "Hello3"
|
|
|
|
assert bp2.host == ["sub1.example.com", "sub2.example.com"]
|
|
headers = {"Host": "sub2.example.com"}
|
|
request, response = app.test_client.get("/test/", headers=headers)
|
|
assert response.text == "Hello2"
|
|
request, response = app.test_client.get("/test/other/", headers=headers)
|
|
assert response.text == "Hello3"
|
|
|
|
|
|
def test_bp_middleware(app):
|
|
blueprint = Blueprint("test_bp_middleware")
|
|
|
|
@blueprint.middleware("response")
|
|
async def process_response(request, response):
|
|
return text("OK")
|
|
|
|
@app.route("/")
|
|
async def handler(request):
|
|
return text("FAIL")
|
|
|
|
app.blueprint(blueprint)
|
|
|
|
request, response = app.test_client.get("/")
|
|
|
|
assert response.status == 200
|
|
assert response.text == "FAIL"
|
|
|
|
|
|
def test_bp_middleware_with_route(app):
|
|
blueprint = Blueprint("test_bp_middleware")
|
|
|
|
@blueprint.middleware("response")
|
|
async def process_response(request, response):
|
|
return text("OK")
|
|
|
|
@app.route("/")
|
|
async def handler(request):
|
|
return text("FAIL")
|
|
|
|
@blueprint.route("/bp")
|
|
async def bp_handler(request):
|
|
return text("FAIL")
|
|
|
|
app.blueprint(blueprint)
|
|
|
|
request, response = app.test_client.get("/bp")
|
|
|
|
assert response.status == 200
|
|
assert response.text == "OK"
|
|
|
|
|
|
def test_bp_middleware_order(app):
|
|
blueprint = Blueprint("test_bp_middleware_order")
|
|
order = []
|
|
|
|
@blueprint.middleware("request")
|
|
def mw_1(request):
|
|
order.append(1)
|
|
|
|
@blueprint.middleware("request")
|
|
def mw_2(request):
|
|
order.append(2)
|
|
|
|
@blueprint.middleware("request")
|
|
def mw_3(request):
|
|
order.append(3)
|
|
|
|
@blueprint.middleware("response")
|
|
def mw_4(request, response):
|
|
order.append(6)
|
|
|
|
@blueprint.middleware("response")
|
|
def mw_5(request, response):
|
|
order.append(5)
|
|
|
|
@blueprint.middleware("response")
|
|
def mw_6(request, response):
|
|
order.append(4)
|
|
|
|
@blueprint.route("/")
|
|
def process_response(request):
|
|
return text("OK")
|
|
|
|
app.blueprint(blueprint)
|
|
order.clear()
|
|
request, response = app.test_client.get("/")
|
|
|
|
assert response.status == 200
|
|
assert order == [1, 2, 3, 4, 5, 6]
|
|
|
|
|
|
def test_bp_exception_handler(app):
|
|
blueprint = Blueprint("test_middleware")
|
|
|
|
@blueprint.route("/1")
|
|
def handler_1(request):
|
|
raise InvalidUsage("OK")
|
|
|
|
@blueprint.route("/2")
|
|
def handler_2(request):
|
|
raise ServerError("OK")
|
|
|
|
@blueprint.route("/3")
|
|
def handler_3(request):
|
|
raise NotFound("OK")
|
|
|
|
@blueprint.exception(NotFound, ServerError)
|
|
def handler_exception(request, exception):
|
|
return text("OK")
|
|
|
|
app.blueprint(blueprint)
|
|
|
|
request, response = app.test_client.get("/1")
|
|
assert response.status == 400
|
|
|
|
request, response = app.test_client.get("/2")
|
|
assert response.status == 200
|
|
assert response.text == "OK"
|
|
|
|
request, response = app.test_client.get("/3")
|
|
assert response.status == 200
|
|
|
|
|
|
def test_bp_exception_handler_applied(app):
|
|
class Error(Exception):
|
|
pass
|
|
|
|
handled = Blueprint("handled")
|
|
nothandled = Blueprint("nothandled")
|
|
|
|
@handled.exception(Error)
|
|
def handle_error(req, e):
|
|
return text("handled {}".format(e))
|
|
|
|
@handled.route("/ok")
|
|
def ok(request):
|
|
raise Error("uh oh")
|
|
|
|
@nothandled.route("/notok")
|
|
def notok(request):
|
|
raise Error("uh oh")
|
|
|
|
app.blueprint(handled)
|
|
app.blueprint(nothandled)
|
|
|
|
_, response = app.test_client.get("/ok")
|
|
assert response.status == 200
|
|
assert response.text == "handled uh oh"
|
|
|
|
_, response = app.test_client.get("/notok")
|
|
assert response.status == 500
|
|
|
|
|
|
def test_bp_exception_handler_not_applied(app):
|
|
class Error(Exception):
|
|
pass
|
|
|
|
handled = Blueprint("handled")
|
|
nothandled = Blueprint("nothandled")
|
|
|
|
@handled.exception(Error)
|
|
def handle_error(req, e):
|
|
return text("handled {}".format(e))
|
|
|
|
@nothandled.route("/notok")
|
|
def notok(request):
|
|
raise Error("uh oh")
|
|
|
|
app.blueprint(handled)
|
|
app.blueprint(nothandled)
|
|
|
|
_, response = app.test_client.get("/notok")
|
|
assert response.status == 500
|
|
|
|
|
|
def test_bp_listeners(app):
|
|
app.route("/")(lambda x: x)
|
|
blueprint = Blueprint("test_middleware")
|
|
|
|
order = []
|
|
|
|
@blueprint.listener("before_server_start")
|
|
def handler_1(sanic, loop):
|
|
order.append(1)
|
|
|
|
@blueprint.listener("after_server_start")
|
|
def handler_2(sanic, loop):
|
|
order.append(2)
|
|
|
|
@blueprint.listener("after_server_start")
|
|
def handler_3(sanic, loop):
|
|
order.append(3)
|
|
|
|
@blueprint.listener("before_server_stop")
|
|
def handler_4(sanic, loop):
|
|
order.append(5)
|
|
|
|
@blueprint.listener("before_server_stop")
|
|
def handler_5(sanic, loop):
|
|
order.append(4)
|
|
|
|
@blueprint.listener("after_server_stop")
|
|
def handler_6(sanic, loop):
|
|
order.append(6)
|
|
|
|
app.blueprint(blueprint)
|
|
|
|
request, response = app.test_client.get("/")
|
|
|
|
assert order == [1, 2, 3, 4, 5, 6]
|
|
|
|
|
|
def test_bp_static(app):
|
|
current_file = inspect.getfile(inspect.currentframe())
|
|
with open(current_file, "rb") as file:
|
|
current_file_contents = file.read()
|
|
|
|
blueprint = Blueprint("test_static")
|
|
|
|
blueprint.static("/testing.file", current_file)
|
|
|
|
app.blueprint(blueprint)
|
|
|
|
request, response = app.test_client.get("/testing.file")
|
|
assert response.status == 200
|
|
assert response.body == current_file_contents
|
|
|
|
|
|
@pytest.mark.parametrize("file_name", ["test.html"])
|
|
def test_bp_static_content_type(app, file_name):
|
|
# This is done here, since no other test loads a file here
|
|
current_file = inspect.getfile(inspect.currentframe())
|
|
current_directory = os.path.dirname(os.path.abspath(current_file))
|
|
static_directory = os.path.join(current_directory, "static")
|
|
|
|
blueprint = Blueprint("test_static")
|
|
blueprint.static(
|
|
"/testing.file",
|
|
get_file_path(static_directory, file_name),
|
|
content_type="text/html; charset=utf-8",
|
|
)
|
|
|
|
app.blueprint(blueprint)
|
|
|
|
request, response = app.test_client.get("/testing.file")
|
|
assert response.status == 200
|
|
assert response.body == get_file_content(static_directory, file_name)
|
|
assert response.headers["Content-Type"] == "text/html; charset=utf-8"
|
|
|
|
|
|
def test_bp_shorthand(app):
|
|
blueprint = Blueprint("test_shorhand_routes")
|
|
ev = asyncio.Event()
|
|
|
|
@blueprint.get("/get")
|
|
def handler(request):
|
|
return text("OK")
|
|
|
|
@blueprint.put("/put")
|
|
def put_handler(request):
|
|
return text("OK")
|
|
|
|
@blueprint.post("/post")
|
|
def post_handler(request):
|
|
return text("OK")
|
|
|
|
@blueprint.head("/head")
|
|
def head_handler(request):
|
|
return text("OK")
|
|
|
|
@blueprint.options("/options")
|
|
def options_handler(request):
|
|
return text("OK")
|
|
|
|
@blueprint.patch("/patch")
|
|
def patch_handler(request):
|
|
return text("OK")
|
|
|
|
@blueprint.delete("/delete")
|
|
def delete_handler(request):
|
|
return text("OK")
|
|
|
|
@blueprint.websocket("/ws/", strict_slashes=True)
|
|
async def websocket_handler(request, ws):
|
|
ev.set()
|
|
|
|
app.blueprint(blueprint)
|
|
|
|
request, response = app.test_client.get("/get")
|
|
assert response.body == b"OK"
|
|
|
|
request, response = app.test_client.post("/get")
|
|
assert response.status == 405
|
|
|
|
request, response = app.test_client.put("/put")
|
|
assert response.body == b"OK"
|
|
|
|
request, response = app.test_client.get("/post")
|
|
assert response.status == 405
|
|
|
|
request, response = app.test_client.post("/post")
|
|
assert response.body == b"OK"
|
|
|
|
request, response = app.test_client.get("/post")
|
|
assert response.status == 405
|
|
|
|
request, response = app.test_client.head("/head")
|
|
assert response.status == 200
|
|
|
|
request, response = app.test_client.get("/head")
|
|
assert response.status == 405
|
|
|
|
request, response = app.test_client.options("/options")
|
|
assert response.body == b"OK"
|
|
|
|
request, response = app.test_client.get("/options")
|
|
assert response.status == 405
|
|
|
|
request, response = app.test_client.patch("/patch")
|
|
assert response.body == b"OK"
|
|
|
|
request, response = app.test_client.get("/patch")
|
|
assert response.status == 405
|
|
|
|
request, response = app.test_client.delete("/delete")
|
|
assert response.body == b"OK"
|
|
|
|
request, response = app.test_client.get("/delete")
|
|
assert response.status == 405
|
|
|
|
request, response = app.test_client.websocket("/ws/")
|
|
assert response.opened is True
|
|
assert ev.is_set()
|
|
|
|
|
|
def test_bp_group(app):
|
|
deep_0 = Blueprint("deep_0", url_prefix="/deep")
|
|
deep_1 = Blueprint("deep_1", url_prefix="/deep1")
|
|
|
|
@deep_0.route("/")
|
|
def handler(request):
|
|
return text("D0_OK")
|
|
|
|
@deep_1.route("/bottom")
|
|
def bottom_handler(request):
|
|
return text("D1B_OK")
|
|
|
|
mid_0 = Blueprint.group(deep_0, deep_1, url_prefix="/mid")
|
|
mid_1 = Blueprint("mid_tier", url_prefix="/mid1")
|
|
|
|
@mid_1.route("/")
|
|
def handler1(request):
|
|
return text("M1_OK")
|
|
|
|
top = Blueprint.group(mid_0, mid_1)
|
|
|
|
app.blueprint(top)
|
|
|
|
@app.route("/")
|
|
def handler2(request):
|
|
return text("TOP_OK")
|
|
|
|
request, response = app.test_client.get("/")
|
|
assert response.text == "TOP_OK"
|
|
|
|
request, response = app.test_client.get("/mid1")
|
|
assert response.text == "M1_OK"
|
|
|
|
request, response = app.test_client.get("/mid/deep")
|
|
assert response.text == "D0_OK"
|
|
|
|
request, response = app.test_client.get("/mid/deep1/bottom")
|
|
assert response.text == "D1B_OK"
|
|
|
|
|
|
def test_bp_group_with_default_url_prefix(app):
|
|
from sanic.response import json
|
|
|
|
bp_resources = Blueprint("bp_resources")
|
|
|
|
@bp_resources.get("/")
|
|
def list_resources_handler(request):
|
|
resource = {}
|
|
return json([resource])
|
|
|
|
bp_resource = Blueprint("bp_resource", url_prefix="/<resource_id>")
|
|
|
|
@bp_resource.get("/")
|
|
def get_resource_hander(request, resource_id):
|
|
resource = {"resource_id": resource_id}
|
|
return json(resource)
|
|
|
|
bp_resources_group = Blueprint.group(
|
|
bp_resources, bp_resource, url_prefix="/resources"
|
|
)
|
|
bp_api_v1 = Blueprint("bp_api_v1")
|
|
|
|
@bp_api_v1.get("/info")
|
|
def api_v1_info(request):
|
|
return text("api_version: v1")
|
|
|
|
bp_api_v1_group = Blueprint.group(
|
|
bp_api_v1, bp_resources_group, url_prefix="/v1"
|
|
)
|
|
bp_api_group = Blueprint.group(bp_api_v1_group, url_prefix="/api")
|
|
app.blueprint(bp_api_group)
|
|
|
|
request, response = app.test_client.get("/api/v1/info")
|
|
assert response.text == "api_version: v1"
|
|
|
|
request, response = app.test_client.get("/api/v1/resources")
|
|
assert response.json == [{}]
|
|
|
|
from uuid import uuid4
|
|
|
|
resource_id = str(uuid4())
|
|
request, response = app.test_client.get(f"/api/v1/resources/{resource_id}")
|
|
assert response.json == {"resource_id": resource_id}
|
|
|
|
|
|
def test_blueprint_middleware_with_args(app: Sanic):
|
|
bp = Blueprint(name="with_args_bp", url_prefix="/wa")
|
|
|
|
@bp.middleware
|
|
def middleware_with_no_tag(request: Request):
|
|
if request.headers.get("content-type") == "application/json":
|
|
request.headers["accepts"] = "plain/text"
|
|
else:
|
|
request.headers["accepts"] = "application/json"
|
|
|
|
@bp.route("/")
|
|
def default_route(request):
|
|
if request.headers.get("accepts") == "application/json":
|
|
return json({"test": "value"})
|
|
else:
|
|
return text("value")
|
|
|
|
app.blueprint(bp)
|
|
|
|
_, response = app.test_client.get(
|
|
"/wa", headers={"content-type": "application/json"}
|
|
)
|
|
assert response.text == "value"
|
|
|
|
_, response = app.test_client.get(
|
|
"/wa", headers={"content-type": "plain/text"}
|
|
)
|
|
assert response.json.get("test") == "value"
|
|
|
|
|
|
@pytest.mark.parametrize("file_name", ["test.file"])
|
|
def test_static_blueprint_name(static_file_directory, file_name):
|
|
app = Sanic("app")
|
|
current_file = inspect.getfile(inspect.currentframe())
|
|
with open(current_file, "rb") as file:
|
|
file.read()
|
|
|
|
bp = Blueprint(name="static", url_prefix="/static", strict_slashes=False)
|
|
|
|
bp.static(
|
|
"/test.file/",
|
|
get_file_path(static_file_directory, file_name),
|
|
name="static.testing",
|
|
strict_slashes=True,
|
|
)
|
|
|
|
app.blueprint(bp)
|
|
|
|
uri = app.url_for("static", name="static.testing")
|
|
assert uri == "/static/test.file/"
|
|
|
|
_, response = app.test_client.get("/static/test.file")
|
|
assert response.status == 404
|
|
|
|
_, response = app.test_client.get("/static/test.file/")
|
|
assert response.status == 200
|
|
|
|
|
|
@pytest.mark.parametrize("file_name", ["test.file"])
|
|
def test_static_blueprintp_mw(app: Sanic, static_file_directory, file_name):
|
|
current_file = inspect.getfile(inspect.currentframe())
|
|
with open(current_file, "rb") as file:
|
|
file.read()
|
|
|
|
triggered = False
|
|
|
|
bp = Blueprint(name="test_mw", url_prefix="")
|
|
|
|
@bp.middleware("request")
|
|
def bp_mw1(request):
|
|
nonlocal triggered
|
|
triggered = True
|
|
|
|
bp.static(
|
|
"/test.file",
|
|
get_file_path(static_file_directory, file_name),
|
|
strict_slashes=True,
|
|
name="static",
|
|
)
|
|
|
|
app.blueprint(bp)
|
|
|
|
uri = app.url_for("test_mw.static")
|
|
assert uri == "/test.file"
|
|
|
|
_, response = app.test_client.get("/test.file")
|
|
assert triggered is True
|
|
|
|
|
|
def test_route_handler_add(app: Sanic):
|
|
view = CompositionView()
|
|
|
|
async def get_handler(request):
|
|
return json({"response": "OK"})
|
|
|
|
view.add(["GET"], get_handler, stream=False)
|
|
|
|
async def default_handler(request):
|
|
return text("OK")
|
|
|
|
bp = Blueprint(name="handler", url_prefix="/handler")
|
|
bp.add_route(default_handler, uri="/default/", strict_slashes=True)
|
|
|
|
bp.add_route(view, uri="/view", name="test")
|
|
|
|
app.blueprint(bp)
|
|
|
|
_, response = app.test_client.get("/handler/default/")
|
|
assert response.text == "OK"
|
|
|
|
_, response = app.test_client.get("/handler/view")
|
|
assert response.json["response"] == "OK"
|
|
|
|
|
|
def test_websocket_route(app: Sanic):
|
|
event = asyncio.Event()
|
|
|
|
async def websocket_handler(request, ws):
|
|
assert ws.subprotocol is None
|
|
event.set()
|
|
|
|
bp = Blueprint(name="handler", url_prefix="/ws")
|
|
bp.add_websocket_route(websocket_handler, "/test", name="test")
|
|
|
|
app.blueprint(bp)
|
|
|
|
_, response = app.test_client.websocket("/ws/test")
|
|
assert response.opened is True
|
|
assert event.is_set()
|
|
|
|
|
|
def test_duplicate_blueprint(app):
|
|
bp_name = "bp"
|
|
bp = Blueprint(bp_name)
|
|
bp1 = Blueprint(bp_name)
|
|
|
|
app.blueprint(bp)
|
|
|
|
with pytest.raises(AssertionError) as excinfo:
|
|
app.blueprint(bp1)
|
|
|
|
assert str(excinfo.value) == (
|
|
f'A blueprint with the name "{bp_name}" is already registered. '
|
|
"Blueprint names must be unique."
|
|
)
|
|
|
|
|
|
def test_strict_slashes_behavior_adoption():
|
|
app = Sanic("app")
|
|
app.strict_slashes = True
|
|
bp = Blueprint("bp")
|
|
bp2 = Blueprint("bp2", strict_slashes=False)
|
|
|
|
@app.get("/test")
|
|
def handler_test(request):
|
|
return text("Test")
|
|
|
|
@app.get("/f1", strict_slashes=False)
|
|
def f1(request):
|
|
return text("f1")
|
|
|
|
@bp.get("/one", strict_slashes=False)
|
|
def one(request):
|
|
return text("one")
|
|
|
|
@bp.get("/second")
|
|
def second(request):
|
|
return text("second")
|
|
|
|
@bp2.get("/third")
|
|
def third(request):
|
|
return text("third")
|
|
|
|
app.blueprint(bp)
|
|
app.blueprint(bp2)
|
|
|
|
assert app.test_client.get("/test")[1].status == 200
|
|
assert app.test_client.get("/test/")[1].status == 404
|
|
|
|
assert app.test_client.get("/one")[1].status == 200
|
|
assert app.test_client.get("/one/")[1].status == 200
|
|
|
|
assert app.test_client.get("/second")[1].status == 200
|
|
assert app.test_client.get("/second/")[1].status == 404
|
|
|
|
assert app.test_client.get("/third")[1].status == 200
|
|
assert app.test_client.get("/third/")[1].status == 200
|
|
|
|
assert app.test_client.get("/f1")[1].status == 200
|
|
assert app.test_client.get("/f1/")[1].status == 200
|
|
|
|
|
|
def test_blueprint_group_versioning():
|
|
app = Sanic(name="blueprint-group-test")
|
|
|
|
bp1 = Blueprint(name="bp1", url_prefix="/bp1")
|
|
bp2 = Blueprint(name="bp2", url_prefix="/bp2", version=2)
|
|
|
|
bp3 = Blueprint(name="bp3", url_prefix="/bp3")
|
|
|
|
@bp3.get("/r1")
|
|
async def bp3_r1(request):
|
|
return json({"from": "bp3/r1"})
|
|
|
|
@bp1.get("/pre-group")
|
|
async def pre_group(request):
|
|
return json({"from": "bp1/pre-group"})
|
|
|
|
group = Blueprint.group([bp1, bp2], url_prefix="/group1", version=1)
|
|
|
|
group2 = Blueprint.group([bp3])
|
|
|
|
@bp1.get("/r1")
|
|
async def r1(request):
|
|
return json({"from": "bp1/r1"})
|
|
|
|
@bp2.get("/r2")
|
|
async def r2(request):
|
|
return json({"from": "bp2/r2"})
|
|
|
|
@bp2.get("/r3", version=3)
|
|
async def r3(request):
|
|
return json({"from": "bp2/r3"})
|
|
|
|
app.blueprint([group, group2])
|
|
|
|
assert app.test_client.get("/v1/group1/bp1/r1/")[1].status == 200
|
|
assert app.test_client.get("/v2/group1/bp2/r2")[1].status == 200
|
|
assert app.test_client.get("/v1/group1/bp1/pre-group")[1].status == 200
|
|
assert app.test_client.get("/v3/group1/bp2/r3")[1].status == 200
|
|
assert app.test_client.get("/bp3/r1")[1].status == 200
|
|
|
|
assert group.version == 1
|
|
assert group2.strict_slashes is None
|
|
|
|
|
|
def test_blueprint_group_strict_slashes():
|
|
app = Sanic(name="blueprint-group-test")
|
|
bp1 = Blueprint(name="bp1", url_prefix=None, strict_slashes=False)
|
|
|
|
bp2 = Blueprint(
|
|
name="bp2", version=3, url_prefix="/bp2", strict_slashes=None
|
|
)
|
|
|
|
bp3 = Blueprint(
|
|
name="bp3", version=None, url_prefix="/bp3/", strict_slashes=None
|
|
)
|
|
|
|
@bp1.get("/r1")
|
|
async def bp1_r1(request):
|
|
return json({"from": "bp1/r1"})
|
|
|
|
@bp2.get("/r1")
|
|
async def bp2_r1(request):
|
|
return json({"from": "bp2/r1"})
|
|
|
|
@bp2.get("/r2/")
|
|
async def bp2_r2(request):
|
|
return json({"from": "bp2/r2"})
|
|
|
|
@bp3.get("/r1")
|
|
async def bp3_r1(request):
|
|
return json({"from": "bp3/r1"})
|
|
|
|
group = Blueprint.group(
|
|
[bp1, bp2],
|
|
url_prefix="/slash-check/",
|
|
version=1.3,
|
|
strict_slashes=True,
|
|
)
|
|
|
|
group2 = Blueprint.group(
|
|
[bp3], url_prefix="/other-prefix/", version="v2", strict_slashes=False
|
|
)
|
|
|
|
app.blueprint(group)
|
|
app.blueprint(group2)
|
|
|
|
assert app.test_client.get("/v1.3/slash-check/r1")[1].status == 200
|
|
assert app.test_client.get("/v1.3/slash-check/r1/")[1].status == 200
|
|
assert app.test_client.get("/v3/slash-check/bp2/r1")[1].status == 200
|
|
assert app.test_client.get("/v3/slash-check/bp2/r1/")[1].status == 404
|
|
assert app.test_client.get("/v3/slash-check/bp2/r2")[1].status == 404
|
|
assert app.test_client.get("/v3/slash-check/bp2/r2/")[1].status == 200
|
|
assert app.test_client.get("/v2/other-prefix/bp3/r1")[1].status == 200
|
|
|
|
|
|
def test_blueprint_registered_multiple_apps():
|
|
app1 = Sanic("app1")
|
|
app2 = Sanic("app2")
|
|
bp = Blueprint("bp")
|
|
|
|
@bp.get("/")
|
|
async def handler(request):
|
|
return text(request.route.name)
|
|
|
|
app1.blueprint(bp)
|
|
app2.blueprint(bp)
|
|
|
|
for app in (app1, app2):
|
|
_, response = app.test_client.get("/")
|
|
assert response.text == f"{app.name}.bp.handler"
|
|
|
|
|
|
def test_bp_set_attribute_warning():
|
|
bp = Blueprint("bp")
|
|
with pytest.warns(DeprecationWarning) as record:
|
|
bp.foo = 1
|
|
|
|
assert len(record) == 1
|
|
assert record[0].message.args[0] == (
|
|
"Setting variables on Blueprint instances is deprecated "
|
|
"and will be removed in version 21.12. You should change your "
|
|
"Blueprint instance to use instance.ctx.foo instead."
|
|
)
|