(self)
| 4091 | server.close() |
| 4092 | |
| 4093 | def test_server_accept(self): |
| 4094 | # Issue #16357: accept() on a SSLSocket created through |
| 4095 | # SSLContext.wrap_socket(). |
| 4096 | client_ctx, server_ctx, hostname = testing_context() |
| 4097 | server = socket.socket(socket.AF_INET) |
| 4098 | host = "127.0.0.1" |
| 4099 | port = socket_helper.bind_port(server) |
| 4100 | server = server_ctx.wrap_socket(server, server_side=True) |
| 4101 | self.assertTrue(server.server_side) |
| 4102 | |
| 4103 | evt = threading.Event() |
| 4104 | remote = None |
| 4105 | peer = None |
| 4106 | def serve(): |
| 4107 | nonlocal remote, peer |
| 4108 | server.listen() |
| 4109 | # Block on the accept and wait on the connection to close. |
| 4110 | evt.set() |
| 4111 | remote, peer = server.accept() |
| 4112 | remote.send(remote.recv(4)) |
| 4113 | |
| 4114 | t = threading.Thread(target=serve) |
| 4115 | t.start() |
| 4116 | # Client wait until server setup and perform a connect. |
| 4117 | evt.wait() |
| 4118 | client = client_ctx.wrap_socket( |
| 4119 | socket.socket(), server_hostname=hostname |
| 4120 | ) |
| 4121 | client.connect((hostname, port)) |
| 4122 | client.send(b'data') |
| 4123 | client.recv() |
| 4124 | client_addr = client.getsockname() |
| 4125 | client.close() |
| 4126 | t.join() |
| 4127 | remote.close() |
| 4128 | server.close() |
| 4129 | # Sanity checks. |
| 4130 | self.assertIsInstance(remote, ssl.SSLSocket) |
| 4131 | self.assertEqual(peer, client_addr) |
| 4132 | |
| 4133 | def test_getpeercert_enotconn(self): |
| 4134 | context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT) |
nothing calls this directly
no test coverage detected