diff --git a/tests/test_stateful_interactions.py b/tests/test_stateful_interactions.py new file mode 100644 index 000000000..955639bae --- /dev/null +++ b/tests/test_stateful_interactions.py @@ -0,0 +1,235 @@ +"""Generate sequences of exchanges between two in-memory HTTP/2 peers.""" + +from __future__ import annotations + +from dataclasses import dataclass, field + +from hypothesis import settings +from hypothesis import strategies as st +from hypothesis.stateful import RuleBasedStateMachine, invariant, precondition, rule + +from h2.config import H2Configuration +from h2.connection import H2Connection +from h2.events import DataReceived, PingAckReceived, RequestReceived, ResponseReceived, StreamEnded, StreamReset, TrailersReceived +from h2.settings import SettingCodes + + +@dataclass +class Exchange: + request: bytearray = field(default_factory=bytearray) + response: bytearray = field(default_factory=bytearray) + received_request: bytearray = field(default_factory=bytearray) + received_response: bytearray = field(default_factory=bytearray) + observed_request: bool = False + observed_response: bool = False + observed_reset: bool = False + request_ended: bool = False + response_started: bool = False + response_ended: bool = False + reset: bool = False + observed_request_end: bool = False + observed_response_end: bool = False + request_trailers: bool = False + response_trailers: bool = False + observed_request_trailers: bool = False + observed_response_trailers: bool = False + + +class InteractingPeers(RuleBasedStateMachine): + """Keep a small application model independent of either peer's state machine.""" + + def __init__(self) -> None: + super().__init__() + self.client = H2Connection(config=H2Configuration(client_side=True)) + self.server = H2Connection(config=H2Configuration(client_side=False)) + self.exchanges: dict[int, Exchange] = {} + self.next_stream_id = 1 + self.ping_acks: list[bytes] = [] + self.client.initiate_connection() + self.server.initiate_connection() + self.deliver(self.client, self.server, 17) + self.deliver(self.server, self.client, 17) + self.deliver(self.client, self.server, 17) + + def deliver(self, sender: H2Connection, receiver: H2Connection, chunk: int) -> None: + """Deliver fragmented wire bytes, including automatic ACKs, without I/O.""" + pending = [(sender, receiver)] + rounds = 0 + while pending: + rounds += 1 + assert rounds < 100, "automatic peer responses did not settle" + source, destination = pending.pop() + wire = source.data_to_send() + if not wire: + continue + for offset in range(0, len(wire), chunk): + for event in destination.receive_data(wire[offset : offset + chunk]): + self.observe(destination, event) + pending.append((destination, source)) + + def observe(self, receiver: H2Connection, event) -> None: + if isinstance(event, PingAckReceived): + self.ping_acks.append(event.ping_data) + elif isinstance(event, RequestReceived): + assert receiver is self.server + exchange = self.exchanges[event.stream_id] + assert not exchange.observed_request + exchange.observed_request = True + assert (b":method", b"POST") in event.headers + elif isinstance(event, ResponseReceived): + assert receiver is self.client + exchange = self.exchanges[event.stream_id] + assert exchange.response_started + assert not exchange.observed_response + exchange.observed_response = True + assert (b":status", b"200") in event.headers + elif isinstance(event, DataReceived): + exchange = self.exchanges[event.stream_id] + received = exchange.received_request if receiver is self.server else exchange.received_response + received.extend(event.data) + receiver.acknowledge_received_data(event.flow_controlled_length, event.stream_id) + elif isinstance(event, StreamEnded): + exchange = self.exchanges[event.stream_id] + if receiver is self.server: + assert exchange.request_ended + assert not exchange.observed_request_end + exchange.observed_request_end = True + else: + assert exchange.response_ended + assert not exchange.observed_response_end + exchange.observed_response_end = True + elif isinstance(event, StreamReset): + exchange = self.exchanges[event.stream_id] + assert exchange.reset + assert not exchange.observed_reset + exchange.observed_reset = True + elif isinstance(event, TrailersReceived): + exchange = self.exchanges[event.stream_id] + if receiver is self.server: + assert exchange.request_trailers + assert not exchange.observed_request_trailers + exchange.observed_request_trailers = True + else: + assert exchange.response_trailers + assert not exchange.observed_response_trailers + exchange.observed_response_trailers = True + assert event.headers == [(b"checksum", b"done")] + + @precondition(lambda self: len(self.exchanges) < 8) + @rule(end_stream=st.booleans(), chunk=st.integers(1, 257)) + def request(self, end_stream: bool, chunk: int) -> None: + stream_id = self.next_stream_id + self.next_stream_id += 2 + self.exchanges[stream_id] = Exchange(request_ended=end_stream) + self.client.send_headers( + stream_id, + [ + (":method", "POST"), + (":scheme", "https"), + (":authority", "example.com"), + (":path", f"/{stream_id}"), + ], + end_stream=end_stream, + ) + self.deliver(self.client, self.server, chunk) + + def request_streams(self) -> list[int]: + return [sid for sid, exchange in self.exchanges.items() if not exchange.request_ended and not exchange.reset] + + def response_streams(self) -> list[int]: + return [sid for sid, exchange in self.exchanges.items() if exchange.response_started and not exchange.response_ended and not exchange.reset] + + @precondition(lambda self: bool(self.request_streams())) + @rule(data=st.data(), payload=st.binary(max_size=2048), end_stream=st.booleans(), chunk=st.integers(1, 257)) + def request_data(self, data, payload: bytes, end_stream: bool, chunk: int) -> None: + stream_id = data.draw(st.sampled_from(self.request_streams())) + exchange = self.exchanges[stream_id] + payload = payload[: max(0, self.client.local_flow_control_window(stream_id))] + exchange.request.extend(payload) + exchange.request_ended = end_stream + self.client.send_data(stream_id, payload, end_stream=end_stream) + self.deliver(self.client, self.server, chunk) + + @precondition(lambda self: any(not e.response_started and not e.reset for e in self.exchanges.values())) + @rule(data=st.data(), end_stream=st.booleans(), chunk=st.integers(1, 257)) + def response(self, data, end_stream: bool, chunk: int) -> None: + available = [sid for sid, e in self.exchanges.items() if not e.response_started and not e.reset] + stream_id = data.draw(st.sampled_from(available)) + exchange = self.exchanges[stream_id] + exchange.response_started = True + exchange.response_ended = end_stream + self.server.send_headers(stream_id, [(":status", "200")], end_stream=end_stream) + self.deliver(self.server, self.client, chunk) + + @precondition(lambda self: bool(self.response_streams())) + @rule(data=st.data(), payload=st.binary(max_size=2048), end_stream=st.booleans(), chunk=st.integers(1, 257)) + def response_data(self, data, payload: bytes, end_stream: bool, chunk: int) -> None: + stream_id = data.draw(st.sampled_from(self.response_streams())) + exchange = self.exchanges[stream_id] + payload = payload[: max(0, self.server.local_flow_control_window(stream_id))] + exchange.response.extend(payload) + exchange.response_ended = end_stream + self.server.send_data(stream_id, payload, end_stream=end_stream) + self.deliver(self.server, self.client, chunk) + + @precondition(lambda self: bool(self.request_streams() or self.response_streams())) + @rule(data=st.data(), response_side=st.booleans(), chunk=st.integers(1, 257)) + def trailers(self, data, response_side: bool, chunk: int) -> None: + available = self.response_streams() if response_side else self.request_streams() + if not available: + return + stream_id = data.draw(st.sampled_from(available)) + exchange = self.exchanges[stream_id] + if response_side: + exchange.response_trailers = True + exchange.response_ended = True + sender, receiver = self.server, self.client + else: + exchange.request_trailers = True + exchange.request_ended = True + sender, receiver = self.client, self.server + sender.send_headers(stream_id, [("checksum", "done")], end_stream=True) + self.deliver(sender, receiver, chunk) + + @precondition(lambda self: any(not e.reset and not (e.request_ended and e.response_ended) for e in self.exchanges.values())) + @rule(data=st.data(), client_side=st.booleans(), chunk=st.integers(1, 257)) + def reset(self, data, client_side: bool, chunk: int) -> None: + available = [sid for sid, e in self.exchanges.items() if not e.reset and not (e.request_ended and e.response_ended)] + stream_id = data.draw(st.sampled_from(available)) + self.exchanges[stream_id].reset = True + sender, receiver = (self.client, self.server) if client_side else (self.server, self.client) + sender.reset_stream(stream_id) + self.deliver(sender, receiver, chunk) + + @rule(client_side=st.booleans(), window=st.integers(16384, 65535), chunk=st.integers(1, 257)) + def change_window(self, client_side: bool, window: int, chunk: int) -> None: + sender, receiver = (self.client, self.server) if client_side else (self.server, self.client) + sender.update_settings({SettingCodes.INITIAL_WINDOW_SIZE: window}) + self.deliver(sender, receiver, chunk) + assert sender.local_settings.initial_window_size == window + assert receiver.remote_settings.initial_window_size == window + + @rule(client_side=st.booleans(), opaque=st.binary(min_size=8, max_size=8), chunk=st.integers(1, 257)) + def ping(self, client_side: bool, opaque: bytes, chunk: int) -> None: + sender, receiver = (self.client, self.server) if client_side else (self.server, self.client) + count = len(self.ping_acks) + sender.ping(opaque) + self.deliver(sender, receiver, chunk) + assert self.ping_acks[count:] == [opaque] + + @invariant() + def application_data_agrees(self) -> None: + for exchange in self.exchanges.values(): + assert exchange.observed_request + assert exchange.response_started == exchange.observed_response + assert exchange.reset == exchange.observed_reset + assert exchange.request == exchange.received_request + assert exchange.response == exchange.received_response + assert exchange.request_ended == exchange.observed_request_end + assert exchange.response_ended == exchange.observed_response_end + assert exchange.request_trailers == exchange.observed_request_trailers + assert exchange.response_trailers == exchange.observed_response_trailers + + +TestInteractingPeers = InteractingPeers.TestCase +TestInteractingPeers.settings = settings(max_examples=100, stateful_step_count=50, deadline=None)