(self)
| 4013 | self.assertEqual(bytes(buf), b"data\0\0") |
| 4014 | |
| 4015 | def test_nonblocking_send(self): |
| 4016 | server = ThreadedEchoServer(CERTFILE, |
| 4017 | certreqs=ssl.CERT_NONE, |
| 4018 | ssl_version=ssl.PROTOCOL_TLS_SERVER, |
| 4019 | cacerts=CERTFILE, |
| 4020 | chatty=True, |
| 4021 | connectionchatty=False) |
| 4022 | with server: |
| 4023 | s = test_wrap_socket(socket.socket(), |
| 4024 | server_side=False, |
| 4025 | certfile=CERTFILE, |
| 4026 | ca_certs=CERTFILE, |
| 4027 | cert_reqs=ssl.CERT_NONE) |
| 4028 | s.connect((HOST, server.port)) |
| 4029 | s.setblocking(False) |
| 4030 | |
| 4031 | # If we keep sending data, at some point the buffers |
| 4032 | # will be full and the call will block |
| 4033 | buf = bytearray(8192) |
| 4034 | def fill_buffer(): |
| 4035 | while True: |
| 4036 | s.send(buf) |
| 4037 | self.assertRaises((ssl.SSLWantWriteError, |
| 4038 | ssl.SSLWantReadError), fill_buffer) |
| 4039 | |
| 4040 | # Now read all the output and discard it |
| 4041 | s.setblocking(True) |
| 4042 | s.close() |
| 4043 | |
| 4044 | def test_handshake_timeout(self): |
| 4045 | # Issue #5103: SSL handshake must respect the socket timeout |
nothing calls this directly
no test coverage detected