(self)
| 59 | self._connect() |
| 60 | |
| 61 | def _connect(self): |
| 62 | if self.sock: |
| 63 | self.sock.close() |
| 64 | self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) |
| 65 | self.sock.connect((REDIRECT_API_HOST, REDIRECT_API_PORT)) |
| 66 | |
| 67 | self.wfile = self.sock.makefile("wb") |
| 68 | self.rfile = self.sock.makefile("rb") |
| 69 | write(os.getpid(), self.wfile) |
| 70 | |
| 71 | def original_addr(self, csock: socket.socket): |
| 72 | ip, port = csock.getpeername()[:2] |
no test coverage detected