def test_conns()

in asfpy/aioldap.py [0:0]

• 30 lines of code
• 4 McCabe index (conditional complexity)

def test_conns(client):
    # Run three tasks in parallel: heartbeat, connA, connB. The latter
    # two will open, run a couple LDAP queries, then close. Then re-open.
    # These will run in one event loop (the LDAPClient has some private
    # loops; no peeking).
    #
    # Should see: smooth heartbeat, even when an artifical delay is
    # introduced to the connect() process.

    # Some random services to read
    SERVICE_BASE = 'cn=%s,ou=groups,ou=services,dc=apache,dc=org'
    SERVICES = [ 'board', 'infrastructure-root', 'asf-secretary', ]

    # Start hanging tasks off this.
    loop = asyncio.new_event_loop()

    t0 = loop.time()
    def ts():
        return f'[{loop.time() - t0 :.2f}]'

    async def print_me():
        async with client.connect() as conn:
            print('ME:', await conn.whoami())

    async def heartbeat():
        while True:
            print(f'{ts()} heartbeat')
            await asyncio.sleep(2)

    import random
    async def conn_usage(name):
        while True:
            # Stagger the connections
            await asyncio.sleep(random.randint(0, 3))
            async with client.connect() as conn:
                for _ in range(5):
                    s = random.choice(SERVICES)
                    rv = await conn.search(SERVICE_BASE % (s,), ['owner', 'member',])
                    print(f'{ts()} CONN[{name}]: RV=', rv)
                    await asyncio.sleep(random.randint(1, 3))
                # between reconnections
                await asyncio.sleep(random.randint(0, 5))

    loop.create_task(print_me())
    loop.create_task(heartbeat())
    loop.create_task(conn_usage('A'))
    loop.create_task(conn_usage('B'))

    loop.run_forever()