commit
d7a873c3fa
54
sanic/utils.py
Normal file
54
sanic/utils.py
Normal file
|
@ -0,0 +1,54 @@
|
|||
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=HOST, port=42101, 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))
|
||||
else:
|
||||
try:
|
||||
return results[0]
|
||||
except:
|
||||
raise ValueError(
|
||||
"request object expected, got ({})".format(results))
|
|
@ -1,47 +0,0 @@
|
|||
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=HOST, port=42101, 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))
|
||||
else:
|
||||
try:
|
||||
return results[0]
|
||||
except:
|
||||
raise ValueError("request object expected, got ({})".format(results))
|
|
@ -1,72 +1,53 @@
|
|||
from json import loads as json_loads, dumps as json_dumps
|
||||
from sanic import Sanic
|
||||
from sanic.response import json, text
|
||||
from sanic.response import text
|
||||
from sanic.exceptions import InvalidUsage, ServerError, NotFound
|
||||
from helpers import sanic_endpoint_test
|
||||
from sanic.utils import sanic_endpoint_test
|
||||
|
||||
# ------------------------------------------------------------ #
|
||||
# GET
|
||||
# ------------------------------------------------------------ #
|
||||
|
||||
def test_exception_response():
|
||||
app = Sanic('test_text')
|
||||
exception_app = Sanic('test_exceptions')
|
||||
|
||||
@app.route('/')
|
||||
def handler(request):
|
||||
return text('OK')
|
||||
|
||||
@app.route('/error')
|
||||
def handler_error(request):
|
||||
raise ServerError("OK")
|
||||
@exception_app.route('/')
|
||||
def handler(request):
|
||||
return text('OK')
|
||||
|
||||
@app.route('/404')
|
||||
def handler_404(request):
|
||||
raise NotFound("OK")
|
||||
|
||||
@app.route('/invalid')
|
||||
def handler_invalid(request):
|
||||
raise InvalidUsage("OK")
|
||||
@exception_app.route('/error')
|
||||
def handler_error(request):
|
||||
raise ServerError("OK")
|
||||
|
||||
request, response = sanic_endpoint_test(app)
|
||||
assert response.status == 200
|
||||
assert response.text == 'OK'
|
||||
|
||||
request, response = sanic_endpoint_test(app, uri='/error')
|
||||
assert response.status == 500
|
||||
@exception_app.route('/404')
|
||||
def handler_404(request):
|
||||
raise NotFound("OK")
|
||||
|
||||
request, response = sanic_endpoint_test(app, uri='/invalid')
|
||||
assert response.status == 400
|
||||
|
||||
request, response = sanic_endpoint_test(app, uri='/404')
|
||||
assert response.status == 404
|
||||
@exception_app.route('/invalid')
|
||||
def handler_invalid(request):
|
||||
raise InvalidUsage("OK")
|
||||
|
||||
def test_exception_handler():
|
||||
app = Sanic('test_text')
|
||||
|
||||
@app.route('/1')
|
||||
def handler_1(request):
|
||||
raise InvalidUsage("OK")
|
||||
@app.route('/2')
|
||||
def handler_2(request):
|
||||
raise ServerError("OK")
|
||||
@app.route('/3')
|
||||
def handler_3(request):
|
||||
raise NotFound("OK")
|
||||
def test_no_exception():
|
||||
request, response = sanic_endpoint_test(exception_app)
|
||||
assert response.status == 200
|
||||
assert response.text == 'OK'
|
||||
|
||||
@app.exception(NotFound, ServerError)
|
||||
def handler_exception(request, exception):
|
||||
return text("OK")
|
||||
|
||||
request, response = sanic_endpoint_test(app, uri='/1')
|
||||
assert response.status == 400
|
||||
def test_server_error_exception():
|
||||
request, response = sanic_endpoint_test(exception_app, uri='/error')
|
||||
assert response.status == 500
|
||||
|
||||
request, response = sanic_endpoint_test(app, uri='/2')
|
||||
assert response.status == 200
|
||||
assert response.text == 'OK'
|
||||
|
||||
request, response = sanic_endpoint_test(app, uri='/3')
|
||||
assert response.status == 200
|
||||
def test_invalid_usage_exception():
|
||||
request, response = sanic_endpoint_test(exception_app, uri='/invalid')
|
||||
assert response.status == 400
|
||||
|
||||
|
||||
def test_not_found_exception():
|
||||
request, response = sanic_endpoint_test(exception_app, uri='/404')
|
||||
assert response.status == 404
|
||||
|
||||
|
||||
request, response = sanic_endpoint_test(app, uri='/random')
|
||||
assert response.status == 200
|
||||
assert response.text == 'OK'
|
49
tests/test_exceptions_handler.py
Normal file
49
tests/test_exceptions_handler.py
Normal file
|
@ -0,0 +1,49 @@
|
|||
from sanic import Sanic
|
||||
from sanic.response import text
|
||||
from sanic.exceptions import InvalidUsage, ServerError, NotFound
|
||||
from sanic.utils import sanic_endpoint_test
|
||||
|
||||
exception_handler_app = Sanic('test_exception_handler')
|
||||
|
||||
|
||||
@exception_handler_app.route('/1')
|
||||
def handler_1(request):
|
||||
raise InvalidUsage("OK")
|
||||
|
||||
|
||||
@exception_handler_app.route('/2')
|
||||
def handler_2(request):
|
||||
raise ServerError("OK")
|
||||
|
||||
|
||||
@exception_handler_app.route('/3')
|
||||
def handler_3(request):
|
||||
raise NotFound("OK")
|
||||
|
||||
|
||||
@exception_handler_app.exception(NotFound, ServerError)
|
||||
def handler_exception(request, exception):
|
||||
return text("OK")
|
||||
|
||||
|
||||
def test_invalid_usage_exception_handler():
|
||||
request, response = sanic_endpoint_test(exception_handler_app, uri='/1')
|
||||
assert response.status == 400
|
||||
|
||||
|
||||
def test_server_error_exception_handler():
|
||||
request, response = sanic_endpoint_test(exception_handler_app, uri='/2')
|
||||
assert response.status == 200
|
||||
assert response.text == 'OK'
|
||||
|
||||
|
||||
def test_not_found_exception_handler():
|
||||
request, response = sanic_endpoint_test(exception_handler_app, uri='/3')
|
||||
assert response.status == 200
|
||||
|
||||
|
||||
def test_text_exception__handler():
|
||||
request, response = sanic_endpoint_test(
|
||||
exception_handler_app, uri='/random')
|
||||
assert response.status == 200
|
||||
assert response.text == 'OK'
|
|
@ -2,7 +2,7 @@ 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
|
||||
from sanic.utils import sanic_endpoint_test
|
||||
|
||||
# ------------------------------------------------------------ #
|
||||
# GET
|
||||
|
|
|
@ -1,7 +1,7 @@
|
|||
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
|
||||
from sanic.utils import sanic_endpoint_test
|
||||
|
||||
# ------------------------------------------------------------ #
|
||||
# GET
|
||||
|
|
|
@ -1,7 +1,7 @@
|
|||
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
|
||||
from sanic.utils import sanic_endpoint_test
|
||||
|
||||
# ------------------------------------------------------------ #
|
||||
# UTF-8
|
||||
|
|
|
@ -1,7 +1,7 @@
|
|||
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
|
||||
from sanic.utils import sanic_endpoint_test
|
||||
|
||||
# ------------------------------------------------------------ #
|
||||
# UTF-8
|
||||
|
|
Loading…
Reference in New Issue
Block a user