(listener: socket.socket)
| 1981 | self.received_headers: list[bytes] = [] |
| 1982 | |
| 1983 | def socket_handler(listener: socket.socket) -> None: |
| 1984 | sock = listener.accept()[0] |
| 1985 | |
| 1986 | buf = b"" |
| 1987 | while not buf.endswith(b"\r\n\r\n"): |
| 1988 | buf += sock.recv(65536) |
| 1989 | |
| 1990 | self.received_headers = [ |
| 1991 | header for header in buf.split(b"\r\n")[1:] if header |
| 1992 | ] |
| 1993 | |
| 1994 | for header in self.received_headers: |
| 1995 | (key, value) = header.split(b": ") |
| 1996 | self.parsed_headers[key.decode("ascii")] = value.decode("ascii") |
| 1997 | |
| 1998 | sock.send(b"HTTP/1.1 204 No Content\r\nContent-Length: 0\r\n\r\n") |
| 1999 | |
| 2000 | sock.close() |
| 2001 | |
| 2002 | self._start_server(socket_handler) |
| 2003 |
nothing calls this directly
no test coverage detected