Added tests and middleware, and improved documentation
This commit is contained in:
47
tests/helpers.py
Normal file
47
tests/helpers.py
Normal file
@@ -0,0 +1,47 @@
|
||||
import aiohttp
|
||||
from sanic.log import log
|
||||
|
||||
HOST = '127.0.0.1'
|
||||
PORT = 42101
|
||||
|
||||
async def local_request(method, uri, *args, **kwargs):
|
||||
url = 'http://{host}:{port}{uri}'.format(host=HOST, port=PORT, uri=uri)
|
||||
log.info(url)
|
||||
async with aiohttp.ClientSession() as session:
|
||||
async with getattr(session, method)(url, *args, **kwargs) as response:
|
||||
response.text = await response.text()
|
||||
return response
|
||||
|
||||
def sanic_endpoint_test(app, method='get', uri='/', gather_request=True, *request_args, **request_kwargs):
|
||||
results = []
|
||||
exceptions = []
|
||||
|
||||
if gather_request:
|
||||
@app.middleware
|
||||
def _collect_request(request):
|
||||
results.append(request)
|
||||
|
||||
async def _collect_response(loop):
|
||||
try:
|
||||
response = await local_request(method, uri, *request_args, **request_kwargs)
|
||||
results.append(response)
|
||||
except Exception as e:
|
||||
exceptions.append(e)
|
||||
app.stop()
|
||||
|
||||
app.run(host='0.0.0.0', port=42101, debug=True, after_start=_collect_response)
|
||||
|
||||
if exceptions:
|
||||
raise ValueError("Exception during request: {}".format(exceptions))
|
||||
|
||||
if gather_request:
|
||||
try:
|
||||
request, response = results
|
||||
return request, response
|
||||
except:
|
||||
raise ValueError("request and response object expected, got ({})".format(results[0].text))
|
||||
else:
|
||||
try:
|
||||
return results[0]
|
||||
except:
|
||||
raise ValueError("request object expected, got ({})".format(results))
|
||||
@@ -14,4 +14,8 @@ app = Sanic("test")
|
||||
async def test(request):
|
||||
return json({ "test": True })
|
||||
|
||||
app.run(host="0.0.0.0", port=sys.argv[1])
|
||||
@app.route("/file")
|
||||
async def test(request):
|
||||
return json({ "test": True, "files": request.files, "fields": request.form })
|
||||
|
||||
app.run(host="0.0.0.0", port=sys.argv[1], debug=True)
|
||||
81
tests/test_middleware.py
Normal file
81
tests/test_middleware.py
Normal file
@@ -0,0 +1,81 @@
|
||||
from json import loads as json_loads, dumps as json_dumps
|
||||
from sanic import Sanic
|
||||
from sanic.request import Request
|
||||
from sanic.response import json, text, HTTPResponse
|
||||
from helpers import sanic_endpoint_test
|
||||
|
||||
# ------------------------------------------------------------ #
|
||||
# GET
|
||||
# ------------------------------------------------------------ #
|
||||
|
||||
def test_middleware_request():
|
||||
app = Sanic('test_middleware_request')
|
||||
|
||||
results = []
|
||||
@app.middleware
|
||||
async def handler(request):
|
||||
results.append(request)
|
||||
|
||||
@app.route('/')
|
||||
async def handler(request):
|
||||
return text('OK')
|
||||
|
||||
request, response = sanic_endpoint_test(app)
|
||||
|
||||
assert response.text == 'OK'
|
||||
assert type(results[0]) is Request
|
||||
|
||||
def test_middleware_response():
|
||||
app = Sanic('test_middleware_response')
|
||||
|
||||
results = []
|
||||
@app.middleware('request')
|
||||
async def process_response(request):
|
||||
results.append(request)
|
||||
@app.middleware('response')
|
||||
async def process_response(request, response):
|
||||
results.append(request)
|
||||
results.append(response)
|
||||
|
||||
@app.route('/')
|
||||
async def handler(request):
|
||||
return text('OK')
|
||||
|
||||
request, response = sanic_endpoint_test(app)
|
||||
|
||||
assert response.text == 'OK'
|
||||
assert type(results[0]) is Request
|
||||
assert type(results[1]) is Request
|
||||
assert issubclass(type(results[2]), HTTPResponse)
|
||||
|
||||
def test_middleware_override_request():
|
||||
app = Sanic('test_middleware_override_request')
|
||||
|
||||
@app.middleware
|
||||
async def halt_request(request):
|
||||
return text('OK')
|
||||
|
||||
@app.route('/')
|
||||
async def handler(request):
|
||||
return text('FAIL')
|
||||
|
||||
response = sanic_endpoint_test(app, gather_request=False)
|
||||
|
||||
assert response.status == 200
|
||||
assert response.text == 'OK'
|
||||
|
||||
def test_middleware_override_response():
|
||||
app = Sanic('test_middleware_override_response')
|
||||
|
||||
@app.middleware('response')
|
||||
async def process_response(request, response):
|
||||
return text('OK')
|
||||
|
||||
@app.route('/')
|
||||
async def handler(request):
|
||||
return text('FAIL')
|
||||
|
||||
request, response = sanic_endpoint_test(app)
|
||||
|
||||
assert response.status == 200
|
||||
assert response.text == 'OK'
|
||||
79
tests/test_requests.py
Normal file
79
tests/test_requests.py
Normal file
@@ -0,0 +1,79 @@
|
||||
from json import loads as json_loads, dumps as json_dumps
|
||||
from sanic import Sanic
|
||||
from sanic.response import json, text
|
||||
from helpers import sanic_endpoint_test
|
||||
|
||||
# ------------------------------------------------------------ #
|
||||
# GET
|
||||
# ------------------------------------------------------------ #
|
||||
|
||||
def test_sync():
|
||||
app = Sanic('test_text')
|
||||
|
||||
@app.route('/')
|
||||
def handler(request):
|
||||
return text('Hello')
|
||||
|
||||
request, response = sanic_endpoint_test(app)
|
||||
|
||||
assert response.text == 'Hello'
|
||||
|
||||
def test_text():
|
||||
app = Sanic('test_text')
|
||||
|
||||
@app.route('/')
|
||||
async def handler(request):
|
||||
return text('Hello')
|
||||
|
||||
request, response = sanic_endpoint_test(app)
|
||||
|
||||
assert response.text == 'Hello'
|
||||
|
||||
|
||||
def test_json():
|
||||
app = Sanic('test_json')
|
||||
|
||||
@app.route('/')
|
||||
async def handler(request):
|
||||
return json({"test":True})
|
||||
|
||||
request, response = sanic_endpoint_test(app)
|
||||
|
||||
try:
|
||||
results = json_loads(response.text)
|
||||
except:
|
||||
raise ValueError("Expected JSON response but got '{}'".format(response))
|
||||
|
||||
assert results.get('test') == True
|
||||
|
||||
|
||||
def test_query_string():
|
||||
app = Sanic('test_query_string')
|
||||
|
||||
@app.route('/')
|
||||
async def handler(request):
|
||||
return text('OK')
|
||||
|
||||
request, response = sanic_endpoint_test(app, params=[("test1", 1), ("test2", "false"), ("test2", "true")])
|
||||
|
||||
assert request.args.get('test1') == '1'
|
||||
assert request.args.get('test2') == 'false'
|
||||
|
||||
# ------------------------------------------------------------ #
|
||||
# POST
|
||||
# ------------------------------------------------------------ #
|
||||
|
||||
def test_post_json():
|
||||
app = Sanic('test_post_json')
|
||||
|
||||
@app.route('/')
|
||||
async def handler(request):
|
||||
return text('OK')
|
||||
|
||||
payload = {'test': 'OK'}
|
||||
headers = {'content-type': 'application/json'}
|
||||
|
||||
request, response = sanic_endpoint_test(app, data=json_dumps(payload), headers=headers)
|
||||
|
||||
assert request.json.get('test') == 'OK'
|
||||
assert response.text == 'OK'
|
||||
95
tests/test_routes.py
Normal file
95
tests/test_routes.py
Normal file
@@ -0,0 +1,95 @@
|
||||
from json import loads as json_loads, dumps as json_dumps
|
||||
from sanic import Sanic
|
||||
from sanic.response import json, text
|
||||
from helpers import sanic_endpoint_test
|
||||
|
||||
# ------------------------------------------------------------ #
|
||||
# UTF-8
|
||||
# ------------------------------------------------------------ #
|
||||
|
||||
def test_dynamic_route():
|
||||
app = Sanic('test_dynamic_route')
|
||||
|
||||
results = []
|
||||
|
||||
@app.route('/folder/<name>')
|
||||
async def handler(request, name):
|
||||
results.append(name)
|
||||
return text('OK')
|
||||
|
||||
request, response = sanic_endpoint_test(app, uri='/folder/test123')
|
||||
|
||||
assert response.text == 'OK'
|
||||
assert results[0] == 'test123'
|
||||
|
||||
def test_dynamic_route_string():
|
||||
app = Sanic('test_dynamic_route_string')
|
||||
|
||||
results = []
|
||||
|
||||
@app.route('/folder/<name:string>')
|
||||
async def handler(request, name):
|
||||
results.append(name)
|
||||
return text('OK')
|
||||
|
||||
request, response = sanic_endpoint_test(app, uri='/folder/test123')
|
||||
|
||||
assert response.text == 'OK'
|
||||
assert results[0] == 'test123'
|
||||
|
||||
def test_dynamic_route_int():
|
||||
app = Sanic('test_dynamic_route_int')
|
||||
|
||||
results = []
|
||||
|
||||
@app.route('/folder/<folder_id:int>')
|
||||
async def handler(request, folder_id):
|
||||
results.append(folder_id)
|
||||
return text('OK')
|
||||
|
||||
request, response = sanic_endpoint_test(app, uri='/folder/12345')
|
||||
assert response.text == 'OK'
|
||||
assert type(results[0]) is int
|
||||
|
||||
request, response = sanic_endpoint_test(app, uri='/folder/asdf')
|
||||
assert response.status == 404
|
||||
|
||||
|
||||
def test_dynamic_route_number():
|
||||
app = Sanic('test_dynamic_route_int')
|
||||
|
||||
results = []
|
||||
|
||||
@app.route('/weight/<weight:number>')
|
||||
async def handler(request, weight):
|
||||
results.append(weight)
|
||||
return text('OK')
|
||||
|
||||
request, response = sanic_endpoint_test(app, uri='/weight/12345')
|
||||
assert response.text == 'OK'
|
||||
assert type(results[0]) is float
|
||||
|
||||
request, response = sanic_endpoint_test(app, uri='/weight/1234.56')
|
||||
assert response.status == 200
|
||||
|
||||
request, response = sanic_endpoint_test(app, uri='/weight/1234-56')
|
||||
assert response.status == 404
|
||||
|
||||
def test_dynamic_route_regex():
|
||||
app = Sanic('test_dynamic_route_int')
|
||||
|
||||
@app.route('/folder/<folder_id:[A-Za-z0-9]{0,4}>')
|
||||
async def handler(request, folder_id):
|
||||
return text('OK')
|
||||
|
||||
request, response = sanic_endpoint_test(app, uri='/folder/test')
|
||||
assert response.status == 200
|
||||
|
||||
request, response = sanic_endpoint_test(app, uri='/folder/test1')
|
||||
assert response.status == 404
|
||||
|
||||
request, response = sanic_endpoint_test(app, uri='/folder/test-123')
|
||||
assert response.status == 404
|
||||
|
||||
request, response = sanic_endpoint_test(app, uri='/folder/')
|
||||
assert response.status == 200
|
||||
54
tests/test_utf8.py
Normal file
54
tests/test_utf8.py
Normal file
@@ -0,0 +1,54 @@
|
||||
from json import loads as json_loads, dumps as json_dumps
|
||||
from sanic import Sanic
|
||||
from sanic.response import json, text
|
||||
from helpers import sanic_endpoint_test
|
||||
|
||||
# ------------------------------------------------------------ #
|
||||
# UTF-8
|
||||
# ------------------------------------------------------------ #
|
||||
|
||||
def test_utf8_query_string():
|
||||
app = Sanic('test_utf8_query_string')
|
||||
|
||||
@app.route('/')
|
||||
async def handler(request):
|
||||
return text('OK')
|
||||
|
||||
request, response = sanic_endpoint_test(app, params=[("utf8", '✓')])
|
||||
assert request.args.get('utf8') == '✓'
|
||||
|
||||
def test_utf8_response():
|
||||
app = Sanic('test_utf8_response')
|
||||
|
||||
@app.route('/')
|
||||
async def handler(request):
|
||||
return text('✓')
|
||||
|
||||
request, response = sanic_endpoint_test(app)
|
||||
assert response.text == '✓'
|
||||
|
||||
def skip_test_utf8_route():
|
||||
app = Sanic('skip_test_utf8_route')
|
||||
|
||||
@app.route('/')
|
||||
async def handler(request):
|
||||
return text('OK')
|
||||
|
||||
# UTF-8 Paths are not supported
|
||||
request, response = sanic_endpoint_test(app, route='/✓', uri='/✓')
|
||||
assert response.text == 'OK'
|
||||
|
||||
def test_utf8_post_json():
|
||||
app = Sanic('test_utf8_post_json')
|
||||
|
||||
@app.route('/')
|
||||
async def handler(request):
|
||||
return text('OK')
|
||||
|
||||
payload = {'test': '✓'}
|
||||
headers = {'content-type': 'application/json'}
|
||||
|
||||
request, response = sanic_endpoint_test(app, data=json_dumps(payload), headers=headers)
|
||||
|
||||
assert request.json.get('test') == '✓'
|
||||
assert response.text == 'OK'
|
||||
Reference in New Issue
Block a user