(self)
| 3730 | self.assertEqual(bytes(buf), b"data\0\0") |
| 3731 | |
| 3732 | def test_nonblocking_send(self): |
| 3733 | server = ThreadedEchoServer(CERTFILE, |
| 3734 | certreqs=ssl.CERT_NONE, |
| 3735 | ssl_version=ssl.PROTOCOL_TLS_SERVER, |
| 3736 | cacerts=CERTFILE, |
| 3737 | chatty=True, |
| 3738 | connectionchatty=False) |
| 3739 | with server: |
| 3740 | s = test_wrap_socket(socket.socket(), |
| 3741 | server_side=False, |
| 3742 | certfile=CERTFILE, |
| 3743 | ca_certs=CERTFILE, |
| 3744 | cert_reqs=ssl.CERT_NONE) |
| 3745 | s.connect((HOST, server.port)) |
| 3746 | s.setblocking(False) |
| 3747 | |
| 3748 | # If we keep sending data, at some point the buffers |
| 3749 | # will be full and the call will block |
| 3750 | buf = bytearray(8192) |
| 3751 | def fill_buffer(): |
| 3752 | while True: |
| 3753 | s.send(buf) |
| 3754 | self.assertRaises((ssl.SSLWantWriteError, |
| 3755 | ssl.SSLWantReadError), fill_buffer) |
| 3756 | |
| 3757 | # Now read all the output and discard it |
| 3758 | s.setblocking(True) |
| 3759 | s.close() |
| 3760 | |
| 3761 | def test_handshake_timeout(self): |
| 3762 | # Issue #5103: SSL handshake must respect the socket timeout |
nothing calls this directly
no test coverage detected