我们从Python开源项目中，提取了以下50个代码示例，用于说明如何使用sanic.response.text()。
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__)
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_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
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.register_blueprint(bp) app.register_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 chapter(request): """ ????????? : content_url ?????U??url????? : url ??????url : novels_name ???? :return: ??????? """ url = request.args.get('url', None) novels_name = request.args.get('novels_name', None) netloc = get_netloc(url) if netloc not in RULES.keys(): return redirect(url) if netloc in REPLACE_RULES.keys(): url = url.replace(REPLACE_RULES[netloc]['old'], REPLACE_RULES[netloc]['new']) content_url = RULES[netloc].content_url content = await cache_owllook_novels_chapter(url=url, netloc=netloc) if content: content = str(content).strip('[],, Jjs').replace(', ', '').replace('onerror', '').replace('js', '').replace( '????', '') return template( 'chapter.html', novels_name=novels_name, url=url, content_url=content_url, soup=content) else: return text('?????????????????????????????????{url}'.format(url=url))
def admin_setting(request): user = request['session'].get('user', None) if user: try: motor_db = motor_base.get_db() data = await motor_db.user.find_one({'user': user}) if data: return template('admin_setting.html', title='{user}??? - owllook'.format(user=user), is_login=1, user=user, register_time=data['register_time'], email=data.get('email', '???????')) else: return text('????') except Exception as e: LOGGER.error(e) return redirect('/') else: return redirect('/')
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 = app.test_client.get( '/', data=json_dumps(payload), headers=headers) assert request.json.get('test') == '?' assert response.text == 'OK'
def test_content_type(): app = Sanic('test_content_type') @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'
def test_remote_addr(): app = Sanic('test_content_type') @app.route('/') async def handler(request): return text(request.remote_addr) 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'
def test_match_info(): app = Sanic('test_match_info') @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"} # ------------------------------------------------------------ # # POST # ------------------------------------------------------------ #
def test_url_attributes_no_ssl(path, query, expected_url): app = Sanic('test_url_attrs_no_ssl') 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
def test_url_attributes_with_ssl(path, query, expected_url): app = Sanic('test_url_attrs_with_ssl') 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
def test_catch_exception_list(): app = Sanic('exception_list') @app.exception([SanicExceptionTestException, NotFound]) def exception_list(request, exception): return text("ok") @app.route('/') def exception(request): raise SanicExceptionTestException("You won't see me") request, response = app.test_client.get('/random') assert response.text == 'ok' request, response = app.test_client.get('/') assert response.text == '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 = app.test_client.get('/') 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 = app.test_client.get('/') assert response.text == 'OK' assert type(results[0]) is Request assert type(results[1]) is Request assert isinstance(results[2], HTTPResponse)
def test_middleware_response_exception(): app = Sanic('test_middleware_response_exception') result = {'status_code': None} @app.middleware('response') async def process_response(request, response): result['status_code'] = response.status return response @app.exception(NotFound) async def error_handler(request, exception): return text('OK', exception.status_code) @app.route('/') async def handler(request): return text('FAIL') request, response = app.test_client.get('/page_not_found') assert response.text == 'OK' assert result['status_code'] == 404
def test_vhosts(): app = Sanic('test_vhosts') @app.route('/', host="example.com") async def handler(request): return text("You're at example.com!") @app.route('/', host="subdomain.example.com") async def handler(request): return text("You're at subdomain.example.com!") headers = {"Host": "example.com"} request, response = app.test_client.get('/', headers=headers) assert response.text == "You're at example.com!" headers = {"Host": "subdomain.example.com"} request, response = app.test_client.get('/', headers=headers) assert response.text == "You're at subdomain.example.com!"
def test_vhosts_with_defaults(): app = Sanic('test_vhosts') @app.route('/', host="hello.com") async def handler(request): return text("Hello, world!") @app.route('/') async def handler(request): return text("default") headers = {"Host": "hello.com"} request, response = app.test_client.get('/', headers=headers) assert response.text == "Hello, world!" request, response = app.test_client.get('/') assert response.text == "default"
def test_create_task(): e = Event() async def coro(): await asyncio.sleep(0.05) e.set() app = Sanic('test_create_task') app.add_task(coro) @app.route('/early') def not_set(request): return text(e.is_set()) @app.route('/late') async def set(request): await asyncio.sleep(0.1) return text(e.is_set()) request, response = app.test_client.get('/early') assert response.body == b'False' request, response = app.test_client.get('/late') assert response.body == b'True'
