Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
example_response_headers = [
(':status', '200'),
('server', 'fake-serv/0.1.0')
]
informational_response_headers = [
(':status', '100'),
('server', 'fake-serv/0.1.0')
]
example_trailers = [
('another', 'field'),
]
server_config = h2.config.H2Configuration(client_side=False)
def test_request_received_related_all(self, frame_factory):
"""
RequestReceived has two possible related events: PriorityUpdated and
StreamEnded, all fired when a single HEADERS frame is received.
"""
c = h2.connection.H2Connection(config=self.server_config)
c.initiate_connection()
c.receive_data(frame_factory.preamble())
input_frame = frame_factory.build_headers_frame(
headers=self.example_request_headers,
flags=['END_STREAM', 'PRIORITY'],
stream_weight=15,
depends_on=0,
exclusive=False,
def test_defaults(self):
"""
The default values of the HTTP/2 config object are sensible.
"""
config = h2.config.H2Configuration()
assert config.client_side
assert config.header_encoding is None
assert isinstance(config.logger, h2.config.DummyLogger)
class TestServerUpgrade(object):
"""
Tests of the server-side of the HTTP/2 upgrade dance.
"""
example_request_headers = [
(b':authority', b'example.com'),
(b':path', b'/'),
(b':scheme', b'https'),
(b':method', b'GET'),
]
example_response_headers = [
(b':status', b'200'),
(b'server', b'fake-serv/0.1.0')
]
server_config = h2.config.H2Configuration(client_side=False)
def test_returns_nothing(self, frame_factory):
"""
Calling initiate_upgrade_connection returns nothing.
"""
conn = h2.connection.H2Connection(config=self.server_config)
curl_header = b"AAMAAABkAAQAAP__"
data = conn.initiate_upgrade_connection(curl_header)
assert data is None
def test_emits_preamble(self, frame_factory):
"""
Calling initiate_upgrade_connection emits the connection preamble.
"""
conn = h2.connection.H2Connection(config=self.server_config)
conn.initiate_upgrade_connection()
def __init__(self, app, backend):
config = h2.config.H2Configuration(client_side=False)
self.conn = h2.connection.H2Connection(config=config)
self.app = app
self.backend = backend
self.buffer = b""
self.requests = {}
self.close_connection = False
self.return_data = {}
self.settings_changed = []
assert not events
class TestStreamsClosedByRstStream(object):
example_request_headers = [
(':authority', 'example.com'),
(':path', '/'),
(':scheme', 'https'),
(':method', 'GET'),
]
example_response_headers = [
(':status', '200'),
('server', 'fake-serv/0.1.0')
]
server_config = h2.config.H2Configuration(client_side=False)
@pytest.mark.parametrize(
"frame",
[
lambda self, ff: ff.build_headers_frame(
self.example_request_headers),
lambda self, ff: ff.build_headers_frame(
self.example_request_headers, flags=['END_STREAM']),
lambda self, ff: ff.build_data_frame(b'hello'),
]
)
def test_resets_further_frames_after_recv_reset(self,
frame_factory,
frame):
"""
A stream that is closed by receive RST_STREAM can receive further
unicode_informational_headers = [
(u':status', u'100'),
(u'server', u'fake-serv/0.1.0')
]
bytes_informational_headers = [
(b':status', b'100'),
(b'server', b'fake-serv/0.1.0')
]
example_response_headers = [
(b':status', b'200'),
(b'server', b'fake-serv/0.1.0')
]
example_trailers = [
(b'trailer', b'you-bet'),
]
server_config = h2.config.H2Configuration(client_side=False)
@pytest.mark.parametrize(
'hdrs', (unicode_informational_headers, bytes_informational_headers),
)
@pytest.mark.parametrize('end_stream', (True, False))
def test_single_informational_response(self,
frame_factory,
hdrs,
end_stream):
"""
When sending a informational response, the appropriate frames are
emitted.
"""
c = h2.connection.H2Connection(config=self.server_config)
c.initiate_connection()
c.receive_data(frame_factory.preamble())
*,
upgrade_request: Optional[h11.Request] = None,
received_data: Optional[bytes] = None,
) -> None:
super().__init__(stream, "h2")
self.app = app
self.config = config
self.streams: Dict[int, H2AsyncStream] = {}
self.flow_control: Dict[int, trio.Event] = {}
self.send_lock = trio.Lock()
self.upgrade_request = upgrade_request
self.received_data = received_data
self.connection = h2.connection.H2Connection(
config=h2.config.H2Configuration(client_side=False, header_encoding=None)
)
self.connection.DEFAULT_MAX_INBOUND_FRAME_SIZE = config.h2_max_inbound_frame_size
self.connection.local_settings = h2.settings.Settings(
client=False,
initial_values={
h2.settings.SettingCodes.MAX_CONCURRENT_STREAMS: config.h2_max_concurrent_streams,
h2.settings.SettingCodes.MAX_HEADER_LIST_SIZE: config.h2_max_header_list_size,
h2.settings.SettingCodes.ENABLE_CONNECT_PROTOCOL: 1,
},
raise ValueError("Origin must be https://[:]")
host = url.hostname
port = url.port or 443
ssl_context = ssl if ssl else create_ssl_context()
if (
OP_NO_TLSv1 not in ssl_context.options # type: ignore # https://github.com/python/typeshed/issues/3920
or OP_NO_TLSv1_1 not in ssl_context.options # type: ignore
):
raise ValueError("SSL Context cannot allow TLS 1.0 or 1.1")
# https://bugs.python.org/issue40111 validate context h2 alpn
protocol = h2.connection.H2Connection(
h2.config.H2Configuration(client_side=True, header_encoding="utf-8")
)
protocol.local_settings = h2.settings.Settings(
client=True,
initial_values={
# Apple server settings:
# HEADER_TABLE_SIZE 4096
# MAX_CONCURRENT_STREAMS 1000
# INITIAL_WINDOW_SIZE 65535
# MAX_FRAME_SIZE 16384
# MAX_HEADER_LIST_SIZE 8000
h2.settings.SettingCodes.ENABLE_PUSH: 0,
h2.settings.SettingCodes.MAX_CONCURRENT_STREAMS: 2 ** 20,
h2.settings.SettingCodes.MAX_HEADER_LIST_SIZE: 2 ** 16 - 1,
h2.settings.SettingCodes.INITIAL_WINDOW_SIZE: MAX_RESPONSE_SIZE,
},
)
def __init__(
self,
app: 'Quart',
loop: asyncio.AbstractEventLoop,
transport: asyncio.BaseTransport,
logger: Optional[Logger],
access_log_format: str,
timeout: int,
*,
upgrade_request: Optional['h11.Request']=None,
) -> None:
super().__init__(app, loop, transport, logger, 'h2', access_log_format, timeout)
self.connection = h2.connection.H2Connection(
config=h2.config.H2Configuration(client_side=False, header_encoding='utf-8'),
)
if upgrade_request is None:
self.connection.initiate_connection()
else:
headers = CIMultiDict()
for name, value in upgrade_request.headers:
headers.add(name.decode().title(), value.decode())
self.connection.initiate_upgrade_connection(headers.get('HTTP2-Settings', ''))
self.handle_request(
1, upgrade_request.method.decode().upper(), upgrade_request.target.decode(),
headers,
)
self.write(self.connection.data_to_send()) # type: ignore
def __init__(
self,
app: 'Quart',
loop: asyncio.AbstractEventLoop,
transport: asyncio.BaseTransport,
logger: Optional[Logger],
access_log_format: str,
timeout: int,
) -> None:
super().__init__(app, loop, transport, logger, access_log_format, timeout)
self.connection = h2.connection.H2Connection(
h2.config.H2Configuration(client_side=False, header_encoding='utf-8'),
)
self.connection.initiate_connection()
self.send(self.connection.data_to_send()) # type: ignore