Vasanko/sanic
2
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity
7674e917e4
sanic/tests/test_requests.py
L. Kärkkäinen 1e4b1c4d1a Forwarded headers and otherwise improved proxy handling (#1638) 
* Added support for HTTP Forwarded header and combined parsing of other proxy headers.

- Accessible via request.forwarded that tries parse_forwarded and then parse_xforwarded
- parse_forwarded uses the Forwarded header, if config.FORWARDED_SECRET is provided and a matching header field is found
- parse_xforwarded uses X-Real-IP and X-Forwarded-* much alike the existing implementation
- This commit does not change existing request properties that still use the old code and won't make use of Forwarded headers.

* Use req.forwarded in req properties server_name, server_port, scheme and remote_addr.

X-Scheme handling moved to parse_xforwarded.

* Cleanup and fix req.server_port; no longer reports socket port if any forwards headers are used.

* Update docstrings to incidate that forwarded header is used first.

* Remove testing function.

* Fix tests and linting.

- One test removed due to change of semantics - no socket port will be used if any forwarded headers are in effect.
- Other tests augmented with X-Forwarded-For, to allow the header being tested take effect (shouldn't affect old implementation).

* Try to workaround buggy tools complaining about incorrect ordering of imports.

* Cleanup forwarded processing, add comments. secret is now also returned.

* Added tests, fixed quoted string handling, cleanup.

* Further tests for full coverage.

* Try'n make linter happy.

* Add support for multiple Forwarded headers. Unify parse_forwarded parameters with parse_xforwarded.

* Implement multiple headers support for X-Forwarded-For.

- Previously only the first header was used, so this BUGFIX may affect functionality.

* Bugfix for request.server_name: strip port and other parts.

- request.server_name docs claim that it returns the hostname only (no port).
- config.SERVER_NAME may be full URL, so strip scheme, port and path
- HTTP Host and consequently forwarded Host may include port number, so
  strip that also for forwarded hosts (previously only done for HTTP Host).
- Possible performance benefit of limiting to one split.

* Fallback to app.url_for and let it handle SERVER_NAME if defined (until a proper solution is implemented).

* Revise previous commit. Only fallback for full URL SERVER_NAMEs; allows host to be defined and proxied information still being used.

* Heil lintnazi.

* Modify testcase not to use underscores in URLs. Use hyphens which the spec allows for.

* Forwarded and Host header parsing improved.

- request.forwarded lowercases hosts, separates host:port into their own fields and lowercases addresses
- forwarded.parse_host helper function added and used for parsing all host-style headers (IPv6 cannot be simply split(":")).
- more tests fixed not to use underscores in hosts as those are no longer accepted and lead to the field being rejected

* Fixed typo in docstring.

* Added IPv6 address tests for Host header.

* Fix regex.

* Further tests and stricter forwarded handling.

* Fix merge commit

* Linter

* Linter

* Linter

* Add  to avoid re-using the  variable. Make a few raw strings non-raw.

* Remove unnecessary or

* Updated docs (work in progress).

* Enable REAL_IP_HEADER parsing irregardless of PROXIES_COUNT setting.

- Also cleanup and added comments

* New defaults for PROXIES_COUNT and REAL_IP_HEADER, updated tests.

* Remove support for PROXIES_COUNT=-1.

* Linter errors.

- This is getting ridiculous: cannot fit an URL on one line, linter requires
  splitting the string literal!

* Add support for by=_proxySecret, updated docs, updated tests.

* Forwarded headers' semantics tuning.

- Forwarded host is now preserved in original format
- request.host now returns a forwarded host if available, else the Host header
- Forwarded options are preserved in original order, and later keys override earlier ones
- Forwarded path is automatically URL-unquoted
- Forwarded 'by' and 'for' are omitted if their value is unknown
- Tests modified accordingly
- Cleanup and improved documentation

* Add ASGI test.

* Linter

* Linter #2
2019-09-02 08:50:56 -05:00

2082 lines
59 KiB
Python
Raw Blame History

import logging
import os
import ssl
from json import dumps as json_dumps
from json import loads as json_loads
from urllib.parse import urlparse
import pytest
from sanic import Blueprint, Sanic
from sanic.exceptions import ServerError
from sanic.request import DEFAULT_HTTP_CONTENT_TYPE, RequestParameters
from sanic.response import json, text
from sanic.testing import ASGI_HOST, HOST, PORT
# ------------------------------------------------------------ #
# GET
# ------------------------------------------------------------ #
def test_sync(app):
@app.route("/")
def handler(request):
return text("Hello")
request, response = app.test_client.get("/")
assert response.text == "Hello"
@pytest.mark.asyncio
async def test_sync_asgi(app):
@app.route("/")
def handler(request):
return text("Hello")
request, response = await app.asgi_client.get("/")
assert response.text == "Hello"
def test_ip(app):
@app.route("/")
def handler(request):
return text("{}".format(request.ip))
request, response = app.test_client.get("/")
assert response.text == "127.0.0.1"
@pytest.mark.asyncio
async def test_ip_asgi(app):
@app.route("/")
def handler(request):
return text("{}".format(request.ip))
request, response = await app.asgi_client.get("/")
assert response.text == "mockserver"
def test_text(app):
@app.route("/")
async def handler(request):
return text("Hello")
request, response = app.test_client.get("/")
assert response.text == "Hello"
@pytest.mark.asyncio
async def test_text_asgi(app):
@app.route("/")
async def handler(request):
return text("Hello")
request, response = await app.asgi_client.get("/")
assert response.text == "Hello"
def test_headers(app):
@app.route("/")
async def handler(request):
headers = {"spam": "great"}
return text("Hello", headers=headers)
request, response = app.test_client.get("/")
assert response.headers.get("spam") == "great"
@pytest.mark.asyncio
async def test_headers_asgi(app):
@app.route("/")
async def handler(request):
headers = {"spam": "great"}
return text("Hello", headers=headers)
request, response = await app.asgi_client.get("/")
assert response.headers.get("spam") == "great"
def test_non_str_headers(app):
@app.route("/")
async def handler(request):
headers = {"answer": 42}
return text("Hello", headers=headers)
request, response = app.test_client.get("/")
assert response.headers.get("answer") == "42"
@pytest.mark.asyncio
async def test_non_str_headers_asgi(app):
@app.route("/")
async def handler(request):
headers = {"answer": 42}
return text("Hello", headers=headers)
request, response = await app.asgi_client.get("/")
assert response.headers.get("answer") == "42"
def test_invalid_response(app):
@app.exception(ServerError)
def handler_exception(request, exception):
return text("Internal Server Error.", 500)
@app.route("/")
async def handler(request):
return "This should fail"
request, response = app.test_client.get("/")
assert response.status == 500
assert response.text == "Internal Server Error."
@pytest.mark.asyncio
async def test_invalid_response_asgi(app):
@app.exception(ServerError)
def handler_exception(request, exception):
return text("Internal Server Error.", 500)
@app.route("/")
async def handler(request):
return "This should fail"
request, response = await app.asgi_client.get("/")
assert response.status == 500
assert response.text == "Internal Server Error."
def test_json(app):
@app.route("/")
async def handler(request):
return json({"test": True})
request, response = app.test_client.get("/")
results = json_loads(response.text)
assert results.get("test") is True
@pytest.mark.asyncio
async def test_json_asgi(app):
@app.route("/")
async def handler(request):
return json({"test": True})
request, response = await app.asgi_client.get("/")
results = json_loads(response.text)
assert results.get("test") is True
def test_empty_json(app):
@app.route("/")
async def handler(request):
assert request.json is None
return json(request.json)
request, response = app.test_client.get("/")
assert response.status == 200
assert response.text == "null"
@pytest.mark.asyncio
async def test_empty_json_asgi(app):
@app.route("/")
async def handler(request):
assert request.json is None
return json(request.json)
request, response = await app.asgi_client.get("/")
assert response.status == 200
assert response.text == "null"
def test_invalid_json(app):
@app.route("/")
async def handler(request):
return json(request.json)
data = "I am not json"
request, response = app.test_client.get("/", data=data)
assert response.status == 400
@pytest.mark.asyncio
async def test_invalid_json_asgi(app):
@app.route("/")
async def handler(request):
return json(request.json)
data = "I am not json"
request, response = await app.asgi_client.get("/", data=data)
assert response.status == 400
def test_query_string(app):
@app.route("/")
async def handler(request):
return text("OK")
request, response = app.test_client.get(
"/", params=[("test1", "1"), ("test2", "false"), ("test2", "true")]
)
assert request.args.get("test1") == "1"
assert request.args.get("test2") == "false"
assert request.args.getlist("test2") == ["false", "true"]
assert request.args.getlist("test1") == ["1"]
assert request.args.get("test3", default="My value") == "My value"
@pytest.mark.asyncio
async def test_query_string_asgi(app):
@app.route("/")
async def handler(request):
return text("OK")
request, response = await app.asgi_client.get(
"/", params=[("test1", "1"), ("test2", "false"), ("test2", "true")]
)
assert request.args.get("test1") == "1"
assert request.args.get("test2") == "false"
assert request.args.getlist("test2") == ["false", "true"]
assert request.args.getlist("test1") == ["1"]
assert request.args.get("test3", default="My value") == "My value"
def test_uri_template(app):
@app.route("/foo/<id:int>/bar/<name:[A-z]+>")
async def handler(request, id, name):
return text("OK")
request, response = app.test_client.get("/foo/123/bar/baz")
assert request.uri_template == "/foo/<id:int>/bar/<name:[A-z]+>"
@pytest.mark.asyncio
async def test_uri_template_asgi(app):
@app.route("/foo/<id:int>/bar/<name:[A-z]+>")
async def handler(request, id, name):
return text("OK")
request, response = await app.asgi_client.get("/foo/123/bar/baz")
assert request.uri_template == "/foo/<id:int>/bar/<name:[A-z]+>"
def test_token(app):
@app.route("/")
async def handler(request):
return text("OK")
# uuid4 generated token.
token = "a1d895e0-553a-421a-8e22-5ff8ecb48cbf"
headers = {
"content-type": "application/json",
"Authorization": "{}".format(token),
}
request, response = app.test_client.get("/", headers=headers)
assert request.token == token
token = "a1d895e0-553a-421a-8e22-5ff8ecb48cbf"
headers = {
"content-type": "application/json",
"Authorization": "Token {}".format(token),
}
request, response = app.test_client.get("/", headers=headers)
assert request.token == token
token = "a1d895e0-553a-421a-8e22-5ff8ecb48cbf"
headers = {
"content-type": "application/json",
"Authorization": "Bearer {}".format(token),
}
request, response = app.test_client.get("/", headers=headers)
assert request.token == token
# no Authorization headers
headers = {"content-type": "application/json"}
request, response = app.test_client.get("/", headers=headers)
assert request.token is None
@pytest.mark.asyncio
async def test_token_asgi(app):
@app.route("/")
async def handler(request):
return text("OK")
# uuid4 generated token.
token = "a1d895e0-553a-421a-8e22-5ff8ecb48cbf"
headers = {
"content-type": "application/json",
"Authorization": "{}".format(token),
}
request, response = await app.asgi_client.get("/", headers=headers)
assert request.token == token
token = "a1d895e0-553a-421a-8e22-5ff8ecb48cbf"
headers = {
"content-type": "application/json",
"Authorization": "Token {}".format(token),
}
request, response = await app.asgi_client.get("/", headers=headers)
assert request.token == token
token = "a1d895e0-553a-421a-8e22-5ff8ecb48cbf"
headers = {
"content-type": "application/json",
"Authorization": "Bearer {}".format(token),
}
request, response = await app.asgi_client.get("/", headers=headers)
assert request.token == token
# no Authorization headers
headers = {"content-type": "application/json"}
request, response = await app.asgi_client.get("/", headers=headers)
assert request.token is None
def test_content_type(app):
@app.route("/")
async def handler(request):
return text(request.content_type)
request, response = app.test_client.get("/")
assert request.content_type == DEFAULT_HTTP_CONTENT_TYPE
assert response.text == DEFAULT_HTTP_CONTENT_TYPE
headers = {"content-type": "application/json"}
request, response = app.test_client.get("/", headers=headers)
assert request.content_type == "application/json"
assert response.text == "application/json"
@pytest.mark.asyncio
async def test_content_type_asgi(app):
@app.route("/")
async def handler(request):
return text(request.content_type)
request, response = await app.asgi_client.get("/")
assert request.content_type == DEFAULT_HTTP_CONTENT_TYPE
assert response.text == DEFAULT_HTTP_CONTENT_TYPE
headers = {"content-type": "application/json"}
request, response = await app.asgi_client.get("/", headers=headers)
assert request.content_type == "application/json"
assert response.text == "application/json"
def test_standard_forwarded(app):
@app.route("/")
async def handler(request):
return json(request.forwarded)
# Without configured FORWARDED_SECRET, x-headers should be respected
app.config.PROXIES_COUNT = 1
app.config.REAL_IP_HEADER = "x-real-ip"
headers = {
"Forwarded": (
'for=1.1.1.1, for=injected;host="'
', for="[::2]";proto=https;host=me.tld;path="/app/";secret=mySecret'
',for=broken;;secret=b0rked'
', for=127.0.0.3;scheme=http;port=1234'
),
"X-Real-IP": "127.0.0.2",
"X-Forwarded-For": "127.0.1.1",
"X-Scheme": "ws",
}
request, response = app.test_client.get("/", headers=headers)
assert response.json == { "for": "127.0.0.2", "proto": "ws" }
assert request.remote_addr == "127.0.0.2"
assert request.scheme == "ws"
assert request.server_port == 80
app.config.FORWARDED_SECRET = "mySecret"
request, response = app.test_client.get("/", headers=headers)
assert response.json == {
"for": "[::2]",
"proto": "https",
"host": "me.tld",
"path": "/app/",
"secret": "mySecret"
}
assert request.remote_addr == "[::2]"
assert request.server_name == "me.tld"
assert request.scheme == "https"
assert request.server_port == 443
# Empty Forwarded header -> use X-headers
headers["Forwarded"] = ""
request, response = app.test_client.get("/", headers=headers)
assert response.json == { "for": "127.0.0.2", "proto": "ws" }
# Header present but not matching anything
request, response = app.test_client.get("/", headers={"Forwarded": "."})
assert response.json == {}
# Forwarded header present but no matching secret -> use X-headers
headers = {
"Forwarded": 'for=1.1.1.1;secret=x, for=127.0.0.1',
"X-Real-IP": "127.0.0.2"
}
request, response = app.test_client.get("/", headers=headers)
assert response.json == {"for": "127.0.0.2"}
assert request.remote_addr == "127.0.0.2"
# Different formatting and hitting both ends of the header
headers = {"Forwarded": 'Secret="mySecret";For=127.0.0.4;Port=1234'}
request, response = app.test_client.get("/", headers=headers)
assert response.json == {
"for": "127.0.0.4",
"port": 1234,
"secret": "mySecret"
}
# Test escapes (modify this if you see anyone implementing quoted-pairs)
headers = {"Forwarded": 'for=test;quoted="\\,x=x;y=\\";secret=mySecret'}
request, response = app.test_client.get("/", headers=headers)
assert response.json == {
"for": "test",
"quoted": '\\,x=x;y=\\',
"secret": "mySecret"
}
# Secret insulated by malformed field #1
headers = {"Forwarded": 'for=test;secret=mySecret;b0rked;proto=wss;'}
request, response = app.test_client.get("/", headers=headers)
assert response.json == {"for": "test", "secret": "mySecret"}
# Secret insulated by malformed field #2
headers = {"Forwarded": 'for=test;b0rked;secret=mySecret;proto=wss'}
request, response = app.test_client.get("/", headers=headers)
assert response.json == {"proto": "wss", "secret": "mySecret"}
# Unexpected termination should not lose existing acceptable values
headers = {"Forwarded": 'b0rked;secret=mySecret;proto=wss'}
request, response = app.test_client.get("/", headers=headers)
assert response.json == {"proto": "wss", "secret": "mySecret"}
# Field normalization
headers = {
"Forwarded": 'PROTO=WSS;BY="CAFE::8000";FOR=unknown;PORT=X;HOST="A:2";'
'PATH="/With%20Spaces%22Quoted%22/sanicApp?key=val";SECRET=mySecret'
}
request, response = app.test_client.get("/", headers=headers)
assert response.json == {
"proto": "wss",
"by": "[cafe::8000]",
"host": "a:2",
"path": '/With Spaces"Quoted"/sanicApp?key=val',
"secret": "mySecret",
}
# Using "by" field as secret
app.config.FORWARDED_SECRET = "_proxySecret"
headers = {"Forwarded": 'for=1.2.3.4; by=_proxySecret'}
request, response = app.test_client.get("/", headers=headers)
assert response.json == {"for": "1.2.3.4", "by": "_proxySecret"}
@pytest.mark.asyncio
async def test_standard_forwarded_asgi(app):
@app.route("/")
async def handler(request):
return json(request.forwarded)
# Without configured FORWARDED_SECRET, x-headers should be respected
app.config.PROXIES_COUNT = 1
app.config.REAL_IP_HEADER = "x-real-ip"
headers = {
"Forwarded": (
'for=1.1.1.1, for=injected;host="'
', for="[::2]";proto=https;host=me.tld;path="/app/";secret=mySecret'
',for=broken;;secret=b0rked'
', for=127.0.0.3;scheme=http;port=1234'
),
"X-Real-IP": "127.0.0.2",
"X-Forwarded-For": "127.0.1.1",
"X-Scheme": "ws",
}
request, response = await app.asgi_client.get("/", headers=headers)
assert response.json() == { "for": "127.0.0.2", "proto": "ws" }
assert request.remote_addr == "127.0.0.2"
assert request.scheme == "ws"
assert request.server_port == 80
app.config.FORWARDED_SECRET = "mySecret"
request, response = await app.asgi_client.get("/", headers=headers)
assert response.json() == {
"for": "[::2]",
"proto": "https",
"host": "me.tld",
"path": "/app/",
"secret": "mySecret"
}
assert request.remote_addr == "[::2]"
assert request.server_name == "me.tld"
assert request.scheme == "https"
assert request.server_port == 443
# Empty Forwarded header -> use X-headers
headers["Forwarded"] = ""
request, response = await app.asgi_client.get("/", headers=headers)
assert response.json() == { "for": "127.0.0.2", "proto": "ws" }
# Header present but not matching anything
request, response = await app.asgi_client.get("/", headers={"Forwarded": "."})
assert response.json() == {}
# Forwarded header present but no matching secret -> use X-headers
headers = {
"Forwarded": 'for=1.1.1.1;secret=x, for=127.0.0.1',
"X-Real-IP": "127.0.0.2"
}
request, response = await app.asgi_client.get("/", headers=headers)
assert response.json() == {"for": "127.0.0.2"}
assert request.remote_addr == "127.0.0.2"
# Different formatting and hitting both ends of the header
headers = {"Forwarded": 'Secret="mySecret";For=127.0.0.4;Port=1234'}
request, response = await app.asgi_client.get("/", headers=headers)
assert response.json() == {
"for": "127.0.0.4",
"port": 1234,
"secret": "mySecret"
}
# Test escapes (modify this if you see anyone implementing quoted-pairs)
headers = {"Forwarded": 'for=test;quoted="\\,x=x;y=\\";secret=mySecret'}
request, response = await app.asgi_client.get("/", headers=headers)
assert response.json() == {
"for": "test",
"quoted": '\\,x=x;y=\\',
"secret": "mySecret"
}
# Secret insulated by malformed field #1
headers = {"Forwarded": 'for=test;secret=mySecret;b0rked;proto=wss;'}
request, response = await app.asgi_client.get("/", headers=headers)
assert response.json() == {"for": "test", "secret": "mySecret"}
# Secret insulated by malformed field #2
headers = {"Forwarded": 'for=test;b0rked;secret=mySecret;proto=wss'}
request, response = await app.asgi_client.get("/", headers=headers)
assert response.json() == {"proto": "wss", "secret": "mySecret"}
# Unexpected termination should not lose existing acceptable values
headers = {"Forwarded": 'b0rked;secret=mySecret;proto=wss'}
request, response = await app.asgi_client.get("/", headers=headers)
assert response.json() == {"proto": "wss", "secret": "mySecret"}
# Field normalization
headers = {
"Forwarded": 'PROTO=WSS;BY="CAFE::8000";FOR=unknown;PORT=X;HOST="A:2";'
'PATH="/With%20Spaces%22Quoted%22/sanicApp?key=val";SECRET=mySecret'
}
request, response = await app.asgi_client.get("/", headers=headers)
assert response.json() == {
"proto": "wss",
"by": "[cafe::8000]",
"host": "a:2",
"path": '/With Spaces"Quoted"/sanicApp?key=val',
"secret": "mySecret",
}
# Using "by" field as secret
app.config.FORWARDED_SECRET = "_proxySecret"
headers = {"Forwarded": 'for=1.2.3.4; by=_proxySecret'}
request, response = await app.asgi_client.get("/", headers=headers)
assert response.json() == {"for": "1.2.3.4", "by": "_proxySecret"}
def test_remote_addr_with_two_proxies(app):
app.config.PROXIES_COUNT = 2
app.config.REAL_IP_HEADER = "x-real-ip"
@app.route("/")
async def handler(request):
return text(request.remote_addr)
headers = {"X-Real-IP": "127.0.0.2", "X-Forwarded-For": "127.0.1.1"}
request, response = app.test_client.get("/", headers=headers)
assert request.remote_addr == "127.0.0.2"
assert response.text == "127.0.0.2"
headers = {"X-Forwarded-For": "127.0.1.1"}
request, response = app.test_client.get("/", headers=headers)
assert request.remote_addr == ""
assert response.text == ""
headers = {"X-Forwarded-For": "127.0.0.1, 127.0.1.2"}
request, response = app.test_client.get("/", headers=headers)
assert request.remote_addr == "127.0.0.1"
assert response.text == "127.0.0.1"
request, response = app.test_client.get("/")
assert request.remote_addr == ""
assert response.text == ""
headers = {"X-Forwarded-For": "127.0.0.1, , ,,127.0.1.2"}
request, response = app.test_client.get("/", headers=headers)
assert request.remote_addr == "127.0.0.1"
assert response.text == "127.0.0.1"
headers = {
"X-Forwarded-For": ", 127.0.2.2, , ,127.0.0.1, , ,,127.0.1.2"
}
request, response = app.test_client.get("/", headers=headers)
assert request.remote_addr == "127.0.0.1"
assert response.text == "127.0.0.1"
@pytest.mark.asyncio
async def test_remote_addr_with_two_proxies_asgi(app):
app.config.PROXIES_COUNT = 2
app.config.REAL_IP_HEADER = "x-real-ip"
@app.route("/")
async def handler(request):
return text(request.remote_addr)
headers = {"X-Real-IP": "127.0.0.2", "X-Forwarded-For": "127.0.1.1"}
request, response = await app.asgi_client.get("/", headers=headers)
assert request.remote_addr == "127.0.0.2"
assert response.text == "127.0.0.2"
headers = {"X-Forwarded-For": "127.0.1.1"}
request, response = await app.asgi_client.get("/", headers=headers)
assert request.remote_addr == ""
assert response.text == ""
headers = {"X-Forwarded-For": "127.0.0.1, 127.0.1.2"}
request, response = await app.asgi_client.get("/", headers=headers)
assert request.remote_addr == "127.0.0.1"
assert response.text == "127.0.0.1"
request, response = await app.asgi_client.get("/")
assert request.remote_addr == ""
assert response.text == ""
headers = {"X-Forwarded-For": "127.0.0.1, , ,,127.0.1.2"}
request, response = await app.asgi_client.get("/", headers=headers)
assert request.remote_addr == "127.0.0.1"
assert response.text == "127.0.0.1"
headers = {
"X-Forwarded-For": ", 127.0.2.2, , ,127.0.0.1, , ,,127.0.1.2"
}
request, response = await app.asgi_client.get("/", headers=headers)
assert request.remote_addr == "127.0.0.1"
assert response.text == "127.0.0.1"
def test_remote_addr_without_proxy(app):
app.config.PROXIES_COUNT = 0
@app.route("/")
async def handler(request):
return text(request.remote_addr)
headers = {"X-Real-IP": "127.0.0.2", "X-Forwarded-For": "127.0.1.1"}
request, response = app.test_client.get("/", headers=headers)
assert request.remote_addr == ""
assert response.text == ""
headers = {"X-Forwarded-For": "127.0.1.1"}
request, response = app.test_client.get("/", headers=headers)
assert request.remote_addr == ""
assert response.text == ""
headers = {"X-Forwarded-For": "127.0.0.1, 127.0.1.2"}
request, response = app.test_client.get("/", headers=headers)
assert request.remote_addr == ""
assert response.text == ""
@pytest.mark.asyncio
async def test_remote_addr_without_proxy_asgi(app):
app.config.PROXIES_COUNT = 0
@app.route("/")
async def handler(request):
return text(request.remote_addr)
headers = {"X-Real-IP": "127.0.0.2", "X-Forwarded-For": "127.0.1.1"}
request, response = await app.asgi_client.get("/", headers=headers)
assert request.remote_addr == ""
assert response.text == ""
headers = {"X-Forwarded-For": "127.0.1.1"}
request, response = await app.asgi_client.get("/", headers=headers)
assert request.remote_addr == ""
assert response.text == ""
headers = {"X-Forwarded-For": "127.0.0.1, 127.0.1.2"}
request, response = await app.asgi_client.get("/", headers=headers)
assert request.remote_addr == ""
assert response.text == ""
def test_remote_addr_custom_headers(app):
app.config.PROXIES_COUNT = 1
app.config.REAL_IP_HEADER = "Client-IP"
app.config.FORWARDED_FOR_HEADER = "Forwarded"
@app.route("/")
async def handler(request):
return text(request.remote_addr)
headers = {"X-Real-IP": "127.0.0.2", "Forwarded": "127.0.1.1"}
request, response = app.test_client.get("/", headers=headers)
assert request.remote_addr == "127.0.1.1"
assert response.text == "127.0.1.1"
headers = {"X-Forwarded-For": "127.0.1.1"}
request, response = app.test_client.get("/", headers=headers)
assert request.remote_addr == ""
assert response.text == ""
headers = {"Client-IP": "127.0.0.2", "Forwarded": "127.0.1.1"}
request, response = app.test_client.get("/", headers=headers)
assert request.remote_addr == "127.0.0.2"
assert response.text == "127.0.0.2"
@pytest.mark.asyncio
async def test_remote_addr_custom_headers_asgi(app):
app.config.PROXIES_COUNT = 1
app.config.REAL_IP_HEADER = "Client-IP"
app.config.FORWARDED_FOR_HEADER = "Forwarded"
@app.route("/")
async def handler(request):
return text(request.remote_addr)
headers = {"X-Real-IP": "127.0.0.2", "Forwarded": "127.0.1.1"}
request, response = await app.asgi_client.get("/", headers=headers)
assert request.remote_addr == "127.0.1.1"
assert response.text == "127.0.1.1"
headers = {"X-Forwarded-For": "127.0.1.1"}
request, response = await app.asgi_client.get("/", headers=headers)
assert request.remote_addr == ""
assert response.text == ""
headers = {"Client-IP": "127.0.0.2", "Forwarded": "127.0.1.1"}
request, response = await app.asgi_client.get("/", headers=headers)
assert request.remote_addr == "127.0.0.2"
assert response.text == "127.0.0.2"
def test_forwarded_scheme(app):
@app.route("/")
async def handler(request):
return text(request.remote_addr)
app.config.PROXIES_COUNT = 1
request, response = app.test_client.get("/")
assert request.scheme == "http"
request, response = app.test_client.get(
"/", headers={"X-Forwarded-For": "127.1.2.3", "X-Forwarded-Proto": "https"}
)
assert request.scheme == "https"
request, response = app.test_client.get("/", headers={"X-Forwarded-For": "127.1.2.3", "X-Scheme": "https"})
assert request.scheme == "https"
def test_match_info(app):
@app.route("/api/v1/user/<user_id>/")
async def handler(request, user_id):
return json(request.match_info)
request, response = app.test_client.get("/api/v1/user/sanic_user/")
assert request.match_info == {"user_id": "sanic_user"}
assert json_loads(response.text) == {"user_id": "sanic_user"}
@pytest.mark.asyncio
async def test_match_info_asgi(app):
@app.route("/api/v1/user/<user_id>/")
async def handler(request, user_id):
return json(request.match_info)
request, response = await app.asgi_client.get("/api/v1/user/sanic_user/")
assert request.match_info == {"user_id": "sanic_user"}
assert json_loads(response.text) == {"user_id": "sanic_user"}
# ------------------------------------------------------------ #
# POST
# ------------------------------------------------------------ #
def test_post_json(app):
@app.route("/", methods=["POST"])
async def handler(request):
return text("OK")
payload = {"test": "OK"}
headers = {"content-type": "application/json"}
request, response = app.test_client.post(
"/", data=json_dumps(payload), headers=headers
)
assert request.json.get("test") == "OK"
assert request.json.get("test") == "OK" # for request.parsed_json
assert response.text == "OK"
@pytest.mark.asyncio
async def test_post_json_asgi(app):
@app.route("/", methods=["POST"])
async def handler(request):
return text("OK")
payload = {"test": "OK"}
headers = {"content-type": "application/json"}
request, response = await app.asgi_client.post(
"/", data=json_dumps(payload), headers=headers
)
assert request.json.get("test") == "OK"
assert request.json.get("test") == "OK" # for request.parsed_json
assert response.text == "OK"
def test_post_form_urlencoded(app):
@app.route("/", methods=["POST"])
async def handler(request):
return text("OK")
payload = "test=OK"
headers = {"content-type": "application/x-www-form-urlencoded"}
request, response = app.test_client.post(
"/", data=payload, headers=headers
)
assert request.form.get("test") == "OK"
assert request.form.get("test") == "OK" # For request.parsed_form
@pytest.mark.asyncio
async def test_post_form_urlencoded_asgi(app):
@app.route("/", methods=["POST"])
async def handler(request):
return text("OK")
payload = "test=OK"
headers = {"content-type": "application/x-www-form-urlencoded"}
request, response = await app.asgi_client.post(
"/", data=payload, headers=headers
)
assert request.form.get("test") == "OK"
assert request.form.get("test") == "OK" # For request.parsed_form
@pytest.mark.parametrize(
"payload",
[
"------sanic\r\n"
'Content-Disposition: form-data; name="test"\r\n'
"\r\n"
"OK\r\n"
"------sanic--\r\n",
"------sanic\r\n"
'content-disposition: form-data; name="test"\r\n'
"\r\n"
"OK\r\n"
"------sanic--\r\n",
],
)
def test_post_form_multipart_form_data(app, payload):
@app.route("/", methods=["POST"])
async def handler(request):
return text("OK")
headers = {"content-type": "multipart/form-data; boundary=----sanic"}
request, response = app.test_client.post(data=payload, headers=headers)
assert request.form.get("test") == "OK"
@pytest.mark.parametrize(
"payload",
[
"------sanic\r\n"
'Content-Disposition: form-data; name="test"\r\n'
"\r\n"
"OK\r\n"
"------sanic--\r\n",
"------sanic\r\n"
'content-disposition: form-data; name="test"\r\n'
"\r\n"
"OK\r\n"
"------sanic--\r\n",
],
)
@pytest.mark.asyncio
async def test_post_form_multipart_form_data_asgi(app, payload):
@app.route("/", methods=["POST"])
async def handler(request):
return text("OK")
headers = {"content-type": "multipart/form-data; boundary=----sanic"}
request, response = await app.asgi_client.post(
"/", data=payload, headers=headers
)
assert request.form.get("test") == "OK"
@pytest.mark.parametrize(
"path,query,expected_url",
[
("/foo", "", "http://{}:{}/foo"),
("/bar/baz", "", "http://{}:{}/bar/baz"),
("/moo/boo", "arg1=val1", "http://{}:{}/moo/boo?arg1=val1"),
],
)
def test_url_attributes_no_ssl(app, path, query, expected_url):
async def handler(request):
return text("OK")
app.add_route(handler, path)
request, response = app.test_client.get(path + "?{}".format(query))
assert request.url == expected_url.format(HOST, PORT)
parsed = urlparse(request.url)
assert parsed.scheme == request.scheme
assert parsed.path == request.path
assert parsed.query == request.query_string
assert parsed.netloc == request.host
@pytest.mark.parametrize(
"path,query,expected_url",
[
("/foo", "", "http://{}/foo"),
("/bar/baz", "", "http://{}/bar/baz"),
("/moo/boo", "arg1=val1", "http://{}/moo/boo?arg1=val1"),
],
)
@pytest.mark.asyncio
async def test_url_attributes_no_ssl_asgi(app, path, query, expected_url):
async def handler(request):
return text("OK")
app.add_route(handler, path)
request, response = await app.asgi_client.get(path + "?{}".format(query))
assert request.url == expected_url.format(ASGI_HOST)
parsed = urlparse(request.url)
assert parsed.scheme == request.scheme
assert parsed.path == request.path
assert parsed.query == request.query_string
assert parsed.netloc == request.host
@pytest.mark.parametrize(
"path,query,expected_url",
[
("/foo", "", "https://{}:{}/foo"),
("/bar/baz", "", "https://{}:{}/bar/baz"),
("/moo/boo", "arg1=val1", "https://{}:{}/moo/boo?arg1=val1"),
],
)
def test_url_attributes_with_ssl_context(app, path, query, expected_url):
current_dir = os.path.dirname(os.path.realpath(__file__))
context = ssl.create_default_context(purpose=ssl.Purpose.CLIENT_AUTH)
context.load_cert_chain(
os.path.join(current_dir, "certs/selfsigned.cert"),
keyfile=os.path.join(current_dir, "certs/selfsigned.key"),
)
async def handler(request):
return text("OK")
app.add_route(handler, path)
request, response = app.test_client.get(
"https://{}:{}".format(HOST, PORT) + path + "?{}".format(query),
server_kwargs={"ssl": context},
)
assert request.url == expected_url.format(HOST, PORT)
parsed = urlparse(request.url)
assert parsed.scheme == request.scheme
assert parsed.path == request.path
assert parsed.query == request.query_string
assert parsed.netloc == request.host
@pytest.mark.parametrize(
"path,query,expected_url",
[
("/foo", "", "https://{}:{}/foo"),
("/bar/baz", "", "https://{}:{}/bar/baz"),
("/moo/boo", "arg1=val1", "https://{}:{}/moo/boo?arg1=val1"),
],
)
def test_url_attributes_with_ssl_dict(app, path, query, expected_url):
current_dir = os.path.dirname(os.path.realpath(__file__))
ssl_cert = os.path.join(current_dir, "certs/selfsigned.cert")
ssl_key = os.path.join(current_dir, "certs/selfsigned.key")
ssl_dict = {"cert": ssl_cert, "key": ssl_key}
async def handler(request):
return text("OK")
app.add_route(handler, path)
request, response = app.test_client.get(
"https://{}:{}".format(HOST, PORT) + path + "?{}".format(query),
server_kwargs={"ssl": ssl_dict},
)
assert request.url == expected_url.format(HOST, PORT)
parsed = urlparse(request.url)
assert parsed.scheme == request.scheme
assert parsed.path == request.path
assert parsed.query == request.query_string
assert parsed.netloc == request.host
def test_invalid_ssl_dict(app):
@app.get("/test")
async def handler(request):
return text("ssl test")
ssl_dict = {"cert": None, "key": None}
with pytest.raises(ValueError) as excinfo:
request, response = app.test_client.get(
"/test", server_kwargs={"ssl": ssl_dict}
)
assert str(excinfo.value) == "SSLContext or certificate and key required."
def test_form_with_multiple_values(app):
@app.route("/", methods=["POST"])
async def handler(request):
return text("OK")
payload = "selectedItems=v1&selectedItems=v2&selectedItems=v3"
headers = {"content-type": "application/x-www-form-urlencoded"}
request, response = app.test_client.post(
"/", data=payload, headers=headers
)
assert request.form.getlist("selectedItems") == ["v1", "v2", "v3"]
@pytest.mark.asyncio
async def test_form_with_multiple_values_asgi(app):
@app.route("/", methods=["POST"])
async def handler(request):
return text("OK")
payload = "selectedItems=v1&selectedItems=v2&selectedItems=v3"
headers = {"content-type": "application/x-www-form-urlencoded"}
request, response = await app.asgi_client.post(
"/", data=payload, headers=headers
)
assert request.form.getlist("selectedItems") == ["v1", "v2", "v3"]
def test_request_string_representation(app):
@app.route("/", methods=["GET"])
async def get(request):
return text("OK")
request, _ = app.test_client.get("/")
assert repr(request) == "<Request: GET />"
@pytest.mark.asyncio
async def test_request_string_representation_asgi(app):
@app.route("/", methods=["GET"])
async def get(request):
return text("OK")
request, _ = await app.asgi_client.get("/")
assert repr(request) == "<Request: GET />"
@pytest.mark.parametrize(
"payload,filename",
[
(
"------sanic\r\n"
'Content-Disposition: form-data; filename="filename"; name="test"\r\n'
"\r\n"
"OK\r\n"
"------sanic--\r\n",
"filename",
),
(
"------sanic\r\n"
'content-disposition: form-data; filename="filename"; name="test"\r\n'
"\r\n"
'content-type: application/json; {"field": "value"}\r\n'
"------sanic--\r\n",
"filename",
),
(
"------sanic\r\n"
'Content-Disposition: form-data; filename=""; name="test"\r\n'
"\r\n"
"OK\r\n"
"------sanic--\r\n",
"",
),
(
"------sanic\r\n"
'content-disposition: form-data; filename=""; name="test"\r\n'
"\r\n"
'content-type: application/json; {"field": "value"}\r\n'
"------sanic--\r\n",
"",
),
(
"------sanic\r\n"
'Content-Disposition: form-data; filename*="utf-8\'\'filename_%C2%A0_test"; name="test"\r\n'
"\r\n"
"OK\r\n"
"------sanic--\r\n",
"filename_\u00A0_test",
),
(
"------sanic\r\n"
'content-disposition: form-data; filename*="utf-8\'\'filename_%C2%A0_test"; name="test"\r\n'
"\r\n"
'content-type: application/json; {"field": "value"}\r\n'
"------sanic--\r\n",
"filename_\u00A0_test",
),
],
)
def test_request_multipart_files(app, payload, filename):
@app.route("/", methods=["POST"])
async def post(request):
return text("OK")
headers = {"content-type": "multipart/form-data; boundary=----sanic"}
request, _ = app.test_client.post(data=payload, headers=headers)
assert request.files.get("test").name == filename
@pytest.mark.parametrize(
"payload,filename",
[
(
"------sanic\r\n"
'Content-Disposition: form-data; filename="filename"; name="test"\r\n'
"\r\n"
"OK\r\n"
"------sanic--\r\n",
"filename",
),
(
"------sanic\r\n"
'content-disposition: form-data; filename="filename"; name="test"\r\n'
"\r\n"
'content-type: application/json; {"field": "value"}\r\n'
"------sanic--\r\n",
"filename",
),
(
"------sanic\r\n"
'Content-Disposition: form-data; filename=""; name="test"\r\n'
"\r\n"
"OK\r\n"
"------sanic--\r\n",
"",
),
(
"------sanic\r\n"
'content-disposition: form-data; filename=""; name="test"\r\n'
"\r\n"
'content-type: application/json; {"field": "value"}\r\n'
"------sanic--\r\n",
"",
),
(
"------sanic\r\n"
'Content-Disposition: form-data; filename*="utf-8\'\'filename_%C2%A0_test"; name="test"\r\n'
"\r\n"
"OK\r\n"
"------sanic--\r\n",
"filename_\u00A0_test",
),
(
"------sanic\r\n"
'content-disposition: form-data; filename*="utf-8\'\'filename_%C2%A0_test"; name="test"\r\n'
"\r\n"
'content-type: application/json; {"field": "value"}\r\n'
"------sanic--\r\n",
"filename_\u00A0_test",
),
],
)
@pytest.mark.asyncio
async def test_request_multipart_files_asgi(app, payload, filename):
@app.route("/", methods=["POST"])
async def post(request):
return text("OK")
headers = {"content-type": "multipart/form-data; boundary=----sanic"}
request, _ = await app.asgi_client.post("/", data=payload, headers=headers)
assert request.files.get("test").name == filename
def test_request_multipart_file_with_json_content_type(app):
@app.route("/", methods=["POST"])
async def post(request):
return text("OK")
payload = (
"------sanic\r\n"
'Content-Disposition: form-data; name="file"; filename="test.json"\r\n'
"Content-Type: application/json\r\n"
"Content-Length: 0"
"\r\n"
"\r\n"
"------sanic--"
)
headers = {"content-type": "multipart/form-data; boundary=------sanic"}
request, _ = app.test_client.post(data=payload, headers=headers)
assert request.files.get("file").type == "application/json"
@pytest.mark.asyncio
async def test_request_multipart_file_with_json_content_type_asgi(app):
@app.route("/", methods=["POST"])
async def post(request):
return text("OK")
payload = (
"------sanic\r\n"
'Content-Disposition: form-data; name="file"; filename="test.json"\r\n'
"Content-Type: application/json\r\n"
"Content-Length: 0"
"\r\n"
"\r\n"
"------sanic--"
)
headers = {"content-type": "multipart/form-data; boundary=------sanic"}
request, _ = await app.asgi_client.post("/", data=payload, headers=headers)
assert request.files.get("file").type == "application/json"
def test_request_multipart_file_without_field_name(app, caplog):
@app.route("/", methods=["POST"])
async def post(request):
return text("OK")
payload = (
'------sanic\r\nContent-Disposition: form-data; filename="test.json"'
"\r\nContent-Type: application/json\r\n\r\n\r\n------sanic--"
)
headers = {"content-type": "multipart/form-data; boundary=------sanic"}
request, _ = app.test_client.post(
data=payload, headers=headers, debug=True
)
with caplog.at_level(logging.DEBUG):
request.form
assert caplog.record_tuples[-1] == (
"sanic.root",
logging.DEBUG,
"Form-data field does not have a 'name' parameter "
"in the Content-Disposition header",
)
def test_request_multipart_file_duplicate_filed_name(app):
@app.route("/", methods=["POST"])
async def post(request):
return text("OK")
payload = (
"--e73ffaa8b1b2472b8ec848de833cb05b\r\n"
'Content-Disposition: form-data; name="file"\r\n'
"Content-Type: application/octet-stream\r\n"
"Content-Length: 15\r\n"
"\r\n"
'{"test":"json"}\r\n'
"--e73ffaa8b1b2472b8ec848de833cb05b\r\n"
'Content-Disposition: form-data; name="file"\r\n'
"Content-Type: application/octet-stream\r\n"
"Content-Length: 15\r\n"
"\r\n"
'{"test":"json2"}\r\n'
"--e73ffaa8b1b2472b8ec848de833cb05b--\r\n"
)
headers = {
"Content-Type": "multipart/form-data; boundary=e73ffaa8b1b2472b8ec848de833cb05b"
}
request, _ = app.test_client.post(
data=payload, headers=headers, debug=True
)
assert request.form.getlist("file") == [
'{"test":"json"}',
'{"test":"json2"}',
]
@pytest.mark.asyncio
async def test_request_multipart_file_duplicate_filed_name_asgi(app):
@app.route("/", methods=["POST"])
async def post(request):
return text("OK")
payload = (
"--e73ffaa8b1b2472b8ec848de833cb05b\r\n"
'Content-Disposition: form-data; name="file"\r\n'
"Content-Type: application/octet-stream\r\n"
"Content-Length: 15\r\n"
"\r\n"
'{"test":"json"}\r\n'
"--e73ffaa8b1b2472b8ec848de833cb05b\r\n"
'Content-Disposition: form-data; name="file"\r\n'
"Content-Type: application/octet-stream\r\n"
"Content-Length: 15\r\n"
"\r\n"
'{"test":"json2"}\r\n'
"--e73ffaa8b1b2472b8ec848de833cb05b--\r\n"
)
headers = {
"Content-Type": "multipart/form-data; boundary=e73ffaa8b1b2472b8ec848de833cb05b"
}
request, _ = await app.asgi_client.post("/", data=payload, headers=headers)
assert request.form.getlist("file") == [
'{"test":"json"}',
'{"test":"json2"}',
]
def test_request_multipart_with_multiple_files_and_type(app):
@app.route("/", methods=["POST"])
async def post(request):
return text("OK")
payload = (
'------sanic\r\nContent-Disposition: form-data; name="file"; filename="test.json"'
"\r\nContent-Type: application/json\r\n\r\n\r\n"
'------sanic\r\nContent-Disposition: form-data; name="file"; filename="some_file.pdf"\r\n'
"Content-Type: application/pdf\r\n\r\n\r\n------sanic--"
)
headers = {"content-type": "multipart/form-data; boundary=------sanic"}
request, _ = app.test_client.post(data=payload, headers=headers)
assert len(request.files.getlist("file")) == 2
assert request.files.getlist("file")[0].type == "application/json"
assert request.files.getlist("file")[1].type == "application/pdf"
@pytest.mark.asyncio
async def test_request_multipart_with_multiple_files_and_type_asgi(app):
@app.route("/", methods=["POST"])
async def post(request):
return text("OK")
payload = (
'------sanic\r\nContent-Disposition: form-data; name="file"; filename="test.json"'
"\r\nContent-Type: application/json\r\n\r\n\r\n"
'------sanic\r\nContent-Disposition: form-data; name="file"; filename="some_file.pdf"\r\n'
"Content-Type: application/pdf\r\n\r\n\r\n------sanic--"
)
headers = {"content-type": "multipart/form-data; boundary=------sanic"}
request, _ = await app.asgi_client.post("/", data=payload, headers=headers)
assert len(request.files.getlist("file")) == 2
assert request.files.getlist("file")[0].type == "application/json"
assert request.files.getlist("file")[1].type == "application/pdf"
def test_request_repr(app):
@app.get("/")
def handler(request):
return text("pass")
request, response = app.test_client.get("/")
assert repr(request) == "<Request: GET />"
request.method = None
assert repr(request) == "<Request: None />"
@pytest.mark.asyncio
async def test_request_repr_asgi(app):
@app.get("/")
def handler(request):
return text("pass")
request, response = await app.asgi_client.get("/")
assert repr(request) == "<Request: GET />"
request.method = None
assert repr(request) == "<Request: None />"
def test_request_bool(app):
@app.get("/")
def handler(request):
return text("pass")
request, response = app.test_client.get("/")
assert bool(request)
request.transport = False
assert not bool(request)
def test_request_parsing_form_failed(app, caplog):
@app.route("/", methods=["POST"])
async def handler(request):
return text("OK")
payload = "test=OK"
headers = {"content-type": "multipart/form-data"}
request, response = app.test_client.post(
"/", data=payload, headers=headers
)
with caplog.at_level(logging.ERROR):
request.form
assert caplog.record_tuples[-1] == (
"sanic.error",
logging.ERROR,
"Failed when parsing form",
)
@pytest.mark.asyncio
async def test_request_parsing_form_failed_asgi(app, caplog):
@app.route("/", methods=["POST"])
async def handler(request):
return text("OK")
payload = "test=OK"
headers = {"content-type": "multipart/form-data"}
request, response = await app.asgi_client.post(
"/", data=payload, headers=headers
)
with caplog.at_level(logging.ERROR):
request.form
assert caplog.record_tuples[-1] == (
"sanic.error",
logging.ERROR,
"Failed when parsing form",
)
def test_request_args_no_query_string(app):
@app.get("/")
def handler(request):
return text("pass")
request, response = app.test_client.get("/")
assert request.args == {}
@pytest.mark.asyncio
async def test_request_args_no_query_string_await(app):
@app.get("/")
def handler(request):
return text("pass")
request, response = await app.asgi_client.get("/")
assert request.args == {}
def test_request_raw_args(app):
params = {"test": "OK"}
@app.get("/")
def handler(request):
return text("pass")
request, response = app.test_client.get("/", params=params)
assert request.raw_args == params
@pytest.mark.asyncio
async def test_request_raw_args_asgi(app):
params = {"test": "OK"}
@app.get("/")
def handler(request):
return text("pass")
request, response = await app.asgi_client.get("/", params=params)
assert request.raw_args == params
def test_request_query_args(app):
# test multiple params with the same key
params = [("test", "value1"), ("test", "value2")]
@app.get("/")
def handler(request):
return text("pass")
request, response = app.test_client.get("/", params=params)
assert request.query_args == params
# test cached value
assert (
request.parsed_not_grouped_args[(False, False, "utf-8", "replace")]
== request.query_args
)
# test params directly in the url
request, response = app.test_client.get("/?test=value1&test=value2")
assert request.query_args == params
# test unique params
params = [("test1", "value1"), ("test2", "value2")]
request, response = app.test_client.get("/", params=params)
assert request.query_args == params
# test no params
request, response = app.test_client.get("/")
assert not request.query_args
@pytest.mark.asyncio
async def test_request_query_args_asgi(app):
# test multiple params with the same key
params = [("test", "value1"), ("test", "value2")]
@app.get("/")
def handler(request):
return text("pass")
request, response = await app.asgi_client.get("/", params=params)
assert request.query_args == params
# test cached value
assert (
request.parsed_not_grouped_args[(False, False, "utf-8", "replace")]
== request.query_args
)
# test params directly in the url
request, response = await app.asgi_client.get("/?test=value1&test=value2")
assert request.query_args == params
# test unique params
params = [("test1", "value1"), ("test2", "value2")]
request, response = await app.asgi_client.get("/", params=params)
assert request.query_args == params
# test no params
request, response = await app.asgi_client.get("/")
assert not request.query_args
def test_request_query_args_custom_parsing(app):
@app.get("/")
def handler(request):
return text("pass")
request, response = app.test_client.get(
"/?test1=value1&test2=&test3=value3"
)
assert request.get_query_args(keep_blank_values=True) == [
("test1", "value1"),
("test2", ""),
("test3", "value3"),
]
assert request.query_args == [("test1", "value1"), ("test3", "value3")]
assert request.get_query_args(keep_blank_values=False) == [
("test1", "value1"),
("test3", "value3"),
]
assert request.get_args(keep_blank_values=True) == RequestParameters(
{"test1": ["value1"], "test2": [""], "test3": ["value3"]}
)
assert request.args == RequestParameters(
{"test1": ["value1"], "test3": ["value3"]}
)
assert request.get_args(keep_blank_values=False) == RequestParameters(
{"test1": ["value1"], "test3": ["value3"]}
)
@pytest.mark.asyncio
async def test_request_query_args_custom_parsing_asgi(app):
@app.get("/")
def handler(request):
return text("pass")
request, response = await app.asgi_client.get(
"/?test1=value1&test2=&test3=value3"
)
assert request.get_query_args(keep_blank_values=True) == [
("test1", "value1"),
("test2", ""),
("test3", "value3"),
]
assert request.query_args == [("test1", "value1"), ("test3", "value3")]
assert request.get_query_args(keep_blank_values=False) == [
("test1", "value1"),
("test3", "value3"),
]
assert request.get_args(keep_blank_values=True) == RequestParameters(
{"test1": ["value1"], "test2": [""], "test3": ["value3"]}
)
assert request.args == RequestParameters(
{"test1": ["value1"], "test3": ["value3"]}
)
assert request.get_args(keep_blank_values=False) == RequestParameters(
{"test1": ["value1"], "test3": ["value3"]}
)
def test_request_cookies(app):
cookies = {"test": "OK"}
@app.get("/")
def handler(request):
return text("OK")
request, response = app.test_client.get("/", cookies=cookies)
assert request.cookies == cookies
assert request.cookies == cookies # For request._cookies
@pytest.mark.asyncio
async def test_request_cookies_asgi(app):
cookies = {"test": "OK"}
@app.get("/")
def handler(request):
return text("OK")
request, response = await app.asgi_client.get("/", cookies=cookies)
assert request.cookies == cookies
assert request.cookies == cookies # For request._cookies
def test_request_cookies_without_cookies(app):
@app.get("/")
def handler(request):
return text("OK")
request, response = app.test_client.get("/")
assert request.cookies == {}
@pytest.mark.asyncio
async def test_request_cookies_without_cookies_asgi(app):
@app.get("/")
def handler(request):
return text("OK")
request, response = await app.asgi_client.get("/")
assert request.cookies == {}
def test_request_port(app):
@app.get("/")
def handler(request):
return text("OK")
request, response = app.test_client.get("/")
port = request.port
assert isinstance(port, int)
delattr(request, "_socket")
delattr(request, "_port")
port = request.port
assert isinstance(port, int)
assert hasattr(request, "_socket")
assert hasattr(request, "_port")
@pytest.mark.asyncio
async def test_request_port_asgi(app):
@app.get("/")
def handler(request):
return text("OK")
request, response = await app.asgi_client.get("/")
port = request.port
assert isinstance(port, int)
delattr(request, "_socket")
delattr(request, "_port")
port = request.port
assert isinstance(port, int)
assert hasattr(request, "_socket")
assert hasattr(request, "_port")
def test_request_socket(app):
@app.get("/")
def handler(request):
return text("OK")
request, response = app.test_client.get("/")
socket = request.socket
assert isinstance(socket, tuple)
ip = socket[0]
port = socket[1]
assert ip == request.ip
assert port == request.port
delattr(request, "_socket")
socket = request.socket
assert isinstance(socket, tuple)
assert hasattr(request, "_socket")
def test_request_server_name(app):
@app.get("/")
def handler(request):
return text("OK")
request, response = app.test_client.get("/")
assert request.server_name == "127.0.0.1"
def test_request_server_name_in_host_header(app):
@app.get("/")
def handler(request):
return text("OK")
request, response = app.test_client.get(
"/", headers={"Host": "my-server:5555"}
)
assert request.server_name == "my-server"
request, response = app.test_client.get(
"/", headers={"Host": "[2a00:1450:400f:80c::200e]:5555"}
)
assert request.server_name == "[2a00:1450:400f:80c::200e]"
request, response = app.test_client.get(
"/", headers={"Host": "mal_formed"}
)
assert request.server_name == None # For now (later maybe 127.0.0.1)
def test_request_server_name_forwarded(app):
@app.get("/")
def handler(request):
return text("OK")
app.config.PROXIES_COUNT = 1
request, response = app.test_client.get(
"/",
headers={"Host": "my-server:5555", "X-Forwarded-For": "127.1.2.3", "X-Forwarded-Host": "your-server"},
)
assert request.server_name == "your-server"
def test_request_server_port(app):
@app.get("/")
def handler(request):
return text("OK")
request, response = app.test_client.get("/", headers={"Host": "my-server"})
assert request.server_port == app.test_client.port
def test_request_server_port_in_host_header(app):
@app.get("/")
def handler(request):
return text("OK")
request, response = app.test_client.get(
"/", headers={"Host": "my-server:5555"}
)
assert request.server_port == 5555
request, response = app.test_client.get(
"/", headers={"Host": "[2a00:1450:400f:80c::200e]:5555"}
)
assert request.server_port == 5555
request, response = app.test_client.get(
"/", headers={"Host": "mal_formed:5555"}
)
assert request.server_port == app.test_client.port
def test_request_server_port_forwarded(app):
@app.get("/")
def handler(request):
return text("OK")
app.config.PROXIES_COUNT = 1
request, response = app.test_client.get(
"/", headers={"Host": "my-server:5555", "X-Forwarded-For": "127.1.2.3", "X-Forwarded-Port": "4444"}
)
assert request.server_port == 4444
def test_request_form_invalid_content_type(app):
@app.route("/", methods=["POST"])
async def post(request):
return text("OK")
request, response = app.test_client.post("/", json={"test": "OK"})
assert request.form == {}
def test_server_name_and_url_for(app):
@app.get("/foo")
def handler(request):
return text("ok")
app.config.SERVER_NAME = "my-server"
assert app.url_for("handler", _external=True) == "http://my-server/foo"
request, response = app.test_client.get("/foo")
assert request.url_for("handler") == f"http://my-server:{app.test_client.port}/foo"
app.config.SERVER_NAME = "https://my-server/path"
request, response = app.test_client.get("/foo")
url = f"https://my-server/path/foo"
assert app.url_for("handler", _external=True) == url
assert request.url_for("handler") == url
def test_url_for_with_forwarded_request(app):
@app.get("/")
def handler(request):
return text("OK")
@app.get("/another_view/")
def view_name(request):
return text("OK")
app.config.SERVER_NAME = "my-server"
app.config.PROXIES_COUNT = 1
request, response = app.test_client.get(
"/", headers={"X-Forwarded-For": "127.1.2.3", "X-Forwarded-Proto": "https", "X-Forwarded-Port": "6789"}
)
assert app.url_for("view_name") == "/another_view"
assert (
app.url_for("view_name", _external=True)
== "http://my-server/another_view"
)
assert (
request.url_for("view_name") == "https://my-server:6789/another_view"
)
request, response = app.test_client.get(
"/", headers={"X-Forwarded-For": "127.1.2.3", "X-Forwarded-Proto": "https", "X-Forwarded-Port": "443"}
)
assert request.url_for("view_name") == "https://my-server/another_view"
@pytest.mark.asyncio
async def test_request_form_invalid_content_type_asgi(app):
@app.route("/", methods=["POST"])
async def post(request):
return text("OK")
request, response = await app.asgi_client.post("/", json={"test": "OK"})
assert request.form == {}
def test_endpoint_basic():
app = Sanic()
@app.route("/")
def my_unique_handler(request):
return text("Hello")
request, response = app.test_client.get("/")
assert request.endpoint == "test_requests.my_unique_handler"
@pytest.mark.asyncio
async def test_endpoint_basic_asgi():
app = Sanic()
@app.route("/")
def my_unique_handler(request):
return text("Hello")
request, response = await app.asgi_client.get("/")
assert request.endpoint == "test_requests.my_unique_handler"
def test_endpoint_named_app():
app = Sanic("named")
@app.route("/")
def my_unique_handler(request):
return text("Hello")
request, response = app.test_client.get("/")
assert request.endpoint == "named.my_unique_handler"
@pytest.mark.asyncio
async def test_endpoint_named_app_asgi():
app = Sanic("named")
@app.route("/")
def my_unique_handler(request):
return text("Hello")
request, response = await app.asgi_client.get("/")
assert request.endpoint == "named.my_unique_handler"
def test_endpoint_blueprint():
bp = Blueprint("my_blueprint", url_prefix="/bp")
@bp.route("/")
async def bp_root(request):
return text("Hello")
app = Sanic("named")
app.blueprint(bp)
request, response = app.test_client.get("/bp")
assert request.endpoint == "named.my_blueprint.bp_root"
@pytest.mark.asyncio
async def test_endpoint_blueprint_asgi():
bp = Blueprint("my_blueprint", url_prefix="/bp")
@bp.route("/")
async def bp_root(request):
return text("Hello")
app = Sanic("named")
app.blueprint(bp)
request, response = await app.asgi_client.get("/bp")
assert request.endpoint == "named.my_blueprint.bp_root"
Reference in New Issue View Git Blame Copy Permalink