| # -*- coding: utf-8 -*- |
| """ |
| test_settings |
| ~~~~~~~~~~~~~ |
| |
| Test the Settings object. |
| """ |
| import pytest |
| |
| import h2.errors |
| import h2.exceptions |
| import h2.settings |
| |
| from hypothesis import given, assume |
| from hypothesis.strategies import integers |
| |
| |
| class TestSettings(object): |
| """ |
| Test the Settings object behaves as expected. |
| """ |
| def test_settings_defaults_client(self): |
| """ |
| The Settings object begins with the appropriate defaults for clients. |
| """ |
| s = h2.settings.Settings(client=True) |
| |
| assert s[h2.settings.SettingCodes.HEADER_TABLE_SIZE] == 4096 |
| assert s[h2.settings.SettingCodes.ENABLE_PUSH] == 1 |
| assert s[h2.settings.SettingCodes.INITIAL_WINDOW_SIZE] == 65535 |
| assert s[h2.settings.SettingCodes.MAX_FRAME_SIZE] == 16384 |
| |
| def test_settings_defaults_server(self): |
| """ |
| The Settings object begins with the appropriate defaults for servers. |
| """ |
| s = h2.settings.Settings(client=False) |
| |
| assert s[h2.settings.SettingCodes.HEADER_TABLE_SIZE] == 4096 |
| assert s[h2.settings.SettingCodes.ENABLE_PUSH] == 0 |
| assert s[h2.settings.SettingCodes.INITIAL_WINDOW_SIZE] == 65535 |
| assert s[h2.settings.SettingCodes.MAX_FRAME_SIZE] == 16384 |
| |
| @pytest.mark.parametrize('client', [True, False]) |
| def test_can_set_initial_values(self, client): |
| """ |
| The Settings object can be provided initial values that override the |
| defaults. |
| """ |
| overrides = { |
| h2.settings.SettingCodes.HEADER_TABLE_SIZE: 8080, |
| h2.settings.SettingCodes.MAX_FRAME_SIZE: 16388, |
| h2.settings.SettingCodes.MAX_CONCURRENT_STREAMS: 100, |
| h2.settings.SettingCodes.MAX_HEADER_LIST_SIZE: 2**16, |
| } |
| s = h2.settings.Settings(client=client, initial_values=overrides) |
| |
| assert s[h2.settings.SettingCodes.HEADER_TABLE_SIZE] == 8080 |
| assert s[h2.settings.SettingCodes.ENABLE_PUSH] == bool(client) |
| assert s[h2.settings.SettingCodes.INITIAL_WINDOW_SIZE] == 65535 |
| assert s[h2.settings.SettingCodes.MAX_FRAME_SIZE] == 16388 |
| assert s[h2.settings.SettingCodes.MAX_CONCURRENT_STREAMS] == 100 |
| assert s[h2.settings.SettingCodes.MAX_HEADER_LIST_SIZE] == 2**16 |
| |
| @pytest.mark.parametrize( |
| 'setting,value', |
| [ |
| (h2.settings.SettingCodes.ENABLE_PUSH, 2), |
| (h2.settings.SettingCodes.ENABLE_PUSH, -1), |
| (h2.settings.SettingCodes.INITIAL_WINDOW_SIZE, -1), |
| (h2.settings.SettingCodes.INITIAL_WINDOW_SIZE, 2**34), |
| (h2.settings.SettingCodes.MAX_FRAME_SIZE, 1), |
| (h2.settings.SettingCodes.MAX_FRAME_SIZE, 2**30), |
| (h2.settings.SettingCodes.MAX_HEADER_LIST_SIZE, -1), |
| ] |
| ) |
| def test_cannot_set_invalid_initial_values(self, setting, value): |
| """ |
| The Settings object can be provided initial values that override the |
| defaults. |
| """ |
| overrides = {setting: value} |
| |
| with pytest.raises(h2.exceptions.InvalidSettingsValueError): |
| h2.settings.Settings(initial_values=overrides) |
| |
| def test_applying_value_doesnt_take_effect_immediately(self): |
| """ |
| When a value is applied to the settings object, it doesn't immediately |
| take effect. |
| """ |
| s = h2.settings.Settings(client=True) |
| s[h2.settings.SettingCodes.HEADER_TABLE_SIZE] == 8000 |
| |
| assert s[h2.settings.SettingCodes.HEADER_TABLE_SIZE] == 4096 |
| |
| def test_acknowledging_values(self): |
| """ |
| When we acknowledge settings, the values change. |
| """ |
| s = h2.settings.Settings(client=True) |
| old_settings = dict(s) |
| |
| new_settings = { |
| h2.settings.SettingCodes.HEADER_TABLE_SIZE: 4000, |
| h2.settings.SettingCodes.ENABLE_PUSH: 0, |
| h2.settings.SettingCodes.INITIAL_WINDOW_SIZE: 60, |
| h2.settings.SettingCodes.MAX_FRAME_SIZE: 16385, |
| } |
| s.update(new_settings) |
| |
| assert dict(s) == old_settings |
| s.acknowledge() |
| assert dict(s) == new_settings |
| |
| def test_acknowledging_returns_the_changed_settings(self): |
| """ |
| Acknowledging settings returns the changes. |
| """ |
| s = h2.settings.Settings(client=True) |
| s[h2.settings.SettingCodes.HEADER_TABLE_SIZE] = 8000 |
| s[h2.settings.SettingCodes.ENABLE_PUSH] = 0 |
| |
| changes = s.acknowledge() |
| assert len(changes) == 2 |
| |
| table_size_change = ( |
| changes[h2.settings.SettingCodes.HEADER_TABLE_SIZE] |
| ) |
| push_change = changes[h2.settings.SettingCodes.ENABLE_PUSH] |
| |
| assert table_size_change.setting == ( |
| h2.settings.SettingCodes.HEADER_TABLE_SIZE |
| ) |
| assert table_size_change.original_value == 4096 |
| assert table_size_change.new_value == 8000 |
| |
| assert push_change.setting == h2.settings.SettingCodes.ENABLE_PUSH |
| assert push_change.original_value == 1 |
| assert push_change.new_value == 0 |
| |
| def test_acknowledging_only_returns_changed_settings(self): |
| """ |
| Acknowledging settings does not return unchanged settings. |
| """ |
| s = h2.settings.Settings(client=True) |
| s[h2.settings.SettingCodes.INITIAL_WINDOW_SIZE] = 70 |
| |
| changes = s.acknowledge() |
| assert len(changes) == 1 |
| assert list(changes.keys()) == [ |
| h2.settings.SettingCodes.INITIAL_WINDOW_SIZE |
| ] |
| |
| def test_deleting_values_deletes_all_of_them(self): |
| """ |
| When we delete a key we lose all state about it. |
| """ |
| s = h2.settings.Settings(client=True) |
| s[h2.settings.SettingCodes.HEADER_TABLE_SIZE] == 8000 |
| |
| del s[h2.settings.SettingCodes.HEADER_TABLE_SIZE] |
| |
| with pytest.raises(KeyError): |
| s[h2.settings.SettingCodes.HEADER_TABLE_SIZE] |
| |
| def test_length_correctly_reported(self): |
| """ |
| Length is related only to the number of keys. |
| """ |
| s = h2.settings.Settings(client=True) |
| assert len(s) == 4 |
| |
| s[h2.settings.SettingCodes.HEADER_TABLE_SIZE] == 8000 |
| assert len(s) == 4 |
| |
| s.acknowledge() |
| assert len(s) == 4 |
| |
| del s[h2.settings.SettingCodes.HEADER_TABLE_SIZE] |
| assert len(s) == 3 |
| |
| def test_new_values_work(self): |
| """ |
| New values initially don't appear |
| """ |
| s = h2.settings.Settings(client=True) |
| s[80] = 81 |
| |
| with pytest.raises(KeyError): |
| s[80] |
| |
| def test_new_values_follow_basic_acknowledgement_rules(self): |
| """ |
| A new value properly appears when acknowledged. |
| """ |
| s = h2.settings.Settings(client=True) |
| s[80] = 81 |
| changed_settings = s.acknowledge() |
| |
| assert s[80] == 81 |
| assert len(changed_settings) == 1 |
| |
| changed = changed_settings[80] |
| assert changed.setting == 80 |
| assert changed.original_value is None |
| assert changed.new_value == 81 |
| |
| def test_single_values_arent_affected_by_acknowledgement(self): |
| """ |
| When acknowledged, unchanged settings remain unchanged. |
| """ |
| s = h2.settings.Settings(client=True) |
| assert s[h2.settings.SettingCodes.HEADER_TABLE_SIZE] == 4096 |
| |
| s.acknowledge() |
| assert s[h2.settings.SettingCodes.HEADER_TABLE_SIZE] == 4096 |
| |
| def test_settings_getters(self): |
| """ |
| Getters exist for well-known settings. |
| """ |
| s = h2.settings.Settings(client=True) |
| |
| assert s.header_table_size == ( |
| s[h2.settings.SettingCodes.HEADER_TABLE_SIZE] |
| ) |
| assert s.enable_push == s[h2.settings.SettingCodes.ENABLE_PUSH] |
| assert s.initial_window_size == ( |
| s[h2.settings.SettingCodes.INITIAL_WINDOW_SIZE] |
| ) |
| assert s.max_frame_size == s[h2.settings.SettingCodes.MAX_FRAME_SIZE] |
| assert s.max_concurrent_streams == 2**32 + 1 # A sensible default. |
| assert s.max_header_list_size is None |
| |
| def test_settings_setters(self): |
| """ |
| Setters exist for well-known settings. |
| """ |
| s = h2.settings.Settings(client=True) |
| |
| s.header_table_size = 0 |
| s.enable_push = 1 |
| s.initial_window_size = 2 |
| s.max_frame_size = 16385 |
| s.max_concurrent_streams = 4 |
| s.max_header_list_size = 2**16 |
| |
| s.acknowledge() |
| assert s[h2.settings.SettingCodes.HEADER_TABLE_SIZE] == 0 |
| assert s[h2.settings.SettingCodes.ENABLE_PUSH] == 1 |
| assert s[h2.settings.SettingCodes.INITIAL_WINDOW_SIZE] == 2 |
| assert s[h2.settings.SettingCodes.MAX_FRAME_SIZE] == 16385 |
| assert s[h2.settings.SettingCodes.MAX_CONCURRENT_STREAMS] == 4 |
| assert s[h2.settings.SettingCodes.MAX_HEADER_LIST_SIZE] == 2**16 |
| |
| @given(integers()) |
| def test_cannot_set_invalid_values_for_enable_push(self, val): |
| """ |
| SETTINGS_ENABLE_PUSH only allows two values: 0, 1. |
| """ |
| assume(val not in (0, 1)) |
| s = h2.settings.Settings() |
| |
| with pytest.raises(h2.exceptions.InvalidSettingsValueError) as e: |
| s.enable_push = val |
| |
| s.acknowledge() |
| assert e.value.error_code == h2.errors.ErrorCodes.PROTOCOL_ERROR |
| assert s.enable_push == 1 |
| |
| with pytest.raises(h2.exceptions.InvalidSettingsValueError) as e: |
| s[h2.settings.SettingCodes.ENABLE_PUSH] = val |
| |
| s.acknowledge() |
| assert e.value.error_code == h2.errors.ErrorCodes.PROTOCOL_ERROR |
| assert s[h2.settings.SettingCodes.ENABLE_PUSH] == 1 |
| |
| @given(integers()) |
| def test_cannot_set_invalid_vals_for_initial_window_size(self, val): |
| """ |
| SETTINGS_INITIAL_WINDOW_SIZE only allows values between 0 and 2**32 - 1 |
| inclusive. |
| """ |
| s = h2.settings.Settings() |
| |
| if 0 <= val <= 2**31 - 1: |
| s.initial_window_size = val |
| s.acknowledge() |
| assert s.initial_window_size == val |
| else: |
| with pytest.raises(h2.exceptions.InvalidSettingsValueError) as e: |
| s.initial_window_size = val |
| |
| s.acknowledge() |
| assert ( |
| e.value.error_code == h2.errors.ErrorCodes.FLOW_CONTROL_ERROR |
| ) |
| assert s.initial_window_size == 65535 |
| |
| with pytest.raises(h2.exceptions.InvalidSettingsValueError) as e: |
| s[h2.settings.SettingCodes.INITIAL_WINDOW_SIZE] = val |
| |
| s.acknowledge() |
| assert ( |
| e.value.error_code == h2.errors.ErrorCodes.FLOW_CONTROL_ERROR |
| ) |
| assert s[h2.settings.SettingCodes.INITIAL_WINDOW_SIZE] == 65535 |
| |
| @given(integers()) |
| def test_cannot_set_invalid_values_for_max_frame_size(self, val): |
| """ |
| SETTINGS_MAX_FRAME_SIZE only allows values between 2**14 and 2**24 - 1. |
| """ |
| s = h2.settings.Settings() |
| |
| if 2**14 <= val <= 2**24 - 1: |
| s.max_frame_size = val |
| s.acknowledge() |
| assert s.max_frame_size == val |
| else: |
| with pytest.raises(h2.exceptions.InvalidSettingsValueError) as e: |
| s.max_frame_size = val |
| |
| s.acknowledge() |
| assert e.value.error_code == h2.errors.ErrorCodes.PROTOCOL_ERROR |
| assert s.max_frame_size == 16384 |
| |
| with pytest.raises(h2.exceptions.InvalidSettingsValueError) as e: |
| s[h2.settings.SettingCodes.MAX_FRAME_SIZE] = val |
| |
| s.acknowledge() |
| assert e.value.error_code == h2.errors.ErrorCodes.PROTOCOL_ERROR |
| assert s[h2.settings.SettingCodes.MAX_FRAME_SIZE] == 16384 |
| |
| @given(integers()) |
| def test_cannot_set_invalid_values_for_max_header_list_size(self, val): |
| """ |
| SETTINGS_MAX_HEADER_LIST_SIZE only allows non-negative values. |
| """ |
| s = h2.settings.Settings() |
| |
| if val >= 0: |
| s.max_header_list_size = val |
| s.acknowledge() |
| assert s.max_header_list_size == val |
| else: |
| with pytest.raises(h2.exceptions.InvalidSettingsValueError) as e: |
| s.max_header_list_size = val |
| |
| s.acknowledge() |
| assert e.value.error_code == h2.errors.ErrorCodes.PROTOCOL_ERROR |
| assert s.max_header_list_size is None |
| |
| with pytest.raises(h2.exceptions.InvalidSettingsValueError) as e: |
| s[h2.settings.SettingCodes.MAX_HEADER_LIST_SIZE] = val |
| |
| s.acknowledge() |
| assert e.value.error_code == h2.errors.ErrorCodes.PROTOCOL_ERROR |
| |
| with pytest.raises(KeyError): |
| s[h2.settings.SettingCodes.MAX_HEADER_LIST_SIZE] |
| |
| |
| class TestSettingsEquality(object): |
| """ |
| A class defining tests for the standard implementation of == and != . |
| """ |
| |
| def an_instance(self): |
| """ |
| Return an instance of the class under test. Each call to this method |
| must return a different object. All objects returned must be equal to |
| each other. |
| """ |
| overrides = { |
| h2.settings.SettingCodes.HEADER_TABLE_SIZE: 0, |
| h2.settings.SettingCodes.MAX_FRAME_SIZE: 16384, |
| h2.settings.SettingCodes.MAX_CONCURRENT_STREAMS: 4, |
| h2.settings.SettingCodes.MAX_HEADER_LIST_SIZE: 2**16, |
| } |
| return h2.settings.Settings(client=True, initial_values=overrides) |
| |
| def another_instance(self): |
| """ |
| Return an instance of the class under test. Each call to this method |
| must return a different object. The objects must not be equal to the |
| objects returned by an_instance. They may or may not be equal to |
| each other (they will not be compared against each other). |
| """ |
| overrides = { |
| h2.settings.SettingCodes.HEADER_TABLE_SIZE: 8080, |
| h2.settings.SettingCodes.MAX_FRAME_SIZE: 16388, |
| h2.settings.SettingCodes.MAX_CONCURRENT_STREAMS: 100, |
| h2.settings.SettingCodes.MAX_HEADER_LIST_SIZE: 2**16, |
| } |
| return h2.settings.Settings(client=False, initial_values=overrides) |
| |
| def test_identical_eq(self): |
| """ |
| An object compares equal to itself using the == operator. |
| """ |
| o = self.an_instance() |
| assert (o == o) |
| |
| def test_identical_ne(self): |
| """ |
| An object doesn't compare not equal to itself using the != operator. |
| """ |
| o = self.an_instance() |
| assert not (o != o) |
| |
| def test_same_eq(self): |
| """ |
| Two objects that are equal to each other compare equal to each other |
| using the == operator. |
| """ |
| a = self.an_instance() |
| b = self.an_instance() |
| assert (a == b) |
| |
| def test_same_ne(self): |
| """ |
| Two objects that are equal to each other do not compare not equal to |
| each other using the != operator. |
| """ |
| a = self.an_instance() |
| b = self.an_instance() |
| assert not (a != b) |
| |
| def test_different_eq(self): |
| """ |
| Two objects that are not equal to each other do not compare equal to |
| each other using the == operator. |
| """ |
| a = self.an_instance() |
| b = self.another_instance() |
| assert not (a == b) |
| |
| def test_different_ne(self): |
| """ |
| Two objects that are not equal to each other compare not equal to each |
| other using the != operator. |
| """ |
| a = self.an_instance() |
| b = self.another_instance() |
| assert (a != b) |
| |
| def test_another_type_eq(self): |
| """ |
| The object does not compare equal to an object of an unrelated type |
| (which does not implement the comparison) using the == operator. |
| """ |
| a = self.an_instance() |
| b = object() |
| assert not (a == b) |
| |
| def test_another_type_ne(self): |
| """ |
| The object compares not equal to an object of an unrelated type (which |
| does not implement the comparison) using the != operator. |
| """ |
| a = self.an_instance() |
| b = object() |
| assert (a != b) |
| |
| def test_delegated_eq(self): |
| """ |
| The result of comparison using == is delegated to the right-hand |
| operand if it is of an unrelated type. |
| """ |
| class Delegate(object): |
| def __eq__(self, other): |
| return [self] |
| |
| a = self.an_instance() |
| b = Delegate() |
| assert (a == b) == [b] |
| |
| def test_delegate_ne(self): |
| """ |
| The result of comparison using != is delegated to the right-hand |
| operand if it is of an unrelated type. |
| """ |
| class Delegate(object): |
| def __ne__(self, other): |
| return [self] |
| |
| a = self.an_instance() |
| b = Delegate() |
| assert (a != b) == [b] |