django/daphne
1
1
Fork 0
mirror of https://github.com/django/daphne.git synced 2025-07-30 00:49:44 +03:00
Code Issues Packages Projects Releases Wiki Activity

Helper for getting HTTP Response for a given channel message

Browse Source 
To test Daphne's message-to-HTTP part, we need an easy way to fetch the
HTTP response for a given response channel message. I borrowed the
approach from Andrew's existing code. I feel like we might be able to do
with less scaffolding at some point, but didn't have time to
investigate. It's good enough for now.
This commit is contained in:
Maik Hoepfel 2017-03-22 16:55:16 +08:00
parent bfbef8ea96
commit 250f0e30c3
1 changed files with 20 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

24
daphne/tests/factories.py
View File

@ -14,7 +14,23 @@ def message_for_request(method, path, params=None, headers=None, body=None):
that through daphne and returns the emitted channel message.
"""
request = _build_request(method, path, params, headers, body)
return _run_through_daphne(request, 'http.request')
message, factory, transport = _run_through_daphne(request, 'http.request')
return message
def response_for_message(message):
"""
Returns the raw HTTP response that Daphne constructs when sending a reply
to a HTTP request.
The current approach actually first builds a HTTP request (similar to
message_for_request) because we need a valid reply channel. I'm sure
this can be streamlined, but it works for now.
"""
request = _build_request('GET', '/')
request_message, factory, transport = _run_through_daphne(request, 'http.request')
factory.dispatch_reply(request_message['reply_channel'], message)
return transport.value()
def _build_request(method, path, params=None, headers=None, body=None):
@ -78,11 +94,11 @@ def _run_through_daphne(request, channel_name):
channel_layer = ChannelLayer()
factory = HTTPFactory(channel_layer)
proto = factory.buildProtocol(('127.0.0.1', 0))
tr = proto_helpers.StringTransport()
proto.makeConnection(tr)
transport = proto_helpers.StringTransport()
proto.makeConnection(transport)
proto.dataReceived(request)
_, message = channel_layer.receive([channel_name])
return message
return message, factory, transport
def content_length_header(body):