Function test_generator_and_handler

tests/test_notify_async.py:224–264

(aconn, aconn_cls, dsn, recwarn)

@pytest.mark.slow
@skip_free_threaded("warnings are context-local in the free-threaded build >= 3.14")
async def test_generator_and_handler(aconn, aconn_cls, dsn, recwarn):
    # NOTE: we don't support generator+handlers anymore. So, if in the future
    # this behaviour will change, we will not consider it a regression. However
    # we will want to keep the warning check.

    recwarn.clear()

    await aconn.set_autocommit(True)
    await aconn.execute("listen foo")

    n1 = None
    n2 = None

    def set_n2(n):
        nonlocal n2
        n2 = n

    aconn.add_notify_handler(set_n2)

    async def listener():
        nonlocal n1
        async for n1 in aconn.notifies(timeout=1, stop_after=1):
            pass

    worker = spawn(listener)
    try:
        # Make sure the listener is listening
        if not aconn.lock.locked():
            await asleep(0.01)

        async with await aconn_cls.connect(dsn, autocommit=True) as nconn:
            await nconn.execute("notify foo, '1'")

    finally:
        await gather(worker)

    assert n1
    assert n2

    msg = str(recwarn.pop(RuntimeWarning).message)
    assert "notifies()" in msg


@pytest.mark.parametrize("query_between", [True, False])