import asyncio, socket, ssl
from test.support import findfile

CERT = findfile('keycert.pem', subdir='certdata')

async def handler(reader, writer):
    await reader.readline()
    writer.write(bytes(16 << 20))
    writer.close()
    try:
        await asyncio.wait_for(writer.wait_closed(), 5)
    except TimeoutError as e:
        print('server:', repr(e) if e.args else 'still open after 5 s')
        writer.transport.abort()

def client(port, half_close):
    ctx = ssl.create_default_context()
    ctx.check_hostname = False
    ctx.verify_mode = ssl.CERT_NONE
    s = ctx.wrap_socket(socket.create_connection(('127.0.0.1', port)))
    s.sendall(b'x\n')
    if half_close:
        s.shutdown(socket.SHUT_WR)
    return s  # never read

async def main():
    sctx = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
    sctx.load_cert_chain(CERT)
    server = await asyncio.start_server(
        handler, '127.0.0.1', 0, ssl=sctx, ssl_shutdown_timeout=1)
    port = server.sockets[0].getsockname()[1]
    for half_close in (False, True):
        print('peer half-closes:', half_close)
        s = await asyncio.to_thread(client, port, half_close)
        await asyncio.sleep(6)
        s.close()
    server.close()
    await server.wait_closed()

asyncio.run(main())

peer half-closes: False
server: TimeoutError('SSL shutdown timed out')
peer half-closes: True
server: still open after 5 s