(self)
| 3808 | server.close() |
| 3809 | |
| 3810 | def test_server_accept(self): |
| 3811 | # Issue #16357: accept() on a SSLSocket created through |
| 3812 | # SSLContext.wrap_socket(). |
| 3813 | client_ctx, server_ctx, hostname = testing_context() |
| 3814 | server = socket.socket(socket.AF_INET) |
| 3815 | host = "127.0.0.1" |
| 3816 | port = socket_helper.bind_port(server) |
| 3817 | server = server_ctx.wrap_socket(server, server_side=True) |
| 3818 | self.assertTrue(server.server_side) |
| 3819 | |
| 3820 | evt = threading.Event() |
| 3821 | remote = None |
| 3822 | peer = None |
| 3823 | def serve(): |
| 3824 | nonlocal remote, peer |
| 3825 | server.listen() |
| 3826 | # Block on the accept and wait on the connection to close. |
| 3827 | evt.set() |
| 3828 | remote, peer = server.accept() |
| 3829 | remote.send(remote.recv(4)) |
| 3830 | |
| 3831 | t = threading.Thread(target=serve) |
| 3832 | t.start() |
| 3833 | # Client wait until server setup and perform a connect. |
| 3834 | evt.wait() |
| 3835 | client = client_ctx.wrap_socket( |
| 3836 | socket.socket(), server_hostname=hostname |
| 3837 | ) |
| 3838 | client.connect((hostname, port)) |
| 3839 | client.send(b'data') |
| 3840 | client.recv() |
| 3841 | client_addr = client.getsockname() |
| 3842 | client.close() |
| 3843 | t.join() |
| 3844 | remote.close() |
| 3845 | server.close() |
| 3846 | # Sanity checks. |
| 3847 | self.assertIsInstance(remote, ssl.SSLSocket) |
| 3848 | self.assertEqual(peer, client_addr) |
| 3849 | |
| 3850 | def test_getpeercert_enotconn(self): |
| 3851 | context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT) |
nothing calls this directly
no test coverage detected