Source code
Revision control
Copy as Markdown
Other Tools
# -*- coding: utf-8 -*-
"""
helpers
~~~~~~~
This module contains helpers for the h2 tests.
"""
from hyperframe.frame import (
HeadersFrame, DataFrame, SettingsFrame, WindowUpdateFrame, PingFrame,
GoAwayFrame, RstStreamFrame, PushPromiseFrame, PriorityFrame,
ContinuationFrame, AltSvcFrame
)
from hpack.hpack import Encoder
SAMPLE_SETTINGS = {
SettingsFrame.HEADER_TABLE_SIZE: 4096,
SettingsFrame.ENABLE_PUSH: 1,
SettingsFrame.MAX_CONCURRENT_STREAMS: 2,
}
class FrameFactory(object):
"""
A class containing lots of helper methods and state to build frames. This
allows test cases to easily build correct HTTP/2 frames to feed to
hyper-h2.
"""
def __init__(self):
self.encoder = Encoder()
def refresh_encoder(self):
self.encoder = Encoder()
def preamble(self):
return b'PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n'
def build_headers_frame(self,
headers,
flags=[],
stream_id=1,
**priority_kwargs):
"""
Builds a single valid headers frame out of the contained headers.
"""
f = HeadersFrame(stream_id)
f.data = self.encoder.encode(headers)
f.flags.add('END_HEADERS')
for flag in flags:
f.flags.add(flag)
for k, v in priority_kwargs.items():
setattr(f, k, v)
return f
def build_continuation_frame(self, header_block, flags=[], stream_id=1):
"""
Builds a single continuation frame out of the binary header block.
"""
f = ContinuationFrame(stream_id)
f.data = header_block
f.flags = set(flags)
return f
def build_data_frame(self, data, flags=None, stream_id=1, padding_len=0):
"""
Builds a single data frame out of a chunk of data.
"""
flags = set(flags) if flags is not None else set()
f = DataFrame(stream_id)
f.data = data
f.flags = flags
if padding_len:
flags.add('PADDED')
f.pad_length = padding_len
return f
def build_settings_frame(self, settings, ack=False):
"""
Builds a single settings frame.
"""
f = SettingsFrame(0)
if ack:
f.flags.add('ACK')
f.settings = settings
return f
def build_window_update_frame(self, stream_id, increment):
"""
Builds a single WindowUpdate frame.
"""
f = WindowUpdateFrame(stream_id)
f.window_increment = increment
return f
def build_ping_frame(self, ping_data, flags=None):
"""
Builds a single Ping frame.
"""
f = PingFrame(0)
f.opaque_data = ping_data
if flags:
f.flags = set(flags)
return f
def build_goaway_frame(self,
last_stream_id,
error_code=0,
additional_data=b''):
"""
Builds a single GOAWAY frame.
"""
f = GoAwayFrame(0)
f.error_code = error_code
f.last_stream_id = last_stream_id
f.additional_data = additional_data
return f
def build_rst_stream_frame(self, stream_id, error_code=0):
"""
Builds a single RST_STREAM frame.
"""
f = RstStreamFrame(stream_id)
f.error_code = error_code
return f
def build_push_promise_frame(self,
stream_id,
promised_stream_id,
headers,
flags=[]):
"""
Builds a single PUSH_PROMISE frame.
"""
f = PushPromiseFrame(stream_id)
f.promised_stream_id = promised_stream_id
f.data = self.encoder.encode(headers)
f.flags = set(flags)
f.flags.add('END_HEADERS')
return f
def build_priority_frame(self,
stream_id,
weight,
depends_on=0,
exclusive=False):
"""
Builds a single priority frame.
"""
f = PriorityFrame(stream_id)
f.depends_on = depends_on
f.stream_weight = weight
f.exclusive = exclusive
return f
def build_alt_svc_frame(self, stream_id, origin, field):
"""
Builds a single ALTSVC frame.
"""
f = AltSvcFrame(stream_id)
f.origin = origin
f.field = field
return f
def change_table_size(self, new_size):
"""
Causes the encoder to send a dynamic size update in the next header
block it sends.
"""
self.encoder.header_table_size = new_size