Add test and example for custom protocol
This commit is contained in:
parent
2d05243c4a
commit
ac44900fc4
24
examples/custom_protocol.py
Normal file
24
examples/custom_protocol.py
Normal file
|
@ -0,0 +1,24 @@
|
|||
from sanic import Sanic
|
||||
from sanic.server import HttpProtocol
|
||||
from sanic.response import text
|
||||
|
||||
app = Sanic(__name__)
|
||||
|
||||
|
||||
class CustomHttpProtocol(HttpProtocol):
|
||||
|
||||
def write_response(self, response):
|
||||
if isinstance(response, str):
|
||||
response = text(response)
|
||||
self.transport.write(
|
||||
response.output(self.request.version)
|
||||
)
|
||||
self.transport.close()
|
||||
|
||||
|
||||
@app.route("/")
|
||||
async def test(request):
|
||||
return 'Hello, world!'
|
||||
|
||||
|
||||
app.run(host="0.0.0.0", port=8000, protocol=CustomHttpProtocol)
|
|
@ -16,8 +16,8 @@ async def local_request(method, uri, cookies=None, *args, **kwargs):
|
|||
|
||||
|
||||
def sanic_endpoint_test(app, method='get', uri='/', gather_request=True,
|
||||
loop=None, debug=False, *request_args,
|
||||
**request_kwargs):
|
||||
loop=None, debug=False, server_kwargs={},
|
||||
*request_args, **request_kwargs):
|
||||
results = []
|
||||
exceptions = []
|
||||
|
||||
|
@ -36,7 +36,7 @@ def sanic_endpoint_test(app, method='get', uri='/', gather_request=True,
|
|||
app.stop()
|
||||
|
||||
app.run(host=HOST, debug=debug, port=42101,
|
||||
after_start=_collect_response, loop=loop)
|
||||
after_start=_collect_response, loop=loop, **server_kwargs)
|
||||
|
||||
if exceptions:
|
||||
raise ValueError("Exception during request: {}".format(exceptions))
|
||||
|
|
32
tests/test_custom_protocol.py
Normal file
32
tests/test_custom_protocol.py
Normal file
|
@ -0,0 +1,32 @@
|
|||
from sanic import Sanic
|
||||
from sanic.server import HttpProtocol
|
||||
from sanic.response import text
|
||||
from sanic.utils import sanic_endpoint_test
|
||||
|
||||
app = Sanic('test_custom_porotocol')
|
||||
|
||||
|
||||
class CustomHttpProtocol(HttpProtocol):
|
||||
|
||||
def write_response(self, response):
|
||||
if isinstance(response, str):
|
||||
response = text(response)
|
||||
self.transport.write(
|
||||
response.output(self.request.version)
|
||||
)
|
||||
self.transport.close()
|
||||
|
||||
|
||||
@app.route('/1')
|
||||
async def handler_1(request):
|
||||
return 'OK'
|
||||
|
||||
|
||||
def test_use_custom_protocol():
|
||||
server_kwargs = {
|
||||
'protocol': CustomHttpProtocol
|
||||
}
|
||||
request, response = sanic_endpoint_test(app, uri='/1',
|
||||
server_kwargs=server_kwargs)
|
||||
assert response.status == 200
|
||||
assert response.text == 'OK'
|
Loading…
Reference in New Issue
Block a user