(self, msg: str)
| 301 | self.transport.set_protocol(protocol) |
| 302 | |
| 303 | def send_400_response(self, msg: str) -> None: |
| 304 | reason = STATUS_PHRASES[400] |
| 305 | headers: list[tuple[bytes, bytes]] = [ |
| 306 | (b"content-type", b"text/plain; charset=utf-8"), |
| 307 | (b"connection", b"close"), |
| 308 | ] |
| 309 | event = h11.Response(status_code=400, headers=headers, reason=reason) |
| 310 | output = self.conn.send(event) |
| 311 | self.transport.write(output) |
| 312 | |
| 313 | output = self.conn.send(event=h11.Data(data=msg.encode("ascii"))) |
| 314 | self.transport.write(output) |
| 315 | |
| 316 | output = self.conn.send(event=h11.EndOfMessage()) |
| 317 | self.transport.write(output) |
| 318 | |
| 319 | self.transport.close() |
| 320 | |
| 321 | def on_response_complete(self) -> None: |
| 322 | self.server_state.total_requests += 1 |
no test coverage detected