Try to send a file using kTLS if possible.
(self)
| 4729 | self.assertRaises(ValueError, s.write, b'hello') |
| 4730 | |
| 4731 | def test_sendfile(self): |
| 4732 | """Try to send a file using kTLS if possible.""" |
| 4733 | TEST_DATA = b"x" * 512 |
| 4734 | with open(os_helper.TESTFN, 'wb') as f: |
| 4735 | f.write(TEST_DATA) |
| 4736 | self.addCleanup(os_helper.unlink, os_helper.TESTFN) |
| 4737 | client_context, server_context, hostname = testing_context() |
| 4738 | client_context.options |= getattr(ssl, 'OP_ENABLE_KTLS', 0) |
| 4739 | server = ThreadedEchoServer(context=server_context, chatty=False) |
| 4740 | # kTLS seems to work only with a connection created before |
| 4741 | # wrapping `sock` by the SSL context in contrast to calling |
| 4742 | # `sock.connect()` after the wrapping. |
| 4743 | with server, socket.create_connection((HOST, server.port)) as sock: |
| 4744 | with client_context.wrap_socket( |
| 4745 | sock, server_hostname=hostname |
| 4746 | ) as ssock: |
| 4747 | if support.verbose: |
| 4748 | ktls_used = ssock._sslobj.uses_ktls_for_send() |
| 4749 | print( |
| 4750 | 'kTLS is', |
| 4751 | 'available' if ktls_used else 'unavailable', |
| 4752 | ) |
| 4753 | with open(os_helper.TESTFN, 'rb') as file: |
| 4754 | ssock.sendfile(file) |
| 4755 | self.assertEqual(ssock.recv(1024), TEST_DATA) |
| 4756 | |
| 4757 | def test_session(self): |
| 4758 | client_context, server_context, hostname = testing_context() |
nothing calls this directly
no test coverage detected