test_settings.py (16680B)
1 # -*- coding: utf-8 -*- 2 """ 3 test_settings 4 ~~~~~~~~~~~~~ 5 6 Test the Settings object. 7 """ 8 import pytest 9 10 import h2.errors 11 import h2.exceptions 12 import h2.settings 13 14 from hypothesis import given, assume 15 from hypothesis.strategies import ( 16 integers, booleans, fixed_dictionaries, builds 17 ) 18 19 20 class TestSettings(object): 21 """ 22 Test the Settings object behaves as expected. 23 """ 24 def test_settings_defaults_client(self): 25 """ 26 The Settings object begins with the appropriate defaults for clients. 27 """ 28 s = h2.settings.Settings(client=True) 29 30 assert s[h2.settings.SettingCodes.HEADER_TABLE_SIZE] == 4096 31 assert s[h2.settings.SettingCodes.ENABLE_PUSH] == 1 32 assert s[h2.settings.SettingCodes.INITIAL_WINDOW_SIZE] == 65535 33 assert s[h2.settings.SettingCodes.MAX_FRAME_SIZE] == 16384 34 assert s[h2.settings.SettingCodes.ENABLE_CONNECT_PROTOCOL] == 0 35 36 def test_settings_defaults_server(self): 37 """ 38 The Settings object begins with the appropriate defaults for servers. 39 """ 40 s = h2.settings.Settings(client=False) 41 42 assert s[h2.settings.SettingCodes.HEADER_TABLE_SIZE] == 4096 43 assert s[h2.settings.SettingCodes.ENABLE_PUSH] == 0 44 assert s[h2.settings.SettingCodes.INITIAL_WINDOW_SIZE] == 65535 45 assert s[h2.settings.SettingCodes.MAX_FRAME_SIZE] == 16384 46 assert s[h2.settings.SettingCodes.ENABLE_CONNECT_PROTOCOL] == 0 47 48 @pytest.mark.parametrize('client', [True, False]) 49 def test_can_set_initial_values(self, client): 50 """ 51 The Settings object can be provided initial values that override the 52 defaults. 53 """ 54 overrides = { 55 h2.settings.SettingCodes.HEADER_TABLE_SIZE: 8080, 56 h2.settings.SettingCodes.MAX_FRAME_SIZE: 16388, 57 h2.settings.SettingCodes.MAX_CONCURRENT_STREAMS: 100, 58 h2.settings.SettingCodes.MAX_HEADER_LIST_SIZE: 2**16, 59 h2.settings.SettingCodes.ENABLE_CONNECT_PROTOCOL: 1, 60 } 61 s = h2.settings.Settings(client=client, initial_values=overrides) 62 63 assert s[h2.settings.SettingCodes.HEADER_TABLE_SIZE] == 8080 64 assert s[h2.settings.SettingCodes.ENABLE_PUSH] == bool(client) 65 assert s[h2.settings.SettingCodes.INITIAL_WINDOW_SIZE] == 65535 66 assert s[h2.settings.SettingCodes.MAX_FRAME_SIZE] == 16388 67 assert s[h2.settings.SettingCodes.MAX_CONCURRENT_STREAMS] == 100 68 assert s[h2.settings.SettingCodes.MAX_HEADER_LIST_SIZE] == 2**16 69 assert s[h2.settings.SettingCodes.ENABLE_CONNECT_PROTOCOL] == 1 70 71 @pytest.mark.parametrize( 72 'setting,value', 73 [ 74 (h2.settings.SettingCodes.ENABLE_PUSH, 2), 75 (h2.settings.SettingCodes.ENABLE_PUSH, -1), 76 (h2.settings.SettingCodes.INITIAL_WINDOW_SIZE, -1), 77 (h2.settings.SettingCodes.INITIAL_WINDOW_SIZE, 2**34), 78 (h2.settings.SettingCodes.MAX_FRAME_SIZE, 1), 79 (h2.settings.SettingCodes.MAX_FRAME_SIZE, 2**30), 80 (h2.settings.SettingCodes.MAX_HEADER_LIST_SIZE, -1), 81 (h2.settings.SettingCodes.ENABLE_CONNECT_PROTOCOL, -1), 82 ] 83 ) 84 def test_cannot_set_invalid_initial_values(self, setting, value): 85 """ 86 The Settings object can be provided initial values that override the 87 defaults. 88 """ 89 overrides = {setting: value} 90 91 with pytest.raises(h2.exceptions.InvalidSettingsValueError): 92 h2.settings.Settings(initial_values=overrides) 93 94 def test_applying_value_doesnt_take_effect_immediately(self): 95 """ 96 When a value is applied to the settings object, it doesn't immediately 97 take effect. 98 """ 99 s = h2.settings.Settings(client=True) 100 s[h2.settings.SettingCodes.HEADER_TABLE_SIZE] == 8000 101 102 assert s[h2.settings.SettingCodes.HEADER_TABLE_SIZE] == 4096 103 104 def test_acknowledging_values(self): 105 """ 106 When we acknowledge settings, the values change. 107 """ 108 s = h2.settings.Settings(client=True) 109 old_settings = dict(s) 110 111 new_settings = { 112 h2.settings.SettingCodes.HEADER_TABLE_SIZE: 4000, 113 h2.settings.SettingCodes.ENABLE_PUSH: 0, 114 h2.settings.SettingCodes.INITIAL_WINDOW_SIZE: 60, 115 h2.settings.SettingCodes.MAX_FRAME_SIZE: 16385, 116 h2.settings.SettingCodes.ENABLE_CONNECT_PROTOCOL: 1, 117 } 118 s.update(new_settings) 119 120 assert dict(s) == old_settings 121 s.acknowledge() 122 assert dict(s) == new_settings 123 124 def test_acknowledging_returns_the_changed_settings(self): 125 """ 126 Acknowledging settings returns the changes. 127 """ 128 s = h2.settings.Settings(client=True) 129 s[h2.settings.SettingCodes.HEADER_TABLE_SIZE] = 8000 130 s[h2.settings.SettingCodes.ENABLE_PUSH] = 0 131 132 changes = s.acknowledge() 133 assert len(changes) == 2 134 135 table_size_change = ( 136 changes[h2.settings.SettingCodes.HEADER_TABLE_SIZE] 137 ) 138 push_change = changes[h2.settings.SettingCodes.ENABLE_PUSH] 139 140 assert table_size_change.setting == ( 141 h2.settings.SettingCodes.HEADER_TABLE_SIZE 142 ) 143 assert table_size_change.original_value == 4096 144 assert table_size_change.new_value == 8000 145 146 assert push_change.setting == h2.settings.SettingCodes.ENABLE_PUSH 147 assert push_change.original_value == 1 148 assert push_change.new_value == 0 149 150 def test_acknowledging_only_returns_changed_settings(self): 151 """ 152 Acknowledging settings does not return unchanged settings. 153 """ 154 s = h2.settings.Settings(client=True) 155 s[h2.settings.SettingCodes.INITIAL_WINDOW_SIZE] = 70 156 157 changes = s.acknowledge() 158 assert len(changes) == 1 159 assert list(changes.keys()) == [ 160 h2.settings.SettingCodes.INITIAL_WINDOW_SIZE 161 ] 162 163 def test_deleting_values_deletes_all_of_them(self): 164 """ 165 When we delete a key we lose all state about it. 166 """ 167 s = h2.settings.Settings(client=True) 168 s[h2.settings.SettingCodes.HEADER_TABLE_SIZE] == 8000 169 170 del s[h2.settings.SettingCodes.HEADER_TABLE_SIZE] 171 172 with pytest.raises(KeyError): 173 s[h2.settings.SettingCodes.HEADER_TABLE_SIZE] 174 175 def test_length_correctly_reported(self): 176 """ 177 Length is related only to the number of keys. 178 """ 179 s = h2.settings.Settings(client=True) 180 assert len(s) == 5 181 182 s[h2.settings.SettingCodes.HEADER_TABLE_SIZE] == 8000 183 assert len(s) == 5 184 185 s.acknowledge() 186 assert len(s) == 5 187 188 del s[h2.settings.SettingCodes.HEADER_TABLE_SIZE] 189 assert len(s) == 4 190 191 def test_new_values_work(self): 192 """ 193 New values initially don't appear 194 """ 195 s = h2.settings.Settings(client=True) 196 s[80] = 81 197 198 with pytest.raises(KeyError): 199 s[80] 200 201 def test_new_values_follow_basic_acknowledgement_rules(self): 202 """ 203 A new value properly appears when acknowledged. 204 """ 205 s = h2.settings.Settings(client=True) 206 s[80] = 81 207 changed_settings = s.acknowledge() 208 209 assert s[80] == 81 210 assert len(changed_settings) == 1 211 212 changed = changed_settings[80] 213 assert changed.setting == 80 214 assert changed.original_value is None 215 assert changed.new_value == 81 216 217 def test_single_values_arent_affected_by_acknowledgement(self): 218 """ 219 When acknowledged, unchanged settings remain unchanged. 220 """ 221 s = h2.settings.Settings(client=True) 222 assert s[h2.settings.SettingCodes.HEADER_TABLE_SIZE] == 4096 223 224 s.acknowledge() 225 assert s[h2.settings.SettingCodes.HEADER_TABLE_SIZE] == 4096 226 227 def test_settings_getters(self): 228 """ 229 Getters exist for well-known settings. 230 """ 231 s = h2.settings.Settings(client=True) 232 233 assert s.header_table_size == ( 234 s[h2.settings.SettingCodes.HEADER_TABLE_SIZE] 235 ) 236 assert s.enable_push == s[h2.settings.SettingCodes.ENABLE_PUSH] 237 assert s.initial_window_size == ( 238 s[h2.settings.SettingCodes.INITIAL_WINDOW_SIZE] 239 ) 240 assert s.max_frame_size == s[h2.settings.SettingCodes.MAX_FRAME_SIZE] 241 assert s.max_concurrent_streams == 2**32 + 1 # A sensible default. 242 assert s.max_header_list_size is None 243 assert s.enable_connect_protocol == s[ 244 h2.settings.SettingCodes.ENABLE_CONNECT_PROTOCOL 245 ] 246 247 def test_settings_setters(self): 248 """ 249 Setters exist for well-known settings. 250 """ 251 s = h2.settings.Settings(client=True) 252 253 s.header_table_size = 0 254 s.enable_push = 1 255 s.initial_window_size = 2 256 s.max_frame_size = 16385 257 s.max_concurrent_streams = 4 258 s.max_header_list_size = 2**16 259 s.enable_connect_protocol = 1 260 261 s.acknowledge() 262 assert s[h2.settings.SettingCodes.HEADER_TABLE_SIZE] == 0 263 assert s[h2.settings.SettingCodes.ENABLE_PUSH] == 1 264 assert s[h2.settings.SettingCodes.INITIAL_WINDOW_SIZE] == 2 265 assert s[h2.settings.SettingCodes.MAX_FRAME_SIZE] == 16385 266 assert s[h2.settings.SettingCodes.MAX_CONCURRENT_STREAMS] == 4 267 assert s[h2.settings.SettingCodes.MAX_HEADER_LIST_SIZE] == 2**16 268 assert s[h2.settings.SettingCodes.ENABLE_CONNECT_PROTOCOL] == 1 269 270 @given(integers()) 271 def test_cannot_set_invalid_values_for_enable_push(self, val): 272 """ 273 SETTINGS_ENABLE_PUSH only allows two values: 0, 1. 274 """ 275 assume(val not in (0, 1)) 276 s = h2.settings.Settings() 277 278 with pytest.raises(h2.exceptions.InvalidSettingsValueError) as e: 279 s.enable_push = val 280 281 s.acknowledge() 282 assert e.value.error_code == h2.errors.ErrorCodes.PROTOCOL_ERROR 283 assert s.enable_push == 1 284 285 with pytest.raises(h2.exceptions.InvalidSettingsValueError) as e: 286 s[h2.settings.SettingCodes.ENABLE_PUSH] = val 287 288 s.acknowledge() 289 assert e.value.error_code == h2.errors.ErrorCodes.PROTOCOL_ERROR 290 assert s[h2.settings.SettingCodes.ENABLE_PUSH] == 1 291 292 @given(integers()) 293 def test_cannot_set_invalid_vals_for_initial_window_size(self, val): 294 """ 295 SETTINGS_INITIAL_WINDOW_SIZE only allows values between 0 and 2**32 - 1 296 inclusive. 297 """ 298 s = h2.settings.Settings() 299 300 if 0 <= val <= 2**31 - 1: 301 s.initial_window_size = val 302 s.acknowledge() 303 assert s.initial_window_size == val 304 else: 305 with pytest.raises(h2.exceptions.InvalidSettingsValueError) as e: 306 s.initial_window_size = val 307 308 s.acknowledge() 309 assert ( 310 e.value.error_code == h2.errors.ErrorCodes.FLOW_CONTROL_ERROR 311 ) 312 assert s.initial_window_size == 65535 313 314 with pytest.raises(h2.exceptions.InvalidSettingsValueError) as e: 315 s[h2.settings.SettingCodes.INITIAL_WINDOW_SIZE] = val 316 317 s.acknowledge() 318 assert ( 319 e.value.error_code == h2.errors.ErrorCodes.FLOW_CONTROL_ERROR 320 ) 321 assert s[h2.settings.SettingCodes.INITIAL_WINDOW_SIZE] == 65535 322 323 @given(integers()) 324 def test_cannot_set_invalid_values_for_max_frame_size(self, val): 325 """ 326 SETTINGS_MAX_FRAME_SIZE only allows values between 2**14 and 2**24 - 1. 327 """ 328 s = h2.settings.Settings() 329 330 if 2**14 <= val <= 2**24 - 1: 331 s.max_frame_size = val 332 s.acknowledge() 333 assert s.max_frame_size == val 334 else: 335 with pytest.raises(h2.exceptions.InvalidSettingsValueError) as e: 336 s.max_frame_size = val 337 338 s.acknowledge() 339 assert e.value.error_code == h2.errors.ErrorCodes.PROTOCOL_ERROR 340 assert s.max_frame_size == 16384 341 342 with pytest.raises(h2.exceptions.InvalidSettingsValueError) as e: 343 s[h2.settings.SettingCodes.MAX_FRAME_SIZE] = val 344 345 s.acknowledge() 346 assert e.value.error_code == h2.errors.ErrorCodes.PROTOCOL_ERROR 347 assert s[h2.settings.SettingCodes.MAX_FRAME_SIZE] == 16384 348 349 @given(integers()) 350 def test_cannot_set_invalid_values_for_max_header_list_size(self, val): 351 """ 352 SETTINGS_MAX_HEADER_LIST_SIZE only allows non-negative values. 353 """ 354 s = h2.settings.Settings() 355 356 if val >= 0: 357 s.max_header_list_size = val 358 s.acknowledge() 359 assert s.max_header_list_size == val 360 else: 361 with pytest.raises(h2.exceptions.InvalidSettingsValueError) as e: 362 s.max_header_list_size = val 363 364 s.acknowledge() 365 assert e.value.error_code == h2.errors.ErrorCodes.PROTOCOL_ERROR 366 assert s.max_header_list_size is None 367 368 with pytest.raises(h2.exceptions.InvalidSettingsValueError) as e: 369 s[h2.settings.SettingCodes.MAX_HEADER_LIST_SIZE] = val 370 371 s.acknowledge() 372 assert e.value.error_code == h2.errors.ErrorCodes.PROTOCOL_ERROR 373 374 with pytest.raises(KeyError): 375 s[h2.settings.SettingCodes.MAX_HEADER_LIST_SIZE] 376 377 @given(integers()) 378 def test_cannot_set_invalid_values_for_enable_connect_protocol(self, val): 379 """ 380 SETTINGS_ENABLE_CONNECT_PROTOCOL only allows two values: 0, 1. 381 """ 382 assume(val not in (0, 1)) 383 s = h2.settings.Settings() 384 385 with pytest.raises(h2.exceptions.InvalidSettingsValueError) as e: 386 s.enable_connect_protocol = val 387 388 s.acknowledge() 389 assert e.value.error_code == h2.errors.ErrorCodes.PROTOCOL_ERROR 390 assert s.enable_connect_protocol == 0 391 392 with pytest.raises(h2.exceptions.InvalidSettingsValueError) as e: 393 s[h2.settings.SettingCodes.ENABLE_CONNECT_PROTOCOL] = val 394 395 s.acknowledge() 396 assert e.value.error_code == h2.errors.ErrorCodes.PROTOCOL_ERROR 397 assert s[h2.settings.SettingCodes.ENABLE_CONNECT_PROTOCOL] == 0 398 399 400 class TestSettingsEquality(object): 401 """ 402 A class defining tests for the standard implementation of == and != . 403 """ 404 405 SettingsStrategy = builds( 406 h2.settings.Settings, 407 client=booleans(), 408 initial_values=fixed_dictionaries({ 409 h2.settings.SettingCodes.HEADER_TABLE_SIZE: 410 integers(0, 2**32 - 1), 411 h2.settings.SettingCodes.ENABLE_PUSH: integers(0, 1), 412 h2.settings.SettingCodes.INITIAL_WINDOW_SIZE: 413 integers(0, 2**31 - 1), 414 h2.settings.SettingCodes.MAX_FRAME_SIZE: 415 integers(2**14, 2**24 - 1), 416 h2.settings.SettingCodes.MAX_CONCURRENT_STREAMS: 417 integers(0, 2**32 - 1), 418 h2.settings.SettingCodes.MAX_HEADER_LIST_SIZE: 419 integers(0, 2**32 - 1), 420 }) 421 ) 422 423 @given(settings=SettingsStrategy) 424 def test_equality_reflexive(self, settings): 425 """ 426 An object compares equal to itself using the == operator and the != 427 operator. 428 """ 429 assert (settings == settings) 430 assert not (settings != settings) 431 432 @given(settings=SettingsStrategy, o_settings=SettingsStrategy) 433 def test_equality_multiple(self, settings, o_settings): 434 """ 435 Two objects compare themselves using the == operator and the != 436 operator. 437 """ 438 if settings == o_settings: 439 assert settings == o_settings 440 assert not (settings != o_settings) 441 else: 442 assert settings != o_settings 443 assert not (settings == o_settings) 444 445 @given(settings=SettingsStrategy) 446 def test_another_type_equality(self, settings): 447 """ 448 The object does not compare equal to an object of an unrelated type 449 (which does not implement the comparison) using the == operator. 450 """ 451 obj = object() 452 assert (settings != obj) 453 assert not (settings == obj) 454 455 @given(settings=SettingsStrategy) 456 def test_delegated_eq(self, settings): 457 """ 458 The result of comparison is delegated to the right-hand operand if 459 it is of an unrelated type. 460 """ 461 class Delegate(object): 462 def __eq__(self, other): 463 return [self] 464 465 def __ne__(self, other): 466 return [self] 467 468 delg = Delegate() 469 assert (settings == delg) == [delg] 470 assert (settings != delg) == [delg]