(self, client_prog,
family=socket.AF_INET,
timeout=support.SHORT_TIMEOUT)
| 132 | self, sock, server_prog, timeout, max_clients) |
| 133 | |
| 134 | def tcp_client(self, client_prog, |
| 135 | family=socket.AF_INET, |
| 136 | timeout=support.SHORT_TIMEOUT): |
| 137 | |
| 138 | sock = socket.socket(family, socket.SOCK_STREAM) |
| 139 | |
| 140 | if timeout is None: |
| 141 | raise RuntimeError('timeout is required') |
| 142 | if timeout <= 0: |
| 143 | raise RuntimeError('only blocking sockets are supported') |
| 144 | sock.settimeout(timeout) |
| 145 | |
| 146 | return TestThreadedClient( |
| 147 | self, sock, client_prog, timeout) |
| 148 | |
| 149 | def unix_server(self, *args, **kwargs): |
| 150 | return self.tcp_server(*args, family=socket.AF_UNIX, **kwargs) |
no test coverage detected