Add some test coverage
This commit is contained in:
@@ -110,6 +110,11 @@ def test_bp_group(app: Sanic):
|
||||
global MIDDLEWARE_INVOKE_COUNTER
|
||||
MIDDLEWARE_INVOKE_COUNTER["request"] += 1
|
||||
|
||||
@blueprint_group_1.middleware
|
||||
def blueprint_group_1_middleware_not_called(request):
|
||||
global MIDDLEWARE_INVOKE_COUNTER
|
||||
MIDDLEWARE_INVOKE_COUNTER["request"] += 1
|
||||
|
||||
@blueprint_3.route("/")
|
||||
def blueprint_3_default_route(request):
|
||||
return text("BP3_OK")
|
||||
@@ -142,7 +147,7 @@ def test_bp_group(app: Sanic):
|
||||
assert response.text == "BP3_OK"
|
||||
|
||||
assert MIDDLEWARE_INVOKE_COUNTER["response"] == 3
|
||||
assert MIDDLEWARE_INVOKE_COUNTER["request"] == 2
|
||||
assert MIDDLEWARE_INVOKE_COUNTER["request"] == 4
|
||||
|
||||
|
||||
def test_bp_group_list_operations(app: Sanic):
|
||||
@@ -179,3 +184,19 @@ def test_bp_group_list_operations(app: Sanic):
|
||||
assert len(blueprint_group_1) == 2
|
||||
|
||||
assert blueprint_group_1.url_prefix == "/bp"
|
||||
|
||||
|
||||
def test_bp_group_as_list():
|
||||
blueprint_1 = Blueprint("blueprint_1", url_prefix="/bp1")
|
||||
blueprint_2 = Blueprint("blueprint_2", url_prefix="/bp2")
|
||||
blueprint_group_1 = Blueprint.group([blueprint_1, blueprint_2])
|
||||
assert len(blueprint_group_1) == 2
|
||||
|
||||
|
||||
def test_bp_group_as_nested_group():
|
||||
blueprint_1 = Blueprint("blueprint_1", url_prefix="/bp1")
|
||||
blueprint_2 = Blueprint("blueprint_2", url_prefix="/bp2")
|
||||
blueprint_group_1 = Blueprint.group(
|
||||
Blueprint.group(blueprint_1, blueprint_2)
|
||||
)
|
||||
assert len(blueprint_group_1) == 2
|
||||
|
||||
@@ -30,6 +30,23 @@ def test_middleware_request(app):
|
||||
assert type(results[0]) is Request
|
||||
|
||||
|
||||
def test_middleware_request_as_convenience(app):
|
||||
results = []
|
||||
|
||||
@app.on_request
|
||||
async def handler1(request):
|
||||
results.append(request)
|
||||
|
||||
@app.route("/")
|
||||
async def handler2(request):
|
||||
return text("OK")
|
||||
|
||||
request, response = app.test_client.get("/")
|
||||
|
||||
assert response.text == "OK"
|
||||
assert type(results[0]) is Request
|
||||
|
||||
|
||||
def test_middleware_response(app):
|
||||
results = []
|
||||
|
||||
@@ -54,6 +71,54 @@ def test_middleware_response(app):
|
||||
assert isinstance(results[2], HTTPResponse)
|
||||
|
||||
|
||||
def test_middleware_response_as_convenience(app):
|
||||
results = []
|
||||
|
||||
@app.on_request
|
||||
async def process_request(request):
|
||||
results.append(request)
|
||||
|
||||
@app.on_response
|
||||
async def process_response(request, response):
|
||||
results.append(request)
|
||||
results.append(response)
|
||||
|
||||
@app.route("/")
|
||||
async def handler(request):
|
||||
return text("OK")
|
||||
|
||||
request, response = app.test_client.get("/")
|
||||
|
||||
assert response.text == "OK"
|
||||
assert type(results[0]) is Request
|
||||
assert type(results[1]) is Request
|
||||
assert isinstance(results[2], HTTPResponse)
|
||||
|
||||
|
||||
def test_middleware_response_as_convenience_called(app):
|
||||
results = []
|
||||
|
||||
@app.on_request()
|
||||
async def process_request(request):
|
||||
results.append(request)
|
||||
|
||||
@app.on_response()
|
||||
async def process_response(request, response):
|
||||
results.append(request)
|
||||
results.append(response)
|
||||
|
||||
@app.route("/")
|
||||
async def handler(request):
|
||||
return text("OK")
|
||||
|
||||
request, response = app.test_client.get("/")
|
||||
|
||||
assert response.text == "OK"
|
||||
assert type(results[0]) is Request
|
||||
assert type(results[1]) is Request
|
||||
assert isinstance(results[2], HTTPResponse)
|
||||
|
||||
|
||||
def test_middleware_response_exception(app):
|
||||
result = {"status_code": "middleware not run"}
|
||||
|
||||
|
||||
@@ -633,6 +633,19 @@ def test_websocket_route(app, url):
|
||||
assert ev.is_set()
|
||||
|
||||
|
||||
def test_websocket_route_invalid_handler(app):
|
||||
with pytest.raises(ValueError) as e:
|
||||
|
||||
@app.websocket("/")
|
||||
async def handler():
|
||||
...
|
||||
|
||||
assert e.match(
|
||||
r"Required parameter `request` and/or `ws` missing in the "
|
||||
r"handler\(\) route\?"
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize("url", ["/ws", "ws"])
|
||||
async def test_websocket_route_asgi(app, url):
|
||||
|
||||
@@ -8,6 +8,8 @@ import pytest
|
||||
|
||||
from sanic_testing.testing import HOST, PORT
|
||||
|
||||
from sanic.exceptions import InvalidUsage
|
||||
|
||||
|
||||
AVAILABLE_LISTENERS = [
|
||||
"before_server_start",
|
||||
@@ -80,6 +82,18 @@ def test_all_listeners(app):
|
||||
assert app.name + listener_name == output.pop()
|
||||
|
||||
|
||||
@skipif_no_alarm
|
||||
def test_all_listeners_as_convenience(app):
|
||||
output = []
|
||||
for listener_name in AVAILABLE_LISTENERS:
|
||||
listener = create_listener(listener_name, output)
|
||||
method = getattr(app, listener_name)
|
||||
method(listener)
|
||||
start_stop_app(app)
|
||||
for listener_name in AVAILABLE_LISTENERS:
|
||||
assert app.name + listener_name == output.pop()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_trigger_before_events_create_server(app):
|
||||
class MySanicDb:
|
||||
@@ -95,6 +109,20 @@ async def test_trigger_before_events_create_server(app):
|
||||
assert isinstance(app.db, MySanicDb)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_trigger_before_events_create_server_missing_event(app):
|
||||
class MySanicDb:
|
||||
pass
|
||||
|
||||
with pytest.raises(InvalidUsage):
|
||||
|
||||
@app.listener
|
||||
async def init_db(app, loop):
|
||||
app.db = MySanicDb()
|
||||
|
||||
assert not hasattr(app, "db")
|
||||
|
||||
|
||||
def test_create_server_trigger_events(app):
|
||||
"""Test if create_server can trigger server events"""
|
||||
|
||||
|
||||
@@ -127,6 +127,40 @@ def test_static_file_content_type(app, static_file_directory, file_name):
|
||||
assert response.headers["Content-Type"] == "text/html; charset=utf-8"
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"file_name,expected",
|
||||
[
|
||||
("test.html", "text/html; charset=utf-8"),
|
||||
("decode me.txt", "text/plain; charset=utf-8"),
|
||||
("test.file", "application/octet-stream"),
|
||||
],
|
||||
)
|
||||
def test_static_file_content_type_guessed(
|
||||
app, static_file_directory, file_name, expected
|
||||
):
|
||||
app.static(
|
||||
"/testing.file",
|
||||
get_file_path(static_file_directory, file_name),
|
||||
)
|
||||
|
||||
request, response = app.test_client.get("/testing.file")
|
||||
assert response.status == 200
|
||||
assert response.body == get_file_content(static_file_directory, file_name)
|
||||
assert response.headers["Content-Type"] == expected
|
||||
|
||||
|
||||
def test_static_file_content_type_with_charset(app, static_file_directory):
|
||||
app.static(
|
||||
"/testing.file",
|
||||
get_file_path(static_file_directory, "decode me.txt"),
|
||||
content_type="text/plain;charset=ISO-8859-1",
|
||||
)
|
||||
|
||||
request, response = app.test_client.get("/testing.file")
|
||||
assert response.status == 200
|
||||
assert response.headers["Content-Type"] == "text/plain;charset=ISO-8859-1"
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"file_name", ["test.file", "decode me.txt", "symlink", "hard_link"]
|
||||
)
|
||||
|
||||
Reference in New Issue
Block a user