(sock, n, timeout=test.support.SHORT_TIMEOUT)
| 37 | _real_select = select.select |
| 38 | |
| 39 | def receive(sock, n, timeout=test.support.SHORT_TIMEOUT): |
| 40 | r, w, x = _real_select([sock], [], [], timeout) |
| 41 | if sock in r: |
| 42 | return sock.recv(n) |
| 43 | else: |
| 44 | raise RuntimeError("timed out on %r" % (sock,)) |
| 45 | |
| 46 | |
| 47 | @warnings_helper.ignore_fork_in_thread_deprecation_warnings() |
no test coverage detected
searching dependent graphs…