A mock HTTP "server" that just pretends it received a full HTTP request, which is then processed by the proxy core.
| 34 | |
| 35 | |
| 36 | class MockServer(layers.http.HttpConnection): |
| 37 | """ |
| 38 | A mock HTTP "server" that just pretends it received a full HTTP request, |
| 39 | which is then processed by the proxy core. |
| 40 | """ |
| 41 | |
| 42 | flow: http.HTTPFlow |
| 43 | |
| 44 | def __init__(self, flow: http.HTTPFlow, context: Context): |
| 45 | super().__init__(context, context.client) |
| 46 | self.flow = flow |
| 47 | |
| 48 | def _handle_event(self, event: events.Event) -> CommandGenerator[None]: |
| 49 | if isinstance(event, events.Start): |
| 50 | content = self.flow.request.raw_content |
| 51 | self.flow.request.timestamp_start = self.flow.request.timestamp_end = ( |
| 52 | time.time() |
| 53 | ) |
| 54 | yield layers.http.ReceiveHttp( |
| 55 | layers.http.RequestHeaders( |
| 56 | 1, |
| 57 | self.flow.request, |
| 58 | end_stream=not (content or self.flow.request.trailers), |
| 59 | replay_flow=self.flow, |
| 60 | ) |
| 61 | ) |
| 62 | if content: |
| 63 | yield layers.http.ReceiveHttp(layers.http.RequestData(1, content)) |
| 64 | if self.flow.request.trailers: # pragma: no cover |
| 65 | # TODO: Cover this once we support HTTP/1 trailers. |
| 66 | yield layers.http.ReceiveHttp( |
| 67 | layers.http.RequestTrailers(1, self.flow.request.trailers) |
| 68 | ) |
| 69 | yield layers.http.ReceiveHttp(layers.http.RequestEndOfMessage(1)) |
| 70 | elif isinstance( |
| 71 | event, |
| 72 | ( |
| 73 | layers.http.ResponseHeaders, |
| 74 | layers.http.ResponseData, |
| 75 | layers.http.ResponseTrailers, |
| 76 | layers.http.ResponseEndOfMessage, |
| 77 | layers.http.ResponseProtocolError, |
| 78 | ), |
| 79 | ): |
| 80 | pass |
| 81 | else: # pragma: no cover |
| 82 | logger.warning(f"Unexpected event during replay: {event}") |
| 83 | |
| 84 | |
| 85 | class ReplayHandler(server.ConnectionHandler): |
no outgoing calls
no test coverage detected
searching dependent graphs…