| import socket |
| import sys |
| from pathlib import Path |
|
|
| ROOT = Path(__file__).resolve().parents[1] |
| if str(ROOT) not in sys.path: |
| sys.path.insert(0, str(ROOT)) |
|
|
|
|
| def _fallback_socketpair_no_peer_check(family=socket.AF_INET, type=socket.SOCK_STREAM, proto=0): |
| if family == socket.AF_INET: |
| host = "127.0.0.1" |
| elif family == socket.AF_INET6: |
| host = "::1" |
| else: |
| raise ValueError("Only AF_INET and AF_INET6 socket address families are supported") |
| if type != socket.SOCK_STREAM: |
| raise ValueError("Only SOCK_STREAM socket type is supported") |
| if proto != 0: |
| raise ValueError("Only protocol zero is supported") |
|
|
| listener = socket.socket(family, type, proto) |
| try: |
| listener.bind((host, 0)) |
| listener.listen(1) |
| address = listener.getsockname()[:2] |
| client = socket.socket(family, type, proto) |
| try: |
| client.setblocking(False) |
| try: |
| client.connect(address) |
| except (BlockingIOError, InterruptedError): |
| pass |
| client.setblocking(True) |
| server, _ = listener.accept() |
| except Exception: |
| client.close() |
| raise |
| finally: |
| listener.close() |
|
|
| return server, client |
|
|
|
|
| if sys.platform.startswith("win"): |
| socket.socketpair = _fallback_socketpair_no_peer_check |
|
|
