Source code
Revision control
Copy as Markdown
Other Tools
# -*- coding: utf-8 -*-
"""
test_complex_logic
~~~~~~~~~~~~~~~~
More complex tests that try to do more.
Certain tests don't really eliminate incorrect behaviour unless they do quite
a bit. These tests should live here, to keep the pain in once place rather than
hide it in the other parts of the test suite.
"""
import pytest
import h2
import h2.config
import h2.connection
class TestComplexClient(object):
"""
Complex tests for client-side stacks.
"""
example_request_headers = [
(':authority', 'example.com'),
(':path', '/'),
(':scheme', 'https'),
(':method', 'GET'),
]
example_response_headers = [
(':status', '200'),
('server', 'fake-serv/0.1.0')
]
def test_correctly_count_server_streams(self, frame_factory):
"""
We correctly count the number of server streams, both inbound and
outbound.
"""
# This test makes no sense unless you do both inbound and outbound,
# because it's important to confirm that we count them correctly.
c = h2.connection.H2Connection()
c.initiate_connection()
expected_inbound_streams = expected_outbound_streams = 0
assert c.open_inbound_streams == expected_inbound_streams
assert c.open_outbound_streams == expected_outbound_streams
for stream_id in range(1, 15, 2):
# Open an outbound stream
c.send_headers(stream_id, self.example_request_headers)
expected_outbound_streams += 1
assert c.open_inbound_streams == expected_inbound_streams
assert c.open_outbound_streams == expected_outbound_streams
# Receive a pushed stream (to create an inbound one). This doesn't
# open until we also receive headers.
f = frame_factory.build_push_promise_frame(
stream_id=stream_id,
promised_stream_id=stream_id+1,
headers=self.example_request_headers,
)
c.receive_data(f.serialize())
assert c.open_inbound_streams == expected_inbound_streams
assert c.open_outbound_streams == expected_outbound_streams
f = frame_factory.build_headers_frame(
stream_id=stream_id+1,
headers=self.example_response_headers,
)
c.receive_data(f.serialize())
expected_inbound_streams += 1
assert c.open_inbound_streams == expected_inbound_streams
assert c.open_outbound_streams == expected_outbound_streams
for stream_id in range(13, 0, -2):
# Close an outbound stream.
c.end_stream(stream_id)
# Stream doesn't close until both sides close it.
assert c.open_inbound_streams == expected_inbound_streams
assert c.open_outbound_streams == expected_outbound_streams
f = frame_factory.build_headers_frame(
stream_id=stream_id,
headers=self.example_response_headers,
flags=['END_STREAM'],
)
c.receive_data(f.serialize())
expected_outbound_streams -= 1
assert c.open_inbound_streams == expected_inbound_streams
assert c.open_outbound_streams == expected_outbound_streams
# Pushed streams can only be closed remotely.
f = frame_factory.build_data_frame(
stream_id=stream_id+1,
data=b'the content',
flags=['END_STREAM'],
)
c.receive_data(f.serialize())
expected_inbound_streams -= 1
assert c.open_inbound_streams == expected_inbound_streams
assert c.open_outbound_streams == expected_outbound_streams
assert c.open_inbound_streams == 0
assert c.open_outbound_streams == 0
class TestComplexServer(object):
"""
Complex tests for server-side stacks.
"""
example_request_headers = [
(b':authority', b'example.com'),
(b':path', b'/'),
(b':scheme', b'https'),
(b':method', b'GET'),
]
example_response_headers = [
(b':status', b'200'),
(b'server', b'fake-serv/0.1.0')
]
server_config = h2.config.H2Configuration(client_side=False)
def test_correctly_count_server_streams(self, frame_factory):
"""
We correctly count the number of server streams, both inbound and
outbound.
"""
# This test makes no sense unless you do both inbound and outbound,
# because it's important to confirm that we count them correctly.
c = h2.connection.H2Connection(config=self.server_config)
c.receive_data(frame_factory.preamble())
expected_inbound_streams = expected_outbound_streams = 0
assert c.open_inbound_streams == expected_inbound_streams
assert c.open_outbound_streams == expected_outbound_streams
for stream_id in range(1, 15, 2):
# Receive an inbound stream.
f = frame_factory.build_headers_frame(
headers=self.example_request_headers,
stream_id=stream_id,
)
c.receive_data(f.serialize())
expected_inbound_streams += 1
assert c.open_inbound_streams == expected_inbound_streams
assert c.open_outbound_streams == expected_outbound_streams
# Push a stream (to create a outbound one). This doesn't open
# until we send our response headers.
c.push_stream(stream_id, stream_id+1, self.example_request_headers)
assert c.open_inbound_streams == expected_inbound_streams
assert c.open_outbound_streams == expected_outbound_streams
c.send_headers(stream_id+1, self.example_response_headers)
expected_outbound_streams += 1
assert c.open_inbound_streams == expected_inbound_streams
assert c.open_outbound_streams == expected_outbound_streams
for stream_id in range(13, 0, -2):
# Close an inbound stream.
f = frame_factory.build_data_frame(
data=b'',
flags=['END_STREAM'],
stream_id=stream_id,
)
c.receive_data(f.serialize())
# Stream doesn't close until both sides close it.
assert c.open_inbound_streams == expected_inbound_streams
assert c.open_outbound_streams == expected_outbound_streams
c.send_data(stream_id, b'', end_stream=True)
expected_inbound_streams -= 1
assert c.open_inbound_streams == expected_inbound_streams
assert c.open_outbound_streams == expected_outbound_streams
# Pushed streams, however, we can close ourselves.
c.send_data(
stream_id=stream_id+1,
data=b'',
end_stream=True,
)
expected_outbound_streams -= 1
assert c.open_inbound_streams == expected_inbound_streams
assert c.open_outbound_streams == expected_outbound_streams
assert c.open_inbound_streams == 0
assert c.open_outbound_streams == 0
class TestContinuationFrames(object):
"""
Tests for the relatively complex CONTINUATION frame logic.
"""
example_request_headers = [
(b':authority', b'example.com'),
(b':path', b'/'),
(b':scheme', b'https'),
(b':method', b'GET'),
]
server_config = h2.config.H2Configuration(client_side=False)
def _build_continuation_sequence(self, headers, block_size, frame_factory):
f = frame_factory.build_headers_frame(headers)
header_data = f.data
chunks = [
header_data[x:x+block_size]
for x in range(0, len(header_data), block_size)
]
f.data = chunks.pop(0)
frames = [
frame_factory.build_continuation_frame(c) for c in chunks
]
f.flags = {'END_STREAM'}
frames[-1].flags.add('END_HEADERS')
frames.insert(0, f)
return frames
def test_continuation_frame_basic(self, frame_factory):
"""
Test that we correctly decode a header block split across continuation
frames.
"""
c = h2.connection.H2Connection(config=self.server_config)
c.initiate_connection()
c.receive_data(frame_factory.preamble())
frames = self._build_continuation_sequence(
headers=self.example_request_headers,
block_size=5,
frame_factory=frame_factory,
)
data = b''.join(f.serialize() for f in frames)
events = c.receive_data(data)
assert len(events) == 2
first_event, second_event = events
assert isinstance(first_event, h2.events.RequestReceived)
assert first_event.headers == self.example_request_headers
assert first_event.stream_id == 1
assert isinstance(second_event, h2.events.StreamEnded)
assert second_event.stream_id == 1
@pytest.mark.parametrize('stream_id', [3, 1])
def test_continuation_cannot_interleave_headers(self,
frame_factory,
stream_id):
"""
We cannot interleave a new headers block with a CONTINUATION sequence.
"""
c = h2.connection.H2Connection(config=self.server_config)
c.initiate_connection()
c.receive_data(frame_factory.preamble())
c.clear_outbound_data_buffer()
frames = self._build_continuation_sequence(
headers=self.example_request_headers,
block_size=5,
frame_factory=frame_factory,
)
assert len(frames) > 2 # This is mostly defensive.
bogus_frame = frame_factory.build_headers_frame(
headers=self.example_request_headers,
stream_id=stream_id,
flags=['END_STREAM'],
)
frames.insert(len(frames) - 2, bogus_frame)
data = b''.join(f.serialize() for f in frames)
with pytest.raises(h2.exceptions.ProtocolError) as e:
c.receive_data(data)
assert "invalid frame" in str(e.value).lower()
def test_continuation_cannot_interleave_data(self, frame_factory):
"""
We cannot interleave a data frame with a CONTINUATION sequence.
"""
c = h2.connection.H2Connection(config=self.server_config)
c.initiate_connection()
c.receive_data(frame_factory.preamble())
c.clear_outbound_data_buffer()
frames = self._build_continuation_sequence(
headers=self.example_request_headers,
block_size=5,
frame_factory=frame_factory,
)
assert len(frames) > 2 # This is mostly defensive.
bogus_frame = frame_factory.build_data_frame(
data=b'hello',
stream_id=1,
)
frames.insert(len(frames) - 2, bogus_frame)
data = b''.join(f.serialize() for f in frames)
with pytest.raises(h2.exceptions.ProtocolError) as e:
c.receive_data(data)
assert "invalid frame" in str(e.value).lower()
def test_continuation_cannot_interleave_unknown_frame(self, frame_factory):
"""
We cannot interleave an unknown frame with a CONTINUATION sequence.
"""
c = h2.connection.H2Connection(config=self.server_config)
c.initiate_connection()
c.receive_data(frame_factory.preamble())
c.clear_outbound_data_buffer()
frames = self._build_continuation_sequence(
headers=self.example_request_headers,
block_size=5,
frame_factory=frame_factory,
)
assert len(frames) > 2 # This is mostly defensive.
bogus_frame = frame_factory.build_data_frame(
data=b'hello',
stream_id=1,
)
bogus_frame.type = 88
frames.insert(len(frames) - 2, bogus_frame)
data = b''.join(f.serialize() for f in frames)
with pytest.raises(h2.exceptions.ProtocolError) as e:
c.receive_data(data)
assert "invalid frame" in str(e.value).lower()
def test_continuation_frame_multiple_blocks(self, frame_factory):
"""
Test that we correctly decode several header blocks split across
continuation frames.
"""
c = h2.connection.H2Connection(config=self.server_config)
c.initiate_connection()
c.receive_data(frame_factory.preamble())
for stream_id in range(1, 7, 2):
frames = self._build_continuation_sequence(
headers=self.example_request_headers,
block_size=2,
frame_factory=frame_factory,
)
for frame in frames:
frame.stream_id = stream_id
data = b''.join(f.serialize() for f in frames)
events = c.receive_data(data)
assert len(events) == 2
first_event, second_event = events
assert isinstance(first_event, h2.events.RequestReceived)
assert first_event.headers == self.example_request_headers
assert first_event.stream_id == stream_id
assert isinstance(second_event, h2.events.StreamEnded)
assert second_event.stream_id == stream_id
class TestContinuationFramesPushPromise(object):
"""
Tests for the relatively complex CONTINUATION frame logic working with
PUSH_PROMISE frames.
"""
example_request_headers = [
(b':authority', b'example.com'),
(b':path', b'/'),
(b':scheme', b'https'),
(b':method', b'GET'),
]
example_response_headers = [
(b':status', b'200'),
(b'server', b'fake-serv/0.1.0')
]
def _build_continuation_sequence(self, headers, block_size, frame_factory):
f = frame_factory.build_push_promise_frame(
stream_id=1, promised_stream_id=2, headers=headers
)
header_data = f.data
chunks = [
header_data[x:x+block_size]
for x in range(0, len(header_data), block_size)
]
f.data = chunks.pop(0)
frames = [
frame_factory.build_continuation_frame(c) for c in chunks
]
f.flags = {'END_STREAM'}
frames[-1].flags.add('END_HEADERS')
frames.insert(0, f)
return frames
def test_continuation_frame_basic_push_promise(self, frame_factory):
"""
Test that we correctly decode a header block split across continuation
frames when that header block is initiated with a PUSH_PROMISE.
"""
c = h2.connection.H2Connection()
c.initiate_connection()
c.send_headers(stream_id=1, headers=self.example_request_headers)
frames = self._build_continuation_sequence(
headers=self.example_request_headers,
block_size=5,
frame_factory=frame_factory,
)
data = b''.join(f.serialize() for f in frames)
events = c.receive_data(data)
assert len(events) == 1
event = events[0]
assert isinstance(event, h2.events.PushedStreamReceived)
assert event.headers == self.example_request_headers
assert event.parent_stream_id == 1
assert event.pushed_stream_id == 2
@pytest.mark.parametrize('stream_id', [3, 1, 2])
def test_continuation_cannot_interleave_headers_pp(self,
frame_factory,
stream_id):
"""
We cannot interleave a new headers block with a CONTINUATION sequence
when the headers block is based on a PUSH_PROMISE frame.
"""
c = h2.connection.H2Connection()
c.initiate_connection()
c.send_headers(stream_id=1, headers=self.example_request_headers)
frames = self._build_continuation_sequence(
headers=self.example_request_headers,
block_size=5,
frame_factory=frame_factory,
)
assert len(frames) > 2 # This is mostly defensive.
bogus_frame = frame_factory.build_headers_frame(
headers=self.example_response_headers,
stream_id=stream_id,
flags=['END_STREAM'],
)
frames.insert(len(frames) - 2, bogus_frame)
data = b''.join(f.serialize() for f in frames)
with pytest.raises(h2.exceptions.ProtocolError) as e:
c.receive_data(data)
assert "invalid frame" in str(e.value).lower()
def test_continuation_cannot_interleave_data(self, frame_factory):
"""
We cannot interleave a data frame with a CONTINUATION sequence when
that sequence began with a PUSH_PROMISE frame.
"""
c = h2.connection.H2Connection()
c.initiate_connection()
c.send_headers(stream_id=1, headers=self.example_request_headers)
frames = self._build_continuation_sequence(
headers=self.example_request_headers,
block_size=5,
frame_factory=frame_factory,
)
assert len(frames) > 2 # This is mostly defensive.
bogus_frame = frame_factory.build_data_frame(
data=b'hello',
stream_id=1,
)
frames.insert(len(frames) - 2, bogus_frame)
data = b''.join(f.serialize() for f in frames)
with pytest.raises(h2.exceptions.ProtocolError) as e:
c.receive_data(data)
assert "invalid frame" in str(e.value).lower()
def test_continuation_cannot_interleave_unknown_frame(self, frame_factory):
"""
We cannot interleave an unknown frame with a CONTINUATION sequence when
that sequence began with a PUSH_PROMISE frame.
"""
c = h2.connection.H2Connection()
c.initiate_connection()
c.send_headers(stream_id=1, headers=self.example_request_headers)
frames = self._build_continuation_sequence(
headers=self.example_request_headers,
block_size=5,
frame_factory=frame_factory,
)
assert len(frames) > 2 # This is mostly defensive.
bogus_frame = frame_factory.build_data_frame(
data=b'hello',
stream_id=1,
)
bogus_frame.type = 88
frames.insert(len(frames) - 2, bogus_frame)
data = b''.join(f.serialize() for f in frames)
with pytest.raises(h2.exceptions.ProtocolError) as e:
c.receive_data(data)
assert "invalid frame" in str(e.value).lower()
@pytest.mark.parametrize('evict', [True, False])
def test_stream_remotely_closed_disallows_push_promise(self,
evict,
frame_factory):
"""
Streams closed normally by the remote peer disallow PUSH_PROMISE
frames, and cause a GOAWAY.
"""
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(
stream_id=1,
headers=self.example_response_headers,
flags=['END_STREAM']
)
c.receive_data(f.serialize())
c.clear_outbound_data_buffer()
if evict:
# This is annoyingly stateful, but enumerating the list of open
# streams will force us to flush state.
assert not c.open_outbound_streams
f = frame_factory.build_push_promise_frame(
stream_id=1,
promised_stream_id=2,
headers=self.example_request_headers,
)
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.PROTOCOL_ERROR,
)
assert c.data_to_send() == f.serialize()
def test_continuation_frame_multiple_push_promise(self, frame_factory):
"""
Test that we correctly decode header blocks split across continuation
frames when those header block is initiated with a PUSH_PROMISE, for
more than one pushed stream.
"""
c = h2.connection.H2Connection()
c.initiate_connection()
c.send_headers(stream_id=1, headers=self.example_request_headers)
for promised_stream_id in range(2, 8, 2):
frames = self._build_continuation_sequence(
headers=self.example_request_headers,
block_size=2,
frame_factory=frame_factory,
)
frames[0].promised_stream_id = promised_stream_id
data = b''.join(f.serialize() for f in frames)
events = c.receive_data(data)
assert len(events) == 1
event = events[0]
assert isinstance(event, h2.events.PushedStreamReceived)
assert event.headers == self.example_request_headers
assert event.parent_stream_id == 1
assert event.pushed_stream_id == promised_stream_id