(self, timed_out, all_read)
| 138 | return socket_pexpect.SocketSpawn(socket, timeout=timeout, use_poll=use_poll) |
| 139 | |
| 140 | def socket_fn(self, timed_out, all_read): |
| 141 | result = 0 |
| 142 | try: |
| 143 | sock = socket.socket(self.af, socket.SOCK_STREAM) |
| 144 | sock.connect((self.host, self.port)) |
| 145 | session = self.spawn(sock, timeout=10) |
| 146 | # Get all data from server |
| 147 | session.read_nonblocking(size=4096) |
| 148 | all_read.set() |
| 149 | # This read should timeout |
| 150 | session.read_nonblocking(size=4096) |
| 151 | except pexpect.TIMEOUT: |
| 152 | timed_out.set() |
| 153 | result = errno.ETIMEDOUT |
| 154 | exit(result) |
| 155 | |
| 156 | def test_socket(self): |
| 157 | sock = socket.socket(self.af, socket.SOCK_STREAM) |
nothing calls this directly
no test coverage detected