(address, timeout=None)
| 272 | |
| 273 | @contextlib.contextmanager |
| 274 | def create_connection(address, timeout=None): |
| 275 | sock = socket.socket() |
| 276 | sock.connect(address) |
| 277 | if timeout is not None: |
| 278 | sock.settimeout(timeout) |
| 279 | try: |
| 280 | yield sock |
| 281 | finally: |
| 282 | sock.close() |
| 283 | |
| 284 | class ThreadableTest: |
| 285 | """Threadable Test class |
no test coverage detected