Method test_create_server_ssl_1

Lib/test/test_asyncio/test_ssl.py:200–290

(self, size)

    @support.bigmemtest(size=25, memuse=90*2**20, dry_run=False)
    def test_create_server_ssl_1(self, size):
        CNT = 0           # number of clients that were successful
        TOTAL_CNT = size  # total number of clients that test will create
        TIMEOUT = support.LONG_TIMEOUT  # timeout for this test

        A_DATA = b'A' * 1024 * BUF_MULTIPLIER
        B_DATA = b'B' * 1024 * BUF_MULTIPLIER

        sslctx = self._create_server_ssl_context(
            test_utils.ONLYCERT, test_utils.ONLYKEY
        )
        client_sslctx = self._create_client_ssl_context()

        clients = []

        async def handle_client(reader, writer):
            nonlocal CNT

            data = await reader.readexactly(len(A_DATA))
            self.assertEqual(data, A_DATA)
            writer.write(b'OK')

            data = await reader.readexactly(len(B_DATA))
            self.assertEqual(data, B_DATA)
            writer.writelines([b'SP', bytearray(b'A'), memoryview(b'M')])

            await writer.drain()
            writer.close()

            CNT += 1

        async def test_client(addr):
            fut = asyncio.Future()

            def prog(sock):
                try:
                    sock.starttls(client_sslctx)
                    sock.connect(addr)
                    sock.send(A_DATA)

                    data = sock.recv_all(2)
                    self.assertEqual(data, b'OK')

                    sock.send(B_DATA)
                    data = sock.recv_all(4)
                    self.assertEqual(data, b'SPAM')

                    sock.close()

                except Exception as ex:
                    self.loop.call_soon_threadsafe(fut.set_exception, ex)
                else:
                    self.loop.call_soon_threadsafe(fut.set_result, None)

            client = self.tcp_client(prog)
            client.start()
            clients.append(client)