def test_file_upload(app):
app.config['WTF_CSRF_ENABLED'] = False
class TestForm(SanicForm):
upload = FileField('upload file')
submit = SubmitField('Upload')
@app.route('/upload', methods=['GET', 'POST'])
async def upload(request):
form = TestForm(request)
if form.validate_on_submit():
return response.text(form.upload.data.name)
content = render_form(form)
return response.html(content)
req, resp = app.test_client.post(
'/upload', data={'upload': open(__file__, 'rb')})
assert resp.status == 200
assert resp.text == os.path.basename(__file__)
python类text()的实例源码
def test_form_validation(app):
app.config['WTF_CSRF_ENABLED'] = False
class TestForm(SanicForm):
msg = StringField('Note', validators=[DataRequired(), Length(max=10)])
submit = SubmitField('Submit')
@app.route('/', methods=['GET', 'POST'])
async def index(request):
form = TestForm(request)
if request.method == 'POST' and form.validate():
return response.text('validated')
content = render_form(form)
return response.html(content)
req, resp = app.test_client.get('/')
assert resp.status == 200
# we disabled it
assert 'csrf_token' not in resp.text
# this is longer than 10
payload = {'msg': 'love is beautiful'}
req, resp = app.test_client.post('/', data=payload)
assert resp.status == 200
assert 'validated' not in resp.text
payload = {'msg': 'happy'}
req, resp = app.test_client.post('/', data=payload)
assert resp.status == 200
assert 'validated' in resp.text
def test_form_csrf_validation(app):
app.config['WTF_CSRF_SECRET_KEY'] = 'top secret !!!'
class TestForm(SanicForm):
msg = StringField('Note', validators=[DataRequired(), Length(max=10)])
submit = SubmitField('Submit')
@app.route('/', methods=['GET', 'POST'])
async def index(request):
form = TestForm(request)
if request.method == 'POST' and form.validate():
return response.text('validated')
content = render_form(form)
return response.html(content)
req, resp = app.test_client.get('/')
assert resp.status == 200
assert 'csrf_token' in resp.text
token = re.findall(csrf_token_pattern, resp.text)[0]
assert token
payload = {'msg': 'happy', 'csrf_token': token}
req, resp = app.test_client.post('/', data=payload)
assert resp.status == 200
assert 'validated' in resp.text
payload = {'msg': 'happy'}
req, resp = app.test_client.post('/', data=payload)
assert resp.status == 200
# should fail, no CSRF token in payload
assert 'validated' not in resp.text
def test_csrf_token(app):
app.config['WTF_CSRF_SECRET_KEY'] = 'top secret !!!'
app.config['WTF_CSRF_FIELD_NAME'] = 'csrf_token'
class TestForm(SanicForm):
msg = StringField('Note', validators=[DataRequired(), Length(max=10)])
submit = SubmitField('Submit')
@app.route('/', methods=['GET', 'POST'])
async def index(request):
form = TestForm(request)
return response.text(form.csrf_token)
req, resp = app.test_client.get('/')
assert resp.status == 200
assert 'csrf_token' in resp.text
token = re.findall(csrf_token_pattern, resp.text)[0]
assert token
def test_no_request_disable_csrf(app):
app.config['WTF_CSRF_ENABLED'] = True
app.config['WTF_CSRF_SECRET_KEY'] = 'look ma'
class TestForm(SanicForm):
msg = StringField('Note', validators=[DataRequired(), Length(max=10)])
submit = SubmitField('Submit')
@app.route('/', methods=['GET', 'POST'])
async def index(request):
form = TestForm(formdata=request.form)
if request.method == 'POST' and form.validate():
return response.text('validated')
content = render_form(form)
return response.html(content)
payload = {'msg': 'happy'}
req, resp = app.test_client.post('/', data=payload)
assert resp.status == 200
# should be okay, no request means CSRF was disabled
assert 'validated' in resp.text
def test_logout(app):
auth = Auth(app)
@app.post('/login')
async def login(request):
name = request.form.get('name')
password = request.form.get('password')
if name == 'demo' and password == '1234':
auth.login_user(request, User(id=1, name=name))
return response.text('okay')
return response.text('failed')
@app.route('/logout')
async def logout(request):
auth.logout_user(request)
return response.redirect('/user')
@app.route('/user')
async def user(request):
user = auth.current_user(request)
if user is not None:
return response.text(user.name)
return response.text('')
payload = {'name': 'demo', 'password': '1234'}
req, resp = app.test_client.post('/login', data=payload)
assert resp.status == 200 and resp.text == 'okay'
req, resp = app.test_client.get('/user')
assert resp.status == 200 and resp.text == 'demo'
req, reps = app.test_client.get('/logout')
assert resp.status == 200 and resp.text == 'demo'
req, resp = app.test_client.get('/user')
assert resp.status == 200 and resp.text == ''
req, reps = app.test_client.get('/logout')
assert resp.status == 200 and resp.text == ''
req, resp = app.test_client.get('/user')
assert resp.status == 200 and resp.text == ''
def test_login_required(app):
# the default is 'auth.login', change to 'login' to avoid using blueprint
app.config.AUTH_LOGIN_ENDPOINT = 'login'
auth = Auth(app)
@app.post('/login')
async def login(request):
name = request.form.get('name')
password = request.form.get('password')
if name == 'demo' and password == '1234':
auth.login_user(request, User(id=1, name=name))
return response.text('okay')
return response.text('failed')
@app.route('/logout')
@auth.login_required
async def logout(request):
auth.logout_user(request)
return response.redirect('/user')
@app.route('/user')
@auth.login_required(user_keyword='user')
async def user(request, user):
return response.text(user.name)
payload = {'name': 'demo', 'password': '1234'}
req, resp = app.test_client.get('/user', allow_redirects=False)
assert resp.status == 302
assert resp.headers['Location'] == app.url_for('login')
payload = {'name': 'demo', 'password': '1234'}
req, resp = app.test_client.post('/login', data=payload)
assert resp.status == 200 and resp.text == 'okay'
req, resp = app.test_client.get('/user')
assert resp.status == 200 and resp.text == 'demo'
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_form_multipart_form_data():
app = Sanic('test_post_form_multipart_form_data')
@app.route('/')
async def handler(request):
return text('OK')
payload = '------sanic\r\n' \
'Content-Disposition: form-data; name="test"\r\n' \
'\r\n' \
'OK\r\n' \
'------sanic--\r\n'
headers = {'content-type': 'multipart/form-data; boundary=----sanic'}
request, response = sanic_endpoint_test(app, data=payload, headers=headers)
assert request.form.get('test') == 'OK'
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_cookie_options():
app = Sanic('test_text')
@app.route('/')
def handler(request):
response = text("OK")
response.cookies['test'] = 'at you'
response.cookies['test']['httponly'] = True
response.cookies['test']['expires'] = datetime.now() + timedelta(seconds=10)
return response
request, response = sanic_endpoint_test(app)
response_cookies = SimpleCookie()
response_cookies.load(response.headers.get('Set-Cookie', {}))
assert response_cookies['test'].value == 'at you'
assert response_cookies['test']['httponly'] == True
def test_with_middleware():
app = Sanic('test_with_middleware')
class DummyView(HTTPMethodView):
def get(self, request):
return text('I am get method')
app.add_route(DummyView(), '/')
results = []
@app.middleware
async def handler(request):
results.append(request)
request, response = sanic_endpoint_test(app)
assert response.text == 'I am get method'
assert type(results[0]) is Request
def test_static_routes():
app = Sanic('test_dynamic_route')
@app.route('/test')
async def handler1(request):
return text('OK1')
@app.route('/pizazz')
async def handler2(request):
return text('OK2')
request, response = sanic_endpoint_test(app, uri='/test')
assert response.text == 'OK1'
request, response = sanic_endpoint_test(app, uri='/pizazz')
assert response.text == 'OK2'
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'
request, response = sanic_endpoint_test(app, uri='/folder/favicon.ico')
assert response.text == 'OK'
assert results[1] == 'favicon.ico'
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_number')
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_regex')
@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
def test_dynamic_route_unhashable():
app = Sanic('test_dynamic_route_unhashable')
@app.route('/folder/<unhashable:[A-Za-z0-9/]+>/end/')
async def handler(request, unhashable):
return text('OK')
request, response = sanic_endpoint_test(app, uri='/folder/test/asdf/end/')
assert response.status == 200
request, response = sanic_endpoint_test(app, uri='/folder/test///////end/')
assert response.status == 200
request, response = sanic_endpoint_test(app, uri='/folder/test/end/')
assert response.status == 200
request, response = sanic_endpoint_test(app, uri='/folder/test/nope/')
assert response.status == 404
def test_static_add_route():
app = Sanic('test_static_add_route')
async def handler1(request):
return text('OK1')
async def handler2(request):
return text('OK2')
app.add_route(handler1, '/test')
app.add_route(handler2, '/test2')
request, response = sanic_endpoint_test(app, uri='/test')
assert response.text == 'OK1'
request, response = sanic_endpoint_test(app, uri='/test2')
assert response.text == 'OK2'
def test_dynamic_add_route_string():
app = Sanic('test_dynamic_add_route_string')
results = []
async def handler(request, name):
results.append(name)
return text('OK')
app.add_route(handler, '/folder/<name:string>')
request, response = sanic_endpoint_test(app, uri='/folder/test123')
assert response.text == 'OK'
assert results[0] == 'test123'
request, response = sanic_endpoint_test(app, uri='/folder/favicon.ico')
assert response.text == 'OK'
assert results[1] == 'favicon.ico'
def test_dynamic_add_route_int():
app = Sanic('test_dynamic_add_route_int')
results = []
async def handler(request, folder_id):
results.append(folder_id)
return text('OK')
app.add_route(handler, '/folder/<folder_id:int>')
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_add_route_number():
app = Sanic('test_dynamic_add_route_number')
results = []
async def handler(request, weight):
results.append(weight)
return text('OK')
app.add_route(handler, '/weight/<weight:number>')
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_add_route_unhashable():
app = Sanic('test_dynamic_add_route_unhashable')
async def handler(request, unhashable):
return text('OK')
app.add_route(handler, '/folder/<unhashable:[A-Za-z0-9/]+>/end/')
request, response = sanic_endpoint_test(app, uri='/folder/test/asdf/end/')
assert response.status == 200
request, response = sanic_endpoint_test(app, uri='/folder/test///////end/')
assert response.status == 200
request, response = sanic_endpoint_test(app, uri='/folder/test/end/')
assert response.status == 200
request, response = sanic_endpoint_test(app, uri='/folder/test/nope/')
assert response.status == 404
def test_several_bp_with_url_prefix():
app = Sanic('test_text')
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 = sanic_endpoint_test(app, uri='/test1/')
assert response.text == 'Hello'
request, response = sanic_endpoint_test(app, uri='/test2/')
assert response.text == 'Hello2'
def test_bp_middleware():
app = Sanic('test_middleware')
blueprint = Blueprint('test_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 = sanic_endpoint_test(app)
assert response.status == 200
assert response.text == 'OK'
def authenticate(request, *args, **kwargs):
if request.method == 'OPTIONS':
return text('', status=204)
try:
user = await request.app.auth.authenticate(request, *args, **kwargs)
except Exception as e:
raise e
access_token, output = await get_access_token_output(request, user)
if request.app.config.SANIC_JWT_REFRESH_TOKEN_ENABLED:
refresh_token = await request.app.auth.get_refresh_token(user)
output.update({
request.app.config.SANIC_JWT_REFRESH_TOKEN_NAME: refresh_token
})
else:
refresh_token = None
response = get_token_reponse(request, access_token, output, refresh_token)
return response
def refresh(request, *args, **kwargs):
if request.method == 'OPTIONS':
return text('', status=204)
# TODO:
# - Add exceptions
payload = request.app.auth.extract_payload(request, verify=False)
user = request.app.auth.retrieve_user(request, payload=payload)
user_id = request.app.auth._get_user_id(user)
refresh_token = request.app.auth.retrieve_refresh_token(request=request, user_id=user_id)
if isinstance(refresh_token, bytes):
refresh_token = refresh_token.decode('utf-8')
refresh_token = str(refresh_token)
print('user_id: ', user_id)
print('Retrieved token: ', refresh_token)
purported_token = request.app.auth.retrieve_refresh_token_from_request(request)
print('Purported token: ', purported_token)
if refresh_token != purported_token:
raise exceptions.AuthenticationFailed()
access_token, output = await get_access_token_output(request, user)
response = get_token_reponse(request, access_token, output)
return response
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_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
