h2/test/test_events.py

# -*- 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)