(self, size)
| 1038 | |
| 1039 | @support.bigmemtest(size=25, memuse=90*2**20, dry_run=False) |
| 1040 | def test_create_server_ssl_over_ssl(self, size): |
| 1041 | CNT = 0 # number of clients that were successful |
| 1042 | TOTAL_CNT = size # total number of clients that test will create |
| 1043 | TIMEOUT = support.LONG_TIMEOUT # timeout for this test |
| 1044 | |
| 1045 | A_DATA = b'A' * 1024 * BUF_MULTIPLIER |
| 1046 | B_DATA = b'B' * 1024 * BUF_MULTIPLIER |
| 1047 | |
| 1048 | sslctx_1 = self._create_server_ssl_context( |
| 1049 | test_utils.ONLYCERT, test_utils.ONLYKEY) |
| 1050 | client_sslctx_1 = self._create_client_ssl_context() |
| 1051 | sslctx_2 = self._create_server_ssl_context( |
| 1052 | test_utils.ONLYCERT, test_utils.ONLYKEY) |
| 1053 | client_sslctx_2 = self._create_client_ssl_context() |
| 1054 | |
| 1055 | clients = [] |
| 1056 | |
| 1057 | async def handle_client(reader, writer): |
| 1058 | nonlocal CNT |
| 1059 | |
| 1060 | data = await reader.readexactly(len(A_DATA)) |
| 1061 | self.assertEqual(data, A_DATA) |
| 1062 | writer.write(b'OK') |
| 1063 | |
| 1064 | data = await reader.readexactly(len(B_DATA)) |
| 1065 | self.assertEqual(data, B_DATA) |
| 1066 | writer.writelines([b'SP', bytearray(b'A'), memoryview(b'M')]) |
| 1067 | |
| 1068 | await writer.drain() |
| 1069 | writer.close() |
| 1070 | |
| 1071 | CNT += 1 |
| 1072 | |
| 1073 | class ServerProtocol(asyncio.StreamReaderProtocol): |
| 1074 | def connection_made(self, transport): |
| 1075 | super_ = super() |
| 1076 | transport.pause_reading() |
| 1077 | fut = self._loop.create_task(self._loop.start_tls( |
| 1078 | transport, self, sslctx_2, server_side=True)) |
| 1079 | |
| 1080 | def cb(_): |
| 1081 | try: |
| 1082 | tr = fut.result() |
| 1083 | except Exception as ex: |
| 1084 | super_.connection_lost(ex) |
| 1085 | else: |
| 1086 | super_.connection_made(tr) |
| 1087 | fut.add_done_callback(cb) |
| 1088 | |
| 1089 | def server_protocol_factory(): |
| 1090 | reader = asyncio.StreamReader() |
| 1091 | protocol = ServerProtocol(reader, handle_client) |
| 1092 | return protocol |
| 1093 | |
| 1094 | async def test_client(addr): |
| 1095 | fut = asyncio.Future() |
| 1096 | |
| 1097 | def prog(sock): |
nothing calls this directly
no test coverage detected