"""Wire and ASGI contracts; no assertions about native implementation details."""
import asyncio
import contextvars
import socket
import struct
from contextlib import asynccontextmanager
import pytest
from skirr import Server
async def reply(send, body=b'ok', status=200, headers=()):
await send({'type': 'http.response.start', 'status': status, 'headers': headers})
await send({'type': 'http.response.body', 'body': body})
async def echo(scope, receive, send):
chunks = []
while True:
event = await receive()
if event['type'] == 'http.disconnect':
return
chunks.append(event['body'])
if not event['more_body']:
break
await reply(send, b''.join(chunks))
@asynccontextmanager
async def running(app=echo, lifespan='off'):
server = Server(app, port=0, lifespan=lifespan)
address = await server.start()
try:
yield server, address
finally:
await server.close(timeout=.2)
async def response(reader, head=False):
start = await asyncio.wait_for(reader.readline(), 2)
assert start.startswith(b'HTTP/1.'), start
status = int(start.split()[1])
headers = []
while True:
line = await reader.readline()
if line == b'\r\n':
break
assert line, 'EOF in response headers'
n, v = line.rstrip(b'\r\n').split(b':', 1)
headers.append((n.lower(), v.strip()))
h = dict(headers)
if head or status in (204, 304):
return status, headers, b''
if h.get(b'transfer-encoding') == b'chunked':
parts = []
while True:
n = int(await reader.readline(), 16)
if not n:
while True:
trailer = await reader.readline()
assert trailer, 'EOF in response trailers'
if trailer == b'\r\n':
break
break
parts.append(await reader.readexactly(n))
assert await reader.readexactly(2) == b'\r\n'
body = b''.join(parts)
elif b'content-length' in h:
body = await reader.readexactly(int(h[b'content-length']))
else:
body = await reader.read()
return status, headers, body
async def exchange(address, request):
reader, writer = await asyncio.open_connection(*address)
try:
writer.write(request)
await writer.drain()
return await response(reader)
finally:
writer.close()
await writer.wait_closed()
@pytest.mark.parametrize('body', [b'hello', b'x' * 262144])
async def test_fixed_upload_and_pipeline(body):
async with running() as (_, address):
reader, writer = await asyncio.open_connection(*address)
request = b'POST / HTTP/1.1\r\nHost: x\r\nContent-Length: ' + str(len(body)).encode() + b'\r\n\r\n' + body
writer.write(request + b'GET / HTTP/1.1\r\nHost: x\r\n\r\n')
assert (await response(reader))[2] == body
assert (await response(reader))[2] == b''
writer.close()
await writer.wait_closed()
async def test_fragmented_chunked_and_trailers():
request = (b'POST / HTTP/1.1\r\nHost: x\r\nTransfer-Encoding: chunked\r\n\r\n'
b'3;name="v"\r\nabc\r\n2\r\nde\r\n0\r\nX-End: yes\r\n\r\n')
async with running() as (_, address):
reader, writer = await asyncio.open_connection(*address)
for b in request:
writer.write(bytes([b]))
await writer.drain()
await asyncio.sleep(0)
assert (await response(reader))[2] == b'abcde'
writer.close()
await writer.wait_closed()
@pytest.mark.parametrize('wire,code', [
(b'GET / HTTP/1.1\r\n\r\n', 400),
(b'GET / HTTP/1.1\r\nHost: a\r\nHost: b\r\n\r\n', 400),
(b'GET / HTTP/1.1\nHost: x\n\n', 400),
(b'GET / HTTP/1.1\r\nHost: x\n\r\n', 400),
(b'GET / HTTP/1.1\r\nHost: x\r\n\n', 400),
(b'GET / HTTP/1.1\r\nHost: x\r\r\n\r\n', 400),
(b'POST / HTTP/1.1\r\nHost: x\r\nContent-Length: +1\r\n\r\n', 400),
(b'POST / HTTP/1.1\r\nHost: x\r\nContent-Length: 1\r\nContent-Length: 1\r\n\r\n', 400),
(b'POST / HTTP/1.1\r\nHost: x\r\nContent-Length: 1\r\nTransfer-Encoding: chunked\r\n\r\n', 400),
(b'GET / HTTP/1.1\r\nHost : x\r\n\r\n', 400),
(b'GET / HTTP/1.1\r\nHost: x\r\nX: a\r\n b\r\n\r\n', 400),
(b'POST / HTTP/1.1\r\nHost: x\r\nTransfer-Encoding: chunked\r\n\r\nZ\r\n', 400),
(b'POST / HTTP/1.1\r\nHost: x\r\nContent-Length: 99999999\r\n\r\n', 413),
(b'GET / HTTP/1.1\r\nHost: x\r\nX: ' + b'x'*33000 + b'\r\n\r\n', 431),
(b'GET / HTTP/1.1\r\nHost: x\r\n' + b'X: v\r\n'*100 + b'\r\n', 431),
])
async def test_bad_framing_closes_without_processing_next_request(wire, code):
async with running() as (_, address):
reader, writer = await asyncio.open_connection(*address)
writer.write(wire + b'GET / HTTP/1.1\r\nHost: x\r\n\r\n')
assert (await response(reader))[0] == code
assert await asyncio.wait_for(reader.read(), 2) == b''
writer.close()
await writer.wait_closed()
async def test_stream_flushes_and_receive_waits_for_disconnect():
waiting, disconnected = asyncio.Event(), asyncio.Event()
async def app(scope, receive, send):
assert (await receive())['more_body'] is False
waiting.set()
event = await receive()
assert event['type'] == 'http.disconnect'
disconnected.set()
async with running(app) as (_, address):
reader, writer = await asyncio.open_connection(*address)
writer.write(b'GET / HTTP/1.1\r\nHost: x\r\n\r\n')
await asyncio.wait_for(waiting.wait(), 2)
await asyncio.sleep(.03)
assert not disconnected.is_set()
writer.close()
await writer.wait_closed()
await asyncio.wait_for(disconnected.wait(), 2)
proceed = asyncio.Event()
async def stream(scope, receive, send):
await send({'type': 'http.response.start', 'status': 200})
await send({'type': 'http.response.body', 'body': b'first', 'more_body': True})
await proceed.wait()
await send({'type': 'http.response.body', 'body': b'last'})
async with running(stream) as (_, address):
reader, writer = await asyncio.open_connection(*address)
writer.write(b'GET / HTTP/1.1\r\nHost: x\r\n\r\n')
wire = await asyncio.wait_for(reader.readuntil(b'first\r\n'), 2)
assert b'transfer-encoding: chunked' in wire
proceed.set()
assert await asyncio.wait_for(reader.readuntil(b'0\r\n\r\n'), 2) == b'4\r\nlast\r\n0\r\n\r\n'
writer.close()
await writer.wait_closed()
async def test_expect_continue_and_early_rejection():
async with running() as (_, address):
reader, writer = await asyncio.open_connection(*address)
writer.write(b'POST / HTTP/1.1\r\nHost: x\r\nExpect: 100-continue\r\nContent-Length: 3\r\n\r\n')
assert await asyncio.wait_for(reader.readuntil(b'\r\n\r\n'), 2) == b'HTTP/1.1 100 Continue\r\n\r\n'
writer.write(b'abc')
assert (await response(reader))[2] == b'abc'
writer.close()
await writer.wait_closed()
async def reject(scope, receive, send):
await reply(send, b'no', status=403)
async with running(reject) as (_, address):
assert (await exchange(address, b'POST / HTTP/1.1\r\nHost: x\r\nExpect: 100-continue\r\nContent-Length: 100\r\n\r\n'))[0] == 403
async def test_scope_headers_and_percent_decoding():
async def app(scope, receive, send):
assert scope['path'] == '/caf\u00e9/a b'
assert scope['raw_path'] == b'/caf%C3%A9/a%20b'
assert scope['query_string'] == b'q=%FF'
assert [v for n,v in scope['headers'] if n == b'x-a'] == [b'1', b'2']
await reply(send, headers=[(b'set-cookie', b'a=1'), (b'set-cookie', b'b=2')])
async with running(app) as (_, address):
status, headers, body = await exchange(address, b'GET /caf%C3%A9/a%20b?q=%FF HTTP/1.1\r\nHost: x\r\nX-A: 1\r\nX-A: 2\r\n\r\n')
assert status == 200 and body == b'ok'
assert [v for n,v in headers if n == b'set-cookie'] == [b'a=1', b'b=2']
@pytest.mark.parametrize('mode', ['on', 'off'])
async def test_lifespan_and_request_state_isolation(mode):
stopped = asyncio.Event()
async def app(scope, receive, send):
if scope['type'] == 'lifespan':
assert (await receive())['type'] == 'lifespan.startup'
scope['state']['value'] = 7
await send({'type': 'lifespan.startup.complete'})
assert (await receive())['type'] == 'lifespan.shutdown'
stopped.set()
await send({'type': 'lifespan.shutdown.complete'})
else:
state = scope.setdefault('state', {})
assert state.get('value') == (7 if mode == 'on' else None)
state['value'] = 99
assert scope['asgi']['version'] == '3.0'
scope['asgi']['version'] = 'mutated'
await reply(send)
async with running(app, mode) as (_, address):
for _ in range(2):
assert (await exchange(address, b'GET / HTTP/1.1\r\nHost: x\r\n\r\n'))[2] == b'ok'
assert stopped.is_set() == (mode == 'on')
@pytest.mark.parametrize('empty_status', [204, 205, 304])
async def test_response_validation_and_bodyless_pipeline(empty_status):
async def app(scope, receive, send):
if scope['path'] == '/bad':
with pytest.raises(ValueError):
await send({'type':'http.response.start', 'status':200, 'headers':[(b'x', b'a\r\nb: c')]})
with pytest.raises(ValueError):
await send({'type':'http.response.start', 'status':True})
status = empty_status if scope['path'] == '/empty' else 200
await reply(send, b'ok', status, [(b'content-length', b'2')])
async with running(app) as (_, address):
reader, writer = await asyncio.open_connection(*address)
writer.write(b'HEAD / HTTP/1.1\r\nHost: x\r\n\r\nGET /empty HTTP/1.1\r\nHost: x\r\n\r\nGET /bad HTTP/1.1\r\nHost: x\r\n\r\n')
assert (await response(reader, head=True))[0] == 200
assert (await response(reader))[0] == empty_status
assert (await response(reader))[2] == b'ok'
writer.close()
await writer.wait_closed()
async def test_response_trailers():
async def app(scope, receive, send):
await send({'type':'http.response.start', 'status':200, 'trailers':True})
await send({'type':'http.response.body', 'body':b'hello'})
await send({'type':'http.response.trailers', 'headers':[(b'x-checksum',b'123')], 'more_trailers':True})
await send({'type':'http.response.trailers', 'headers':[(b'x-end',b'yes')]})
async with running(app) as (_, address):
reader, writer = await asyncio.open_connection(*address)
writer.write(b'GET / HTTP/1.1\r\nHost: x\r\nTE: trailers\r\nConnection: close\r\n\r\n')
wire = await asyncio.wait_for(reader.read(), 2)
assert b'transfer-encoding: chunked\r\n' in wire
assert wire.endswith(b'5\r\nhello\r\n0\r\nx-checksum: 123\r\nx-end: yes\r\n\r\n')
writer.close()
await writer.wait_closed()
async def test_output_backpressure_then_disconnect_raises():
started, finished, closed = asyncio.Event(), asyncio.Event(), asyncio.Event()
async def app(scope, receive, send):
try:
await send({'type':'http.response.start', 'status':200})
started.set()
for _ in range(256):
await send({'type':'http.response.body', 'body':b'x'*65536, 'more_body':True})
finished.set()
except OSError:
closed.set()
async with running(app) as (_, address):
reader, writer = await asyncio.open_connection(*address, limit=1024)
writer.write(b'GET / HTTP/1.1\r\nHost: x\r\n\r\n')
await asyncio.wait_for(started.wait(), 2)
await asyncio.sleep(.05)
assert not finished.is_set(), 'producer bypassed output backpressure'
writer.transport.abort()
await asyncio.wait_for(closed.wait(), 2)
async def test_retained_channels_cannot_write_into_next_request():
channels = []
async def app(scope, receive, send):
if channels:
old_send, old_receive = channels[0]
with pytest.raises(RuntimeError):
await old_send({'type':'http.response.body', 'body':b'wrong'})
assert (await old_receive())['type'] == 'http.disconnect'
channels.append((send, receive))
await reply(send)
async with running(app) as (_, address):
reader, writer = await asyncio.open_connection(*address)
for _ in range(2):
writer.write(b'GET / HTTP/1.1\r\nHost: x\r\n\r\n')
assert (await response(reader))[2] == b'ok'
writer.close()
await writer.wait_closed()
async def test_fast_stream_cannot_starve_other_connections():
stopped = asyncio.Event()
async def app(scope, receive, send):
if scope['path'] != '/push':
return await reply(send)
try:
await send({'type':'http.response.start', 'status':200})
while True:
await send({'type':'http.response.body', 'body':b'x', 'more_body':True})
except OSError:
pass
finally:
stopped.set()
async with running(app) as (_, address):
reader, writer = await asyncio.open_connection(*address)
writer.write(b'GET /push HTTP/1.1\r\nHost: x\r\n\r\n')
async def consume():
while await reader.read(65536):
pass
consumer = asyncio.create_task(consume())
try:
for _ in range(5):
result = await asyncio.wait_for(exchange(address, b'GET / HTTP/1.1\r\nHost: x\r\n\r\n'), 2)
assert result[2] == b'ok'
finally:
writer.transport.abort()
consumer.cancel()
await asyncio.gather(consumer, return_exceptions=True)
await asyncio.wait_for(stopped.wait(), 2)
@pytest.mark.parametrize('scheme', [b'http', b'HTTP', b'https'])
@pytest.mark.parametrize('target,path,query', [
(b'?q=1', '/', b'q=1'),
(b'', '/', b''),
(b'/caf%C3%A9?x=%FF', '/caf\u00e9', b'x=%FF'),
])
async def test_absolute_target_authority_overrides_host(scheme, target, path, query):
async def app(scope, receive, send):
assert scope['path'] == path
assert scope['query_string'] == query
assert [v for n,v in scope['headers'] if n == b'host'] == [b'correct.example']
await reply(send)
async with running(app) as (_, address):
assert (await exchange(address, b'GET ' + scheme + b'://correct.example' + target + b' HTTP/1.1\r\nHost: wrong.example\r\n\r\n'))[2] == b'ok'
async def test_cancelled_receive_can_be_awaited_again():
waiting, resume = asyncio.Event(), asyncio.Event()
async def app(scope, receive, send):
first = asyncio.ensure_future(receive())
first.cancel()
await asyncio.gather(first, return_exceptions=True)
waiting.set()
await resume.wait()
assert (await receive())['body'] == b'abc'
await reply(send)
async with running(app) as (_, address):
reader, writer = await asyncio.open_connection(*address)
writer.write(b'POST / HTTP/1.1\r\nHost: x\r\nContent-Length: 3\r\n\r\n')
await asyncio.wait_for(waiting.wait(), 2)
writer.write(b'abc')
await writer.drain()
await asyncio.sleep(.03)
resume.set()
assert (await response(reader))[2] == b'ok'
writer.close()
await writer.wait_closed()
async def test_lifespan_failure_prevents_startup():
async def app(scope, receive, send):
assert (await receive())['type'] == 'lifespan.startup'
await send({'type':'lifespan.startup.failed', 'message':'cannot start'})
await asyncio.Event().wait()
server = Server(app, port=0)
with pytest.raises(RuntimeError, match='cannot start'):
await server.start()
async def test_response_header_pairs_can_be_lists_or_iterators():
async def app(scope, receive, send):
await send({'type':'http.response.start', 'status':200,
'headers':iter([[b'x-a', b'1'], iter([b'x-b', b'2'])])})
await send({'type':'http.response.body', 'body':b'ok'})
async with running(app) as (_, address):
status, headers, body = await exchange(address, b'GET / HTTP/1.1\r\nHost: x\r\n\r\n')
assert status == 200 and body == b'ok'
assert (b'x-a', b'1') in headers and (b'x-b', b'2') in headers
async def test_content_length_mismatch_before_and_after_wire_start():
async def app(scope, receive, send):
await send({'type':'http.response.start', 'status':200, 'headers':[(b'content-length',b'3')]})
if scope['path'] == '/late':
await send({'type':'http.response.body', 'body':b'a', 'more_body':True})
await send({'type':'http.response.body', 'body':b'z' if scope['path'] == '/late' else b'zz'})
async with running(app) as (_, address):
assert (await exchange(address, b'GET /early HTTP/1.1\r\nHost: x\r\n\r\n'))[0] == 500
reader, writer = await asyncio.open_connection(*address)
writer.write(b'GET /late HTTP/1.1\r\nHost: x\r\n\r\n')
# One byte already sent plus a final one-byte body would be too short.
wire = await asyncio.wait_for(reader.read(), 2)
assert wire.startswith(b'HTTP/1.1 200') and wire.count(b'HTTP/1.1') == 1 and wire.endswith(b'a')
writer.close()
await writer.wait_closed()
async def test_response_completion_wakes_pending_receive():
completed = asyncio.Event()
async def app(scope, receive, send):
await receive()
listener = asyncio.ensure_future(receive())
await reply(send)
assert (await listener)['type'] == 'http.disconnect'
completed.set()
async with running(app) as (_, address):
reader, writer = await asyncio.open_connection(*address)
writer.write(b'GET / HTTP/1.1\r\nHost: x\r\n\r\n')
assert (await response(reader))[2] == b'ok'
await asyncio.wait_for(completed.wait(), .2)
writer.close()
await writer.wait_closed()
async def test_shutdown_waits_for_cancelled_request_cleanup_before_lifespan():
started, cleaned, shutdown = asyncio.Event(), asyncio.Event(), asyncio.Event()
async def app(scope, receive, send):
if scope['type'] == 'lifespan':
await receive()
await send({'type':'lifespan.startup.complete'})
await receive()
assert cleaned.is_set()
shutdown.set()
await send({'type':'lifespan.shutdown.complete'})
else:
started.set()
try:
await asyncio.Event().wait()
finally:
await asyncio.sleep(.03)
cleaned.set()
server = Server(app, port=0, lifespan='on')
address = await server.start()
reader, writer = await asyncio.open_connection(*address)
writer.write(b'GET / HTTP/1.1\r\nHost: x\r\n\r\n')
await asyncio.wait_for(started.wait(), 2)
try:
await server.close(timeout=0)
assert shutdown.is_set()
finally:
writer.close()
await writer.wait_closed()
async def test_buffered_pipeline_waits_for_active_app_and_preserves_order():
started, resume, marker = asyncio.Event(), asyncio.Event(), asyncio.Event()
observed = []
async def app(scope, receive, send):
if scope['path'] == '/0':
started.set()
await resume.wait()
if scope['path'] == '/1':
asyncio.get_running_loop().call_soon(marker.set)
if scope['path'] == '/199':
observed.append(marker.is_set())
await reply(send, scope['path'].encode())
async with running(app) as (_, address):
reader, writer = await asyncio.open_connection(*address)
writer.write(b'GET /0 HTTP/1.1\r\nHost: x\r\n\r\n')
await asyncio.wait_for(started.wait(), 2)
writer.write(b''.join(b'GET /' + str(i).encode() + b' HTTP/1.1\r\nHost: x\r\n\r\n' for i in range(1, 200)))
await writer.drain()
await asyncio.sleep(.02) # The held app must leave the loop available.
resume.set()
for i in range(200):
assert (await response(reader))[2] == b'/' + str(i).encode()
assert observed == [True], 'buffered requests monopolized the event loop'
writer.close()
await writer.wait_closed()
async def test_request_task_context_isolated_for_ready_and_waiting_apps():
value = contextvars.ContextVar('request-value', default='root')
controller = asyncio.current_task()
async def app(scope, receive, send):
assert asyncio.current_task() is not controller
assert value.get() == 'root'
value.set(scope['path'])
if scope['path'] == '/wait':
await asyncio.sleep(.01)
await reply(send, value.get().encode())
async with running(app) as (_, address):
for path in ('/ready', '/wait', '/ready'):
assert (await exchange(address, b'GET '+path.encode()+b' HTTP/1.1\r\nHost: x\r\n\r\n'))[2] == path.encode()
assert value.get() == 'root'
async def test_custom_task_factory_is_honored():
created = []
async def app(scope, receive, send):
assert asyncio.current_task() in created
await reply(send)
def factory(loop, coro, **kwargs):
task = asyncio.Task(coro, loop=loop, **kwargs)
if getattr(coro, 'cr_code', None) is app.__code__:
created.append(task)
return task
async with running(app) as (_, address):
loop = asyncio.get_running_loop()
old = loop.get_task_factory()
loop.set_task_factory(factory)
try:
for _ in range(2):
assert (await exchange(address, b'GET / HTTP/1.1\r\nHost: x\r\n\r\n'))[2] == b'ok'
assert len(created) == 2
finally:
loop.set_task_factory(old)
@pytest.mark.parametrize('streaming', [False, True])
async def test_reset_during_dispatch_does_not_panic_or_stop_server(streaming):
observed = asyncio.Event()
failures = []
async def app(scope, receive, send):
if scope['path'] == '/abort':
client.setsockopt(socket.SOL_SOCKET, socket.SO_LINGER, struct.pack('ii', 1, 0))
client.close()
try:
if streaming:
await send({'type': 'http.response.start', 'status': 200})
await send({'type': 'http.response.body', 'body': b'first', 'more_body': True})
else:
await reply(send)
except OSError:
observed.set()
else:
await reply(send)
loop = asyncio.get_running_loop()
previous = loop.get_exception_handler()
loop.set_exception_handler(lambda _, context: failures.append(context))
try:
async with running(app) as (_, address):
client = socket.socket()
client.setblocking(False)
await loop.sock_connect(client, address)
await loop.sock_sendall(client, b'GET /abort HTTP/1.1\r\nHost: x\r\n\r\n')
await asyncio.wait_for(observed.wait(), 2)
assert (await exchange(address, b'GET / HTTP/1.1\r\nHost: x\r\n\r\n'))[2] == b'ok'
assert not failures
finally:
loop.set_exception_handler(previous)
@pytest.mark.parametrize('size', [4095, 4096, 4097, 8 * 1024 * 1024])
async def test_streamed_body_exact_across_copy_boundary_and_partial_writes(size):
payload = (bytes(range(256)) * ((size + 255) // 256))[:size]
started, completed = asyncio.Event(), asyncio.Event()
async def app(scope, receive, send):
await send({'type':'http.response.start', 'status':200})
await send({'type':'http.response.body', 'body':b'first', 'more_body':True})
started.set()
await send({'type':'http.response.body', 'body':payload, 'more_body':True})
await send({'type':'http.response.body', 'body':b'last'})
completed.set()
async with running(app) as (_, address):
reader, writer = await asyncio.open_connection(*address, limit=1024)
writer.write(b'GET / HTTP/1.1\r\nHost: x\r\n\r\n')
await asyncio.wait_for(started.wait(), 2)
if size > 1024 * 1024:
await asyncio.sleep(.03)
assert not completed.is_set(), 'large body bypassed backpressure'
assert (await response(reader))[2] == b'first' + payload + b'last'
await asyncio.wait_for(completed.wait(), 2)
writer.close()
await writer.wait_closed()
@pytest.mark.parametrize('duplicates', [2, 96])
async def test_header_boundaries_survive_pipeline_and_scope_retention(duplicates):
"""Header offsets must preserve empty/obs-text values and separate requests."""
saved = []
expected = [(b'host', b'x'), (b'x-empty', b''), (b'x-tab', b'a\tb'),
(b'x-obs', b'\x80\xff')] + [(b'x-dup', b'v')] * duplicates
private = (b'x-private', b'first request only')
async def app(scope, receive, send):
if scope['path'] == '/first':
assert list(scope['headers']) == expected
# Middleware may edit its request's mutable header container. A
# server may instead supply an immutable iterable, which we copy.
if not isinstance(scope['headers'], list):
scope['headers'] = list(scope['headers'])
scope['headers'].append(private)
saved.append(scope)
else:
assert list(saved[0]['headers']) == expected + [private]
assert list(scope['headers']) == (
expected if scope['path'] in ('/same', '/again') else
[(b'host', b'y'), *expected[1:]] if scope['path'] == '/changed' else
[(b'host', b'y'), (b'x-long', b'z' * 12000)])
fields = [[b'X-Mixed', b'value'], [b'DaTe', b'fixed'],
[b'Content-Length', b'2']]
await send({'type': 'http.response.start', 'status': 200, 'headers': fields})
fields[0][1] = b'changed'
await send({'type': 'http.response.body', 'body': b'ok'})
async with running(app) as (_, address):
reader, writer = await asyncio.open_connection(*address)
fields = (b'Host: x\r\nX-Empty:\r\nX-Tab: \ta\tb \t\r\n'
b'X-Obs: \x80\xff\r\n' + b'X-Dup: v\r\n' * duplicates + b'\r\n')
writer.write(b'GET /first HTTP/1.1\r\n' + fields +
b'GET /same HTTP/1.1\r\n' + fields +
b'GET /changed HTTP/1.1\r\n' + fields.replace(b'Host: x', b'Host: y') +
b'GET /again HTTP/1.1\r\n' + fields +
b'GET /second HTTP/1.1\r\nHost: y\r\nX-Long: ' + b'z' * 12000 + b'\r\n\r\n')
for _ in range(5):
status, headers, body = await response(reader)
assert status == 200 and body == b'ok'
assert headers.count((b'x-mixed', b'value')) == 1
assert [v for n, v in headers if n == b'date'] == [b'fixed']
writer.close()
await writer.wait_closed()
async def test_ready_asgi_awaitable_iterator_returns_its_value():
"""An awaitable's iterator must return the receive event via StopIteration."""
async def app(scope, receive, send):
operation = receive()
with pytest.raises(StopIteration) as done:
next(operation.__await__())
assert done.value.value == {'type': 'http.request', 'body': b'', 'more_body': False}
await reply(send)
async with running(app) as (_, address):
assert (await exchange(address, b'GET / HTTP/1.1\r\nHost: x\r\n\r\n'))[2] == b'ok'
async def test_scope_method_and_version_defaults_remain_request_specific():
requests = [(b'GET', b'1.1'), (b'PURGE', b'1.0'), (b'HEAD', b'1.1'),
(b'POST', b'1.0'), (b'GET', b'1.1')]
observed = []
async def app(scope, receive, send):
observed.append((scope['method'], scope['http_version']))
await reply(send)
async with running(app) as (_, address):
reader, writer = await asyncio.open_connection(*address)
for method, version in requests:
writer.write(method + b' / HTTP/' + version +
b'\r\nHost: x\r\nConnection: keep-alive\r\n\r\n')
status, _, body = await response(reader, head=method == b'HEAD')
assert status == 200 and body == (b'' if method == b'HEAD' else b'ok')
writer.close()
await writer.wait_closed()
assert observed == [('GET', '1.1'), ('PURGE', '1.0'), ('HEAD', '1.1'),
('POST', '1.0'), ('GET', '1.1')]
async def test_ready_receive_loop_cooperates_without_socket_backpressure():
marker, finished = asyncio.Event(), asyncio.Event()
observed = []
async def app(scope, receive, send):
await reply(send)
asyncio.get_running_loop().call_soon(marker.set)
for _ in range(5000):
assert (await receive())['type'] == 'http.disconnect'
observed.append(marker.is_set())
finished.set()
async with running(app) as (_, address):
assert (await exchange(address, b'GET / HTTP/1.1\r\nHost: x\r\n\r\n'))[2] == b'ok'
await asyncio.wait_for(finished.wait(), 2)
assert observed == [True]
@pytest.mark.parametrize('waits', [False, True])
async def test_finished_task_failure_and_cancellation_are_reported_correctly(waits):
reports = []
loop = asyncio.get_running_loop()
original = loop.get_exception_handler()
loop.set_exception_handler(lambda _, context: reports.append(context))
async def app(scope, receive, send):
if scope['path'] == '/healthy':
return await reply(send)
if waits:
await asyncio.sleep(.001)
if scope['path'] == '/cancel':
raise asyncio.CancelledError()
if scope['path'] == '/invalid-state':
raise asyncio.InvalidStateError('application error')
if scope['path'] == '/os-error':
raise OSError('application error')
raise ValueError('application error')
try:
async with running(app) as (_, address):
for path in (b'/error', b'/cancel', b'/invalid-state', b'/os-error'):
status, _, body = await exchange(address, b'GET ' + path + b' HTTP/1.1\r\nHost: x\r\n\r\n')
assert status == 500 and body == b''
assert (await exchange(address, b'GET /healthy HTTP/1.1\r\nHost: x\r\n\r\n'))[2] == b'ok'
assert [type(report['exception']) for report in reports] == [ValueError, asyncio.InvalidStateError, OSError]
assert all(str(report['exception']) == 'application error' for report in reports)
assert all(report['exception'].__traceback__ is not None for report in reports)
finally:
loop.set_exception_handler(original)
async def test_completed_responses_flush_for_waiting_apps_on_multiple_connections():
release, started = asyncio.Event(), asyncio.Event()
count = 0
async def app(scope, receive, send):
nonlocal count
count += 1
if count == 24:
started.set()
await release.wait()
await reply(send, scope['path'].encode())
async with running(app) as (_, address):
connections = await asyncio.gather(*(asyncio.open_connection(*address) for _ in range(24)))
try:
for i, (_, writer) in enumerate(connections):
writer.write(b'GET /' + str(i).encode() + b' HTTP/1.1\r\nHost: x\r\n\r\n')
await asyncio.wait_for(started.wait(), 2)
release.set()
replies = await asyncio.gather(*(response(reader) for reader, _ in connections))
assert [(status, body) for status, _, body in replies] == [(200, b'/' + str(i).encode()) for i in range(24)]
finally:
for _, writer in connections:
writer.close()
await asyncio.gather(*(writer.wait_closed() for _, writer in connections))
async def test_server_disconnect_error_is_not_logged_as_application_failure():
started, finished = asyncio.Event(), asyncio.Event()
reports = []
async def app(scope, receive, send):
await receive()
started.set()
try:
assert (await receive())['type'] == 'http.disconnect'
await reply(send)
finally:
finished.set()
loop = asyncio.get_running_loop()
previous = loop.get_exception_handler()
loop.set_exception_handler(lambda _, context: reports.append(context))
try:
async with running(app) as (_, address):
_, writer = await asyncio.open_connection(*address)
writer.write(b'GET / HTTP/1.1\r\nHost: x\r\n\r\n')
await asyncio.wait_for(started.wait(), 2)
writer.close()
await writer.wait_closed()
await asyncio.wait_for(finished.wait(), 2)
assert reports == []
finally:
loop.set_exception_handler(previous)
async def test_send_call_arguments_preserve_validation_and_keyword_support():
async def app(scope, receive, send):
for invoke in (
lambda: send(), lambda: send({}, {}), lambda: send(other={}),
lambda: send({}, msg={}), lambda: send(None),
):
with pytest.raises(TypeError):
invoke()
await send.__call__(msg={'type': 'http.response.start', 'status': 200})
await send(msg={'type': 'http.response.body', 'body': b'first', 'more_body': True})
await send({'type': 'http.response.body', 'body': b'ok'}, **{})
async with running(app) as (_, address):
status, _, body = await exchange(address, b'GET / HTTP/1.1\r\nHost: x\r\n\r\n')
assert status == 200 and body == b'firstok'