(self, arg)
| 152 | asynchat.async_chat.push(self, data.encode(self.encoding) + b'\r\n') |
| 153 | |
| 154 | def cmd_port(self, arg): |
| 155 | addr = list(map(int, arg.split(','))) |
| 156 | ip = '%d.%d.%d.%d' %tuple(addr[:4]) |
| 157 | port = (addr[4] * 256) + addr[5] |
| 158 | s = socket.create_connection((ip, port), timeout=TIMEOUT) |
| 159 | self.dtp = self.dtp_handler(s, baseclass=self) |
| 160 | self.push('200 active data connection established') |
| 161 | |
| 162 | def cmd_pasv(self, arg): |
| 163 | with socket.create_server((self.socket.getsockname()[0], 0)) as sock: |
nothing calls this directly
no test coverage detected