Source code
Revision control
Copy as Markdown
Other Tools
# -*- coding: utf-8 -*-
"""
test_informational_responses
~~~~~~~~~~~~~~~~~~~~~~~~~~
Tests that validate that hyper-h2 correctly handles informational (1XX)
responses in its state machine.
"""
import pytest
import h2.config
import h2.connection
import h2.events
import h2.exceptions
class TestReceivingInformationalResponses(object):
"""
Tests for receiving informational responses.
"""
example_request_headers = [
(b':authority', b'example.com'),
(b':path', b'/'),
(b':scheme', b'https'),
(b':method', b'GET'),
(b'expect', b'100-continue'),
]
example_informational_headers = [
(b':status', b'100'),
(b'server', b'fake-serv/0.1.0')
]
example_response_headers = [
(b':status', b'200'),
(b'server', b'fake-serv/0.1.0')
]
example_trailers = [
(b'trailer', b'you-bet'),
]
@pytest.mark.parametrize('end_stream', (True, False))
def test_single_informational_response(self, frame_factory, end_stream):
"""
When receiving a informational response, the appropriate event is
signaled.
"""
c = h2.connection.H2Connection()
c.initiate_connection()
c.send_headers(
stream_id=1,
headers=self.example_request_headers,
end_stream=end_stream
)
f = frame_factory.build_headers_frame(
headers=self.example_informational_headers,
stream_id=1,
)
events = c.receive_data(f.serialize())
assert len(events) == 1
event = events[0]
assert isinstance(event, h2.events.InformationalResponseReceived)
assert event.headers == self.example_informational_headers
assert event.stream_id == 1
@pytest.mark.parametrize('end_stream', (True, False))
def test_receiving_multiple_header_blocks(self, frame_factory, end_stream):
"""
At least three header blocks can be received: informational, headers,
trailers.
"""
c = h2.connection.H2Connection()
c.initiate_connection()
c.send_headers(
stream_id=1,
headers=self.example_request_headers,
end_stream=end_stream
)
f1 = frame_factory.build_headers_frame(
headers=self.example_informational_headers,
stream_id=1,
)
f2 = frame_factory.build_headers_frame(
headers=self.example_response_headers,
stream_id=1,
)
f3 = frame_factory.build_headers_frame(
headers=self.example_trailers,
stream_id=1,
flags=['END_STREAM'],
)
events = c.receive_data(
f1.serialize() + f2.serialize() + f3.serialize()
)
assert len(events) == 4
assert isinstance(events[0], h2.events.InformationalResponseReceived)
assert events[0].headers == self.example_informational_headers
assert events[0].stream_id == 1
assert isinstance(events[1], h2.events.ResponseReceived)
assert events[1].headers == self.example_response_headers
assert events[1].stream_id == 1
assert isinstance(events[2], h2.events.TrailersReceived)
assert events[2].headers == self.example_trailers
assert events[2].stream_id == 1
@pytest.mark.parametrize('end_stream', (True, False))
def test_receiving_multiple_informational_responses(self,
frame_factory,
end_stream):
"""
More than one informational response is allowed.
"""
c = h2.connection.H2Connection()
c.initiate_connection()
c.send_headers(
stream_id=1,
headers=self.example_request_headers,
end_stream=end_stream
)
f1 = frame_factory.build_headers_frame(
headers=self.example_informational_headers,
stream_id=1,
)
f2 = frame_factory.build_headers_frame(
headers=[(':status', '101')],
stream_id=1,
)
events = c.receive_data(f1.serialize() + f2.serialize())
assert len(events) == 2
assert isinstance(events[0], h2.events.InformationalResponseReceived)
assert events[0].headers == self.example_informational_headers
assert events[0].stream_id == 1
assert isinstance(events[1], h2.events.InformationalResponseReceived)
assert events[1].headers == [(b':status', b'101')]
assert events[1].stream_id == 1
@pytest.mark.parametrize('end_stream', (True, False))
def test_receive_provisional_response_with_end_stream(self,
frame_factory,
end_stream):
"""
Receiving provisional responses with END_STREAM set causes
ProtocolErrors.
"""
c = h2.connection.H2Connection()
c.initiate_connection()
c.send_headers(
stream_id=1,
headers=self.example_request_headers,
end_stream=end_stream
)
c.clear_outbound_data_buffer()
f = frame_factory.build_headers_frame(
headers=self.example_informational_headers,
stream_id=1,
flags=['END_STREAM']
)
with pytest.raises(h2.exceptions.ProtocolError):
c.receive_data(f.serialize())
expected = frame_factory.build_goaway_frame(
last_stream_id=0,
error_code=1,
)
assert c.data_to_send() == expected.serialize()
@pytest.mark.parametrize('end_stream', (True, False))
def test_receiving_out_of_order_headers(self, frame_factory, end_stream):
"""
When receiving a informational response after the actual response
headers we consider it a ProtocolError and raise it.
"""
c = h2.connection.H2Connection()
c.initiate_connection()
c.send_headers(
stream_id=1,
headers=self.example_request_headers,
end_stream=end_stream
)
f1 = frame_factory.build_headers_frame(
headers=self.example_response_headers,
stream_id=1,
)
f2 = frame_factory.build_headers_frame(
headers=self.example_informational_headers,
stream_id=1,
)
c.receive_data(f1.serialize())
c.clear_outbound_data_buffer()
with pytest.raises(h2.exceptions.ProtocolError):
c.receive_data(f2.serialize())
expected = frame_factory.build_goaway_frame(
last_stream_id=0,
error_code=1,
)
assert c.data_to_send() == expected.serialize()
class TestSendingInformationalResponses(object):
"""
Tests for sending informational responses.
"""
example_request_headers = [
(b':authority', b'example.com'),
(b':path', b'/'),
(b':scheme', b'https'),
(b':method', b'GET'),
(b'expect', b'100-continue'),
]
unicode_informational_headers = [
(u':status', u'100'),
(u'server', u'fake-serv/0.1.0')
]
bytes_informational_headers = [
(b':status', b'100'),
(b'server', b'fake-serv/0.1.0')
]
example_response_headers = [
(b':status', b'200'),
(b'server', b'fake-serv/0.1.0')
]
example_trailers = [
(b'trailer', b'you-bet'),
]
server_config = h2.config.H2Configuration(client_side=False)
@pytest.mark.parametrize(
'hdrs', (unicode_informational_headers, bytes_informational_headers),
)
@pytest.mark.parametrize('end_stream', (True, False))
def test_single_informational_response(self,
frame_factory,
hdrs,
end_stream):
"""
When sending a informational response, the appropriate frames are
emitted.
"""
c = h2.connection.H2Connection(config=self.server_config)
c.initiate_connection()
c.receive_data(frame_factory.preamble())
flags = ['END_STREAM'] if end_stream else []
f = frame_factory.build_headers_frame(
headers=self.example_request_headers,
stream_id=1,
flags=flags,
)
c.receive_data(f.serialize())
c.clear_outbound_data_buffer()
frame_factory.refresh_encoder()
c.send_headers(
stream_id=1,
headers=hdrs
)
f = frame_factory.build_headers_frame(
headers=hdrs,
stream_id=1,
)
assert c.data_to_send() == f.serialize()
@pytest.mark.parametrize(
'hdrs', (unicode_informational_headers, bytes_informational_headers),
)
@pytest.mark.parametrize('end_stream', (True, False))
def test_sending_multiple_header_blocks(self,
frame_factory,
hdrs,
end_stream):
"""
At least three header blocks can be sent: informational, headers,
trailers.
"""
c = h2.connection.H2Connection(config=self.server_config)
c.initiate_connection()
c.receive_data(frame_factory.preamble())
flags = ['END_STREAM'] if end_stream else []
f = frame_factory.build_headers_frame(
headers=self.example_request_headers,
stream_id=1,
flags=flags,
)
c.receive_data(f.serialize())
c.clear_outbound_data_buffer()
frame_factory.refresh_encoder()
# Send the three header blocks.
c.send_headers(
stream_id=1,
headers=hdrs
)
c.send_headers(
stream_id=1,
headers=self.example_response_headers
)
c.send_headers(
stream_id=1,
headers=self.example_trailers,
end_stream=True
)
# Check that we sent them properly.
f1 = frame_factory.build_headers_frame(
headers=hdrs,
stream_id=1,
)
f2 = frame_factory.build_headers_frame(
headers=self.example_response_headers,
stream_id=1,
)
f3 = frame_factory.build_headers_frame(
headers=self.example_trailers,
stream_id=1,
flags=['END_STREAM']
)
assert (
c.data_to_send() ==
f1.serialize() + f2.serialize() + f3.serialize()
)
@pytest.mark.parametrize(
'hdrs', (unicode_informational_headers, bytes_informational_headers),
)
@pytest.mark.parametrize('end_stream', (True, False))
def test_sending_multiple_informational_responses(self,
frame_factory,
hdrs,
end_stream):
"""
More than one informational response is allowed.
"""
c = h2.connection.H2Connection(config=self.server_config)
c.initiate_connection()
c.receive_data(frame_factory.preamble())
flags = ['END_STREAM'] if end_stream else []
f = frame_factory.build_headers_frame(
headers=self.example_request_headers,
stream_id=1,
flags=flags,
)
c.receive_data(f.serialize())
c.clear_outbound_data_buffer()
frame_factory.refresh_encoder()
# Send two informational responses.
c.send_headers(
stream_id=1,
headers=hdrs,
)
c.send_headers(
stream_id=1,
headers=[(':status', '101')]
)
# Check we sent them both.
f1 = frame_factory.build_headers_frame(
headers=hdrs,
stream_id=1,
)
f2 = frame_factory.build_headers_frame(
headers=[(':status', '101')],
stream_id=1,
)
assert c.data_to_send() == f1.serialize() + f2.serialize()
@pytest.mark.parametrize(
'hdrs', (unicode_informational_headers, bytes_informational_headers),
)
@pytest.mark.parametrize('end_stream', (True, False))
def test_send_provisional_response_with_end_stream(self,
frame_factory,
hdrs,
end_stream):
"""
Sending provisional responses with END_STREAM set causes
ProtocolErrors.
"""
c = h2.connection.H2Connection(config=self.server_config)
c.initiate_connection()
c.receive_data(frame_factory.preamble())
flags = ['END_STREAM'] if end_stream else []
f = frame_factory.build_headers_frame(
headers=self.example_request_headers,
stream_id=1,
flags=flags,
)
c.receive_data(f.serialize())
with pytest.raises(h2.exceptions.ProtocolError):
c.send_headers(
stream_id=1,
headers=hdrs,
end_stream=True,
)
@pytest.mark.parametrize(
'hdrs', (unicode_informational_headers, bytes_informational_headers),
)
@pytest.mark.parametrize('end_stream', (True, False))
def test_reject_sending_out_of_order_headers(self,
frame_factory,
hdrs,
end_stream):
"""
When sending an informational response after the actual response
headers we consider it a ProtocolError and raise it.
"""
c = h2.connection.H2Connection(config=self.server_config)
c.initiate_connection()
c.receive_data(frame_factory.preamble())
flags = ['END_STREAM'] if end_stream else []
f = frame_factory.build_headers_frame(
headers=self.example_request_headers,
stream_id=1,
flags=flags,
)
c.receive_data(f.serialize())
c.send_headers(
stream_id=1,
headers=self.example_response_headers
)
with pytest.raises(h2.exceptions.ProtocolError):
c.send_headers(
stream_id=1,
headers=hdrs
)