Start a TLS server with a self-signed cert. Yields (host, port).
(handler, certfile, keyfile, *, timeout=5.0)
| 1066 | |
| 1067 | @contextlib.contextmanager |
| 1068 | def tls_server(handler, certfile, keyfile, *, timeout=5.0): |
| 1069 | """Start a TLS server with a self-signed cert. Yields (host, port).""" |
| 1070 | import ssl as host_ssl |
| 1071 | |
| 1072 | server_ctx = host_ssl.SSLContext(host_ssl.PROTOCOL_TLS_SERVER) # type: ignore[attr-defined] |
| 1073 | server_ctx.load_cert_chain(certfile, keyfile) |
| 1074 | |
| 1075 | server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) |
| 1076 | server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) |
| 1077 | server_socket.bind(("127.0.0.1", 0)) |
| 1078 | server_socket.listen(1) |
| 1079 | server_socket.settimeout(timeout) |
| 1080 | host, port = server_socket.getsockname() |
| 1081 | tls_sock = server_ctx.wrap_socket(server_socket, server_side=True) |
| 1082 | |
| 1083 | errors: list[str] = [] |
| 1084 | ready = threading.Event() |
| 1085 | |
| 1086 | def _serve(): |
| 1087 | ready.set() |
| 1088 | try: |
| 1089 | conn, addr = tls_sock.accept() |
| 1090 | try: |
| 1091 | handler(conn, addr) |
| 1092 | finally: |
| 1093 | conn.close() |
| 1094 | except Exception as e: |
| 1095 | errors.append(str(e)) |
| 1096 | finally: |
| 1097 | tls_sock.close() |
| 1098 | |
| 1099 | thread = threading.Thread(target=_serve, daemon=True) |
| 1100 | thread.start() |
| 1101 | ready.wait(timeout=timeout) |
| 1102 | |
| 1103 | try: |
| 1104 | yield host, port |
| 1105 | finally: |
| 1106 | thread.join(timeout=timeout) |
| 1107 | assert not errors, f"TLS server error: {errors[0]}" |
| 1108 | |
| 1109 | |
| 1110 | @pytest.mark.skip_refcount_check |
no test coverage detected
searching dependent graphs…