Function test_generator_and_handler

tests/test_notify.py:228–267

(conn, conn_cls, dsn, recwarn)

@pytest.mark.slow
@skip_free_threaded("warnings are context-local in the free-threaded build >= 3.14")
def test_generator_and_handler(conn, conn_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()

    conn.set_autocommit(True)
    conn.execute("listen foo")

    n1 = None
    n2 = None

    def set_n2(n):
        nonlocal n2
        n2 = n

    conn.add_notify_handler(set_n2)

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

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

        with conn_cls.connect(dsn, autocommit=True) as nconn:
            nconn.execute("notify foo, '1'")
    finally:
        gather(worker)

    assert n1
    assert n2

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


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