How to use the h2.config function in h2

To help you get started, we’ve selected a few h2 examples, based on popular ways it is used in public projects.

Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.

github python-hyper / hyper-h2 / test / test_related_events.py View on Github external
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,
github python-hyper / hyper-h2 / test / test_config.py View on Github external
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)
github python-hyper / hyper-h2 / test / test_h2_upgrade.py View on Github external
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()
github encode / httpx / tests / dispatch / utils.py View on Github external
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 = []
github python-hyper / hyper-h2 / test / test_closed_streams.py View on Github external
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
github python-hyper / hyper-h2 / test / test_informational_responses.py View on Github external
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())
github pgjones / hypercorn / hypercorn / trio / h2.py View on Github external
*,
        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,
            },
github HENNGE / aapns / src / aapns / connection.py View on Github external
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,
            },
        )
github pgjones / quart / quart / serving / h2.py View on Github external
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
github pgjones / quart / quart / serving.py View on Github external
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