Source code
Revision control
Copy as Markdown
Other Tools
# -*- coding: utf-8 -*-
"""
test_events.py
~~~~~~~~~~~~~~
Specific tests for any function that is logically self-contained as part of
events.py.
"""
import inspect
import sys
from hypothesis import given
from hypothesis.strategies import (
integers, lists, tuples
)
import pytest
import h2.errors
import h2.events
import h2.settings
# We define a fairly complex Hypothesis strategy here. We want to build a list
# of two tuples of (Setting, value). For Setting we want to make sure we can
# handle settings that the rest of hyper knows nothing about, so we want to
# use integers from 0 to (2**16-1). For values, they're from 0 to (2**32-1).
# Define that strategy here for clarity.
SETTINGS_STRATEGY = lists(
tuples(
integers(min_value=0, max_value=2**16-1),
integers(min_value=0, max_value=2**32-1),
)
)
class TestRemoteSettingsChanged(object):
"""
Validate the function of the RemoteSettingsChanged event.
"""
@given(SETTINGS_STRATEGY)
def test_building_settings_from_scratch(self, settings_list):
"""
Missing old settings are defaulted to None.
"""
settings_dict = dict(settings_list)
e = h2.events.RemoteSettingsChanged.from_settings(
old_settings={},
new_settings=settings_dict,
)
for setting, new_value in settings_dict.items():
assert e.changed_settings[setting].setting == setting
assert e.changed_settings[setting].original_value is None
assert e.changed_settings[setting].new_value == new_value
@given(SETTINGS_STRATEGY, SETTINGS_STRATEGY)
def test_only_reports_changed_settings(self,
old_settings_list,
new_settings_list):
"""
Settings that were not changed are not reported.
"""
old_settings_dict = dict(old_settings_list)
new_settings_dict = dict(new_settings_list)
e = h2.events.RemoteSettingsChanged.from_settings(
old_settings=old_settings_dict,
new_settings=new_settings_dict,
)
assert len(e.changed_settings) == len(new_settings_dict)
assert (
sorted(list(e.changed_settings.keys())) ==
sorted(list(new_settings_dict.keys()))
)
@given(SETTINGS_STRATEGY, SETTINGS_STRATEGY)
def test_correctly_reports_changed_settings(self,
old_settings_list,
new_settings_list):
"""
Settings that are changed are correctly reported.
"""
old_settings_dict = dict(old_settings_list)
new_settings_dict = dict(new_settings_list)
e = h2.events.RemoteSettingsChanged.from_settings(
old_settings=old_settings_dict,
new_settings=new_settings_dict,
)
for setting, new_value in new_settings_dict.items():
original_value = old_settings_dict.get(setting)
assert e.changed_settings[setting].setting == setting
assert e.changed_settings[setting].original_value == original_value
assert e.changed_settings[setting].new_value == new_value
class TestEventReprs(object):
"""
Events have useful representations.
"""
example_request_headers = [
(':authority', 'example.com'),
(':path', '/'),
(':scheme', 'https'),
(':method', 'GET'),
]
example_informational_headers = [
(':status', '100'),
('server', 'fake-serv/0.1.0')
]
example_response_headers = [
(':status', '200'),
('server', 'fake-serv/0.1.0')
]
def test_requestreceived_repr(self):
"""
RequestReceived has a useful debug representation.
"""
e = h2.events.RequestReceived()
e.stream_id = 5
e.headers = self.example_request_headers
assert repr(e) == (
"<RequestReceived stream_id:5, headers:["
"(':authority', 'example.com'), "
"(':path', '/'), "
"(':scheme', 'https'), "
"(':method', 'GET')]>"
)
def test_responsereceived_repr(self):
"""
ResponseReceived has a useful debug representation.
"""
e = h2.events.ResponseReceived()
e.stream_id = 500
e.headers = self.example_response_headers
assert repr(e) == (
"<ResponseReceived stream_id:500, headers:["
"(':status', '200'), "
"('server', 'fake-serv/0.1.0')]>"
)
def test_trailersreceived_repr(self):
"""
TrailersReceived has a useful debug representation.
"""
e = h2.events.TrailersReceived()
e.stream_id = 62
e.headers = self.example_response_headers
assert repr(e) == (
"<TrailersReceived stream_id:62, headers:["
"(':status', '200'), "
"('server', 'fake-serv/0.1.0')]>"
)
def test_informationalresponsereceived_repr(self):
"""
InformationalResponseReceived has a useful debug representation.
"""
e = h2.events.InformationalResponseReceived()
e.stream_id = 62
e.headers = self.example_informational_headers
assert repr(e) == (
"<InformationalResponseReceived stream_id:62, headers:["
"(':status', '100'), "
"('server', 'fake-serv/0.1.0')]>"
)
def test_datareceived_repr(self):
"""
DataReceived has a useful debug representation.
"""
e = h2.events.DataReceived()
e.stream_id = 888
e.data = b"abcdefghijklmnopqrstuvwxyz"
e.flow_controlled_length = 88
assert repr(e) == (
"<DataReceived stream_id:888, flow_controlled_length:88, "
"data:6162636465666768696a6b6c6d6e6f7071727374>"
)
def test_windowupdated_repr(self):
"""
WindowUpdated has a useful debug representation.
"""
e = h2.events.WindowUpdated()
e.stream_id = 0
e.delta = 2**16
assert repr(e) == "<WindowUpdated stream_id:0, delta:65536>"
def test_remotesettingschanged_repr(self):
"""
RemoteSettingsChanged has a useful debug representation.
"""
e = h2.events.RemoteSettingsChanged()
e.changed_settings = {
h2.settings.SettingCodes.INITIAL_WINDOW_SIZE:
h2.settings.ChangedSetting(
h2.settings.SettingCodes.INITIAL_WINDOW_SIZE, 2**16, 2**15
),
}
assert repr(e) == (
"<RemoteSettingsChanged changed_settings:{ChangedSetting("
"setting=SettingCodes.INITIAL_WINDOW_SIZE, original_value=65536, "
"new_value=32768)}>"
)
def test_pingreceived_repr(self):
"""
PingReceived has a useful debug representation.
"""
e = h2.events.PingReceived()
e.ping_data = b'abcdefgh'
assert repr(e) == "<PingReceived ping_data:6162636465666768>"
def test_pingackreceived_repr(self):
"""
PingAckReceived has a useful debug representation.
"""
e = h2.events.PingAckReceived()
e.ping_data = b'abcdefgh'
assert repr(e) == "<PingAckReceived ping_data:6162636465666768>"
def test_streamended_repr(self):
"""
StreamEnded has a useful debug representation.
"""
e = h2.events.StreamEnded()
e.stream_id = 99
assert repr(e) == "<StreamEnded stream_id:99>"
def test_streamreset_repr(self):
"""
StreamEnded has a useful debug representation.
"""
e = h2.events.StreamReset()
e.stream_id = 919
e.error_code = h2.errors.ErrorCodes.ENHANCE_YOUR_CALM
e.remote_reset = False
assert repr(e) == (
"<StreamReset stream_id:919, "
"error_code:ErrorCodes.ENHANCE_YOUR_CALM, remote_reset:False>"
)
def test_pushedstreamreceived_repr(self):
"""
PushedStreamReceived has a useful debug representation.
"""
e = h2.events.PushedStreamReceived()
e.pushed_stream_id = 50
e.parent_stream_id = 11
e.headers = self.example_request_headers
assert repr(e) == (
"<PushedStreamReceived pushed_stream_id:50, parent_stream_id:11, "
"headers:["
"(':authority', 'example.com'), "
"(':path', '/'), "
"(':scheme', 'https'), "
"(':method', 'GET')]>"
)
def test_settingsacknowledged_repr(self):
"""
SettingsAcknowledged has a useful debug representation.
"""
e = h2.events.SettingsAcknowledged()
e.changed_settings = {
h2.settings.SettingCodes.INITIAL_WINDOW_SIZE:
h2.settings.ChangedSetting(
h2.settings.SettingCodes.INITIAL_WINDOW_SIZE, 2**16, 2**15
),
}
assert repr(e) == (
"<SettingsAcknowledged changed_settings:{ChangedSetting("
"setting=SettingCodes.INITIAL_WINDOW_SIZE, original_value=65536, "
"new_value=32768)}>"
)
def test_priorityupdated_repr(self):
"""
PriorityUpdated has a useful debug representation.
"""
e = h2.events.PriorityUpdated()
e.stream_id = 87
e.weight = 32
e.depends_on = 8
e.exclusive = True
assert repr(e) == (
"<PriorityUpdated stream_id:87, weight:32, depends_on:8, "
"exclusive:True>"
)
@pytest.mark.parametrize("additional_data,data_repr", [
(None, "None"),
(b'some data', "736f6d652064617461")
])
def test_connectionterminated_repr(self, additional_data, data_repr):
"""
ConnectionTerminated has a useful debug representation.
"""
e = h2.events.ConnectionTerminated()
e.error_code = h2.errors.ErrorCodes.INADEQUATE_SECURITY
e.last_stream_id = 33
e.additional_data = additional_data
assert repr(e) == (
"<ConnectionTerminated error_code:ErrorCodes.INADEQUATE_SECURITY, "
"last_stream_id:33, additional_data:%s>" % data_repr
)
def test_alternativeserviceavailable_repr(self):
"""
AlternativeServiceAvailable has a useful debug representation.
"""
e = h2.events.AlternativeServiceAvailable()
e.origin = b"example.com"
e.field_value = b'h2=":8000"; ma=60'
assert repr(e) == (
'<AlternativeServiceAvailable origin:example.com, '
'field_value:h2=":8000"; ma=60>'
)
def test_unknownframereceived_repr(self):
"""
UnknownFrameReceived has a useful debug representation.
"""
e = h2.events.UnknownFrameReceived()
assert repr(e) == '<UnknownFrameReceived>'
def all_events():
"""
Generates all the classes (i.e., events) defined in h2.events.
"""
for _, obj in inspect.getmembers(sys.modules['h2.events']):
# We are only interested in objects that are defined in h2.events;
# objects that are imported from other modules are not of interest.
if hasattr(obj, '__module__') and (obj.__module__ != 'h2.events'):
continue
if inspect.isclass(obj):
yield obj
@pytest.mark.parametrize('event', all_events())
def test_all_events_subclass_from_event(event):
"""
Every event defined in h2.events subclasses from h2.events.Event.
"""
assert (event is h2.events.Event) or issubclass(event, h2.events.Event)