python类text()的实例源码

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_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

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'