Source code
Revision control
Copy as Markdown
Other Tools
# -*- coding: utf-8 -*-
"""
test_priority
~~~~~~~~~~~~~
Test the priority logic of Hyper-h2.
"""
import pytest
import h2.config
import h2.connection
import h2.errors
import h2.events
import h2.exceptions
import h2.stream
class TestPriority(object):
"""
Basic priority tests.
"""
example_request_headers = [
(':authority', 'example.com'),
(':path', '/'),
(':scheme', 'https'),
(':method', 'GET'),
]
example_response_headers = [
(':status', '200'),
('server', 'pytest-h2'),
]
server_config = h2.config.H2Configuration(client_side=False)
def test_receiving_priority_emits_priority_update(self, frame_factory):
"""
Receiving a priority frame emits a PriorityUpdated event.
"""
c = h2.connection.H2Connection(config=self.server_config)
c.initiate_connection()
c.receive_data(frame_factory.preamble())
c.clear_outbound_data_buffer()
f = frame_factory.build_priority_frame(
stream_id=1,
weight=255,
)
events = c.receive_data(f.serialize())
assert len(events) == 1
assert not c.data_to_send()
event = events[0]
assert isinstance(event, h2.events.PriorityUpdated)
assert event.stream_id == 1
assert event.depends_on == 0
assert event.weight == 256
assert event.exclusive is False
def test_headers_with_priority_info(self, frame_factory):
"""
Receiving a HEADERS frame with priority information on it emits a
PriorityUpdated event.
"""
c = h2.connection.H2Connection(config=self.server_config)
c.initiate_connection()
c.receive_data(frame_factory.preamble())
c.clear_outbound_data_buffer()
f = frame_factory.build_headers_frame(
headers=self.example_request_headers,
stream_id=3,
flags=['PRIORITY'],
stream_weight=15,
depends_on=1,
exclusive=True,
)
events = c.receive_data(f.serialize())
assert len(events) == 2
assert not c.data_to_send()
event = events[1]
assert isinstance(event, h2.events.PriorityUpdated)
assert event.stream_id == 3
assert event.depends_on == 1
assert event.weight == 16
assert event.exclusive is True
def test_streams_may_not_depend_on_themselves(self, frame_factory):
"""
A stream adjusted to depend on itself causes a Protocol Error.
"""
c = h2.connection.H2Connection(config=self.server_config)
c.initiate_connection()
c.receive_data(frame_factory.preamble())
c.clear_outbound_data_buffer()
f = frame_factory.build_headers_frame(
headers=self.example_request_headers,
stream_id=3,
flags=['PRIORITY'],
stream_weight=15,
depends_on=1,
exclusive=True,
)
c.receive_data(f.serialize())
c.clear_outbound_data_buffer()
f = frame_factory.build_priority_frame(
stream_id=3,
depends_on=3,
weight=15
)
with pytest.raises(h2.exceptions.ProtocolError):
c.receive_data(f.serialize())
expected_frame = frame_factory.build_goaway_frame(
last_stream_id=3,
error_code=h2.errors.ErrorCodes.PROTOCOL_ERROR,
)
assert c.data_to_send() == expected_frame.serialize()
@pytest.mark.parametrize(
'depends_on,weight,exclusive',
[
(0, 256, False),
(3, 128, False),
(3, 128, True),
]
)
def test_can_prioritize_stream(self, depends_on, weight, exclusive,
frame_factory):
"""
hyper-h2 can emit priority frames.
"""
c = h2.connection.H2Connection()
c.initiate_connection()
c.send_headers(headers=self.example_request_headers, stream_id=1)
c.send_headers(headers=self.example_request_headers, stream_id=3)
c.clear_outbound_data_buffer()
c.prioritize(
stream_id=1,
depends_on=depends_on,
weight=weight,
exclusive=exclusive
)
f = frame_factory.build_priority_frame(
stream_id=1,
weight=weight - 1,
depends_on=depends_on,
exclusive=exclusive,
)
assert c.data_to_send() == f.serialize()
@pytest.mark.parametrize(
'depends_on,weight,exclusive',
[
(0, 256, False),
(1, 128, False),
(1, 128, True),
]
)
def test_emit_headers_with_priority_info(self, depends_on, weight,
exclusive, frame_factory):
"""
It is possible to send a headers frame with priority information on
it.
"""
c = h2.connection.H2Connection()
c.initiate_connection()
c.clear_outbound_data_buffer()
c.send_headers(
headers=self.example_request_headers,
stream_id=3,
priority_weight=weight,
priority_depends_on=depends_on,
priority_exclusive=exclusive,
)
f = frame_factory.build_headers_frame(
headers=self.example_request_headers,
stream_id=3,
flags=['PRIORITY'],
stream_weight=weight - 1,
depends_on=depends_on,
exclusive=exclusive,
)
assert c.data_to_send() == f.serialize()
def test_may_not_prioritize_stream_to_depend_on_self(self, frame_factory):
"""
A stream adjusted to depend on itself causes a Protocol Error.
"""
c = h2.connection.H2Connection()
c.initiate_connection()
c.receive_data(frame_factory.preamble())
c.send_headers(
headers=self.example_request_headers,
stream_id=3,
priority_weight=255,
priority_depends_on=0,
priority_exclusive=False,
)
c.clear_outbound_data_buffer()
with pytest.raises(h2.exceptions.ProtocolError):
c.prioritize(
stream_id=3,
depends_on=3,
)
assert not c.data_to_send()
def test_may_not_initially_set_stream_depend_on_self(self, frame_factory):
"""
A stream that starts by depending on itself causes a Protocol Error.
"""
c = h2.connection.H2Connection()
c.initiate_connection()
c.receive_data(frame_factory.preamble())
c.clear_outbound_data_buffer()
with pytest.raises(h2.exceptions.ProtocolError):
c.send_headers(
headers=self.example_request_headers,
stream_id=3,
priority_depends_on=3,
)
assert not c.data_to_send()
@pytest.mark.parametrize('weight', [0, -15, 257])
def test_prioritize_requires_valid_weight(self, weight):
"""
A call to prioritize with an invalid weight causes a ProtocolError.
"""
c = h2.connection.H2Connection()
c.initiate_connection()
c.clear_outbound_data_buffer()
with pytest.raises(h2.exceptions.ProtocolError):
c.prioritize(stream_id=1, weight=weight)
assert not c.data_to_send()
@pytest.mark.parametrize('weight', [0, -15, 257])
def test_send_headers_requires_valid_weight(self, weight):
"""
A call to send_headers with an invalid weight causes a ProtocolError.
"""
c = h2.connection.H2Connection()
c.initiate_connection()
c.clear_outbound_data_buffer()
with pytest.raises(h2.exceptions.ProtocolError):
c.send_headers(
stream_id=1,
headers=self.example_request_headers,
priority_weight=weight
)
assert not c.data_to_send()
def test_prioritize_defaults(self, frame_factory):
"""
When prioritize() is called with no explicit arguments, it emits a
weight of 16, depending on stream zero non-exclusively.
"""
c = h2.connection.H2Connection()
c.initiate_connection()
c.clear_outbound_data_buffer()
c.prioritize(stream_id=1)
f = frame_factory.build_priority_frame(
stream_id=1,
weight=15,
depends_on=0,
exclusive=False,
)
assert c.data_to_send() == f.serialize()
@pytest.mark.parametrize(
'priority_kwargs',
[
{'priority_weight': 16},
{'priority_depends_on': 0},
{'priority_exclusive': False},
]
)
def test_send_headers_defaults(self, priority_kwargs, frame_factory):
"""
When send_headers() is called with only one explicit argument, it emits
default values for everything else.
"""
c = h2.connection.H2Connection()
c.initiate_connection()
c.clear_outbound_data_buffer()
c.send_headers(
stream_id=1,
headers=self.example_request_headers,
**priority_kwargs
)
f = frame_factory.build_headers_frame(
headers=self.example_request_headers,
stream_id=1,
flags=['PRIORITY'],
stream_weight=15,
depends_on=0,
exclusive=False,
)
assert c.data_to_send() == f.serialize()
def test_servers_cannot_prioritize(self, frame_factory):
"""
Server stacks are not allowed to call ``prioritize()``.
"""
c = h2.connection.H2Connection(config=self.server_config)
c.initiate_connection()
c.receive_data(frame_factory.preamble())
c.clear_outbound_data_buffer()
f = frame_factory.build_headers_frame(
stream_id=1,
headers=self.example_request_headers,
)
c.receive_data(f.serialize())
with pytest.raises(h2.exceptions.RFC1122Error):
c.prioritize(stream_id=1)
def test_servers_cannot_prioritize_with_headers(self, frame_factory):
"""
Server stacks are not allowed to prioritize on headers either.
"""
c = h2.connection.H2Connection(config=self.server_config)
c.initiate_connection()
c.receive_data(frame_factory.preamble())
c.clear_outbound_data_buffer()
f = frame_factory.build_headers_frame(
stream_id=1,
headers=self.example_request_headers,
)
c.receive_data(f.serialize())
with pytest.raises(h2.exceptions.RFC1122Error):
c.send_headers(
stream_id=1,
headers=self.example_response_headers,
priority_weight=16,
)