| # -*- coding: utf-8 -*- |
| """ |
| test_closed_streams |
| ~~~~~~~~~~~~~~~~~~~ |
| |
| Tests that we handle closed streams correctly. |
| """ |
| import pytest |
| |
| import h2.config |
| import h2.connection |
| import h2.errors |
| import h2.events |
| import h2.exceptions |
| |
| |
| class TestClosedStreams(object): |
| example_request_headers = [ |
| (':authority', 'example.com'), |
| (':path', '/'), |
| (':scheme', 'https'), |
| (':method', 'GET'), |
| ] |
| example_response_headers = [ |
| (':status', '200'), |
| ('server', 'fake-serv/0.1.0') |
| ] |
| server_config = h2.config.H2Configuration(client_side=False) |
| |
| def test_can_receive_multiple_rst_stream_frames(self, frame_factory): |
| """ |
| Multiple RST_STREAM frames can be received, either at once or well |
| after one another. Only the first fires an event. |
| """ |
| c = h2.connection.H2Connection() |
| c.initiate_connection() |
| c.send_headers(1, self.example_request_headers, end_stream=True) |
| |
| f = frame_factory.build_rst_stream_frame(stream_id=1) |
| events = c.receive_data(f.serialize() * 3) |
| |
| # Force an iteration over all the streams to remove them. |
| c.open_outbound_streams |
| |
| # Receive more data. |
| events += c.receive_data(f.serialize() * 3) |
| |
| assert len(events) == 1 |
| event = events[0] |
| |
| assert isinstance(event, h2.events.StreamReset) |
| |
| def test_receiving_low_stream_id_causes_goaway(self, frame_factory): |
| """ |
| The remote peer creating a stream with a lower ID than one we've seen |
| causes a GOAWAY frame. |
| """ |
| c = h2.connection.H2Connection(config=self.server_config) |
| c.receive_data(frame_factory.preamble()) |
| c.initiate_connection() |
| |
| f = frame_factory.build_headers_frame( |
| self.example_request_headers, |
| stream_id=3, |
| ) |
| c.receive_data(f.serialize()) |
| c.clear_outbound_data_buffer() |
| |
| f = frame_factory.build_headers_frame( |
| self.example_request_headers, |
| stream_id=1, |
| ) |
| |
| with pytest.raises(h2.exceptions.StreamIDTooLowError) as e: |
| c.receive_data(f.serialize()) |
| |
| assert e.value.stream_id == 1 |
| assert e.value.max_stream_id == 3 |
| |
| f = frame_factory.build_goaway_frame( |
| last_stream_id=3, |
| error_code=h2.errors.ErrorCodes.PROTOCOL_ERROR, |
| ) |
| assert c.data_to_send() == f.serialize() |
| |
| def test_closed_stream_not_present_in_streams_dict(self, frame_factory): |
| """ |
| When streams have been closed, they get removed from the streams |
| dictionary the next time we count the open streams. |
| """ |
| c = h2.connection.H2Connection(config=self.server_config) |
| c.receive_data(frame_factory.preamble()) |
| c.initiate_connection() |
| |
| f = frame_factory.build_headers_frame(self.example_request_headers) |
| c.receive_data(f.serialize()) |
| c.push_stream(1, 2, self.example_request_headers) |
| c.reset_stream(1) |
| c.clear_outbound_data_buffer() |
| |
| f = frame_factory.build_rst_stream_frame(stream_id=2) |
| c.receive_data(f.serialize()) |
| |
| # Force a count of the streams. |
| assert not c.open_outbound_streams |
| |
| # The streams dictionary should be empty. |
| assert not c.streams |
| |
| |
| class TestStreamsClosedByEndStream(object): |
| example_request_headers = [ |
| (':authority', 'example.com'), |
| (':path', '/'), |
| (':scheme', 'https'), |
| (':method', 'GET'), |
| ] |
| example_response_headers = [ |
| (':status', '200'), |
| ('server', 'fake-serv/0.1.0') |
| ] |
| server_config = h2.config.H2Configuration(client_side=False) |
| |
| @pytest.mark.parametrize( |
| "frame", |
| [ |
| lambda self, ff: ff.build_data_frame(b'hello'), |
| lambda self, ff: ff.build_headers_frame( |
| self.example_request_headers, flags=['END_STREAM']), |
| lambda self, ff: ff.build_headers_frame( |
| self.example_request_headers), |
| ] |
| ) |
| @pytest.mark.parametrize("clear_streams", [True, False]) |
| def test_frames_after_recv_end_will_error(self, |
| frame_factory, |
| frame, |
| clear_streams): |
| """ |
| A stream that is closed by receiving END_STREAM raises |
| ProtocolError when it receives an unexpected frame. |
| """ |
| c = h2.connection.H2Connection(config=self.server_config) |
| c.receive_data(frame_factory.preamble()) |
| c.initiate_connection() |
| |
| f = frame_factory.build_headers_frame( |
| self.example_request_headers, flags=['END_STREAM'] |
| ) |
| c.receive_data(f.serialize()) |
| c.send_headers( |
| stream_id=1, |
| headers=self.example_response_headers, |
| end_stream=True |
| ) |
| |
| if clear_streams: |
| # Call open_inbound_streams to force the connection to clean |
| # closed streams. |
| c.open_inbound_streams |
| |
| c.clear_outbound_data_buffer() |
| |
| f = frame(self, frame_factory) |
| with pytest.raises(h2.exceptions.ProtocolError): |
| c.receive_data(f.serialize()) |
| |
| f = frame_factory.build_goaway_frame( |
| last_stream_id=1, |
| error_code=h2.errors.ErrorCodes.STREAM_CLOSED, |
| ) |
| assert c.data_to_send() == f.serialize() |
| |
| @pytest.mark.parametrize( |
| "frame", |
| [ |
| lambda self, ff: ff.build_data_frame(b'hello'), |
| lambda self, ff: ff.build_headers_frame( |
| self.example_response_headers, flags=['END_STREAM']), |
| lambda self, ff: ff.build_headers_frame( |
| self.example_response_headers), |
| ] |
| ) |
| @pytest.mark.parametrize("clear_streams", [True, False]) |
| def test_frames_after_send_end_will_error(self, |
| frame_factory, |
| frame, |
| clear_streams): |
| """ |
| A stream that is closed by sending END_STREAM raises |
| ProtocolError when it receives an unexpected frame. |
| """ |
| c = h2.connection.H2Connection() |
| c.initiate_connection() |
| c.send_headers(stream_id=1, headers=self.example_request_headers, |
| end_stream=True) |
| |
| f = frame_factory.build_headers_frame( |
| self.example_response_headers, flags=['END_STREAM'] |
| ) |
| c.receive_data(f.serialize()) |
| |
| if clear_streams: |
| # Call open_outbound_streams to force the connection to clean |
| # closed streams. |
| c.open_outbound_streams |
| |
| c.clear_outbound_data_buffer() |
| |
| f = frame(self, frame_factory) |
| with pytest.raises(h2.exceptions.ProtocolError): |
| c.receive_data(f.serialize()) |
| |
| f = frame_factory.build_goaway_frame( |
| last_stream_id=0, |
| error_code=h2.errors.ErrorCodes.STREAM_CLOSED, |
| ) |
| assert c.data_to_send() == f.serialize() |
| |
| @pytest.mark.parametrize( |
| "frame", |
| [ |
| lambda self, ff: ff.build_window_update_frame(1, 1), |
| lambda self, ff: ff.build_rst_stream_frame(1) |
| ] |
| ) |
| def test_frames_after_send_end_will_be_ignored(self, |
| frame_factory, |
| frame): |
| """ |
| A stream that is closed by sending END_STREAM will raise |
| ProtocolError when received unexpected frame. |
| """ |
| c = h2.connection.H2Connection(config=self.server_config) |
| c.receive_data(frame_factory.preamble()) |
| c.initiate_connection() |
| |
| f = frame_factory.build_headers_frame( |
| self.example_request_headers, flags=['END_STREAM'] |
| ) |
| c.receive_data(f.serialize()) |
| c.send_headers( |
| stream_id=1, |
| headers=self.example_response_headers, |
| end_stream=True |
| ) |
| |
| c.clear_outbound_data_buffer() |
| |
| f = frame(self, frame_factory) |
| events = c.receive_data(f.serialize()) |
| |
| assert not events |
| |
| |
| class TestStreamsClosedByRstStream(object): |
| example_request_headers = [ |
| (':authority', 'example.com'), |
| (':path', '/'), |
| (':scheme', 'https'), |
| (':method', 'GET'), |
| ] |
| example_response_headers = [ |
| (':status', '200'), |
| ('server', 'fake-serv/0.1.0') |
| ] |
| server_config = h2.config.H2Configuration(client_side=False) |
| |
| @pytest.mark.parametrize( |
| "frame", |
| [ |
| lambda self, ff: ff.build_headers_frame( |
| self.example_request_headers), |
| lambda self, ff: ff.build_headers_frame( |
| self.example_request_headers, flags=['END_STREAM']), |
| lambda self, ff: ff.build_data_frame(b'hello'), |
| lambda self, ff: ff.build_window_update_frame(1, 1), |
| ] |
| ) |
| def test_resets_further_frames_after_recv_reset(self, |
| frame_factory, |
| frame): |
| """ |
| A stream that is closed by receive RST_STREAM can receive further |
| frames: it simply sends RST_STREAM for it. |
| """ |
| c = h2.connection.H2Connection(config=self.server_config) |
| c.receive_data(frame_factory.preamble()) |
| c.initiate_connection() |
| |
| header_frame = frame_factory.build_headers_frame( |
| self.example_request_headers, flags=['END_STREAM'] |
| ) |
| c.receive_data(header_frame.serialize()) |
| |
| c.send_headers( |
| stream_id=1, |
| headers=self.example_response_headers, |
| end_stream=False |
| ) |
| |
| rst_frame = frame_factory.build_rst_stream_frame( |
| 1, h2.errors.ErrorCodes.STREAM_CLOSED |
| ) |
| c.receive_data(rst_frame.serialize()) |
| c.clear_outbound_data_buffer() |
| |
| f = frame(self, frame_factory) |
| events = c.receive_data(f.serialize()) |
| |
| rst_frame = frame_factory.build_rst_stream_frame( |
| 1, h2.errors.ErrorCodes.STREAM_CLOSED |
| ) |
| assert not events |
| assert c.data_to_send() == rst_frame.serialize() |
| |
| events = c.receive_data(f.serialize() * 3) |
| assert not events |
| assert c.data_to_send() == rst_frame.serialize() * 3 |
| |
| # Iterate over the streams to make sure it's gone, then confirm the |
| # behaviour is unchanged. |
| c.open_outbound_streams |
| |
| events = c.receive_data(f.serialize() * 3) |
| assert not events |
| assert c.data_to_send() == rst_frame.serialize() * 3 |
| |
| @pytest.mark.parametrize( |
| "frame", |
| [ |
| lambda self, ff: ff.build_headers_frame( |
| self.example_request_headers), |
| lambda self, ff: ff.build_headers_frame( |
| self.example_request_headers, flags=['END_STREAM']), |
| lambda self, ff: ff.build_data_frame(b'hello'), |
| lambda self, ff: ff.build_window_update_frame(1, 1), |
| ] |
| ) |
| def test_resets_further_frames_after_send_reset(self, |
| frame_factory, |
| frame): |
| """ |
| A stream that is closed by sent RST_STREAM can receive further frames: |
| it simply sends RST_STREAM for it. |
| """ |
| c = h2.connection.H2Connection(config=self.server_config) |
| c.receive_data(frame_factory.preamble()) |
| c.initiate_connection() |
| |
| header_frame = frame_factory.build_headers_frame( |
| self.example_request_headers, flags=['END_STREAM'] |
| ) |
| c.receive_data(header_frame.serialize()) |
| |
| c.send_headers( |
| stream_id=1, |
| headers=self.example_response_headers, |
| end_stream=False |
| ) |
| |
| c.reset_stream(1, h2.errors.ErrorCodes.INTERNAL_ERROR) |
| |
| rst_frame = frame_factory.build_rst_stream_frame( |
| 1, h2.errors.ErrorCodes.STREAM_CLOSED |
| ) |
| c.clear_outbound_data_buffer() |
| |
| f = frame(self, frame_factory) |
| events = c.receive_data(f.serialize()) |
| |
| rst_frame = frame_factory.build_rst_stream_frame( |
| 1, h2.errors.ErrorCodes.STREAM_CLOSED |
| ) |
| assert not events |
| assert c.data_to_send() == rst_frame.serialize() |
| |
| events = c.receive_data(f.serialize() * 3) |
| assert not events |
| assert c.data_to_send() == rst_frame.serialize() * 3 |
| |
| # Iterate over the streams to make sure it's gone, then confirm the |
| # behaviour is unchanged. |
| c.open_outbound_streams |
| |
| events = c.receive_data(f.serialize() * 3) |
| assert not events |
| assert c.data_to_send() == rst_frame.serialize() * 3 |
