Function test_stop_after_batch

tests/test_notify.py:183–200

(conn_cls, conn, dsn)

@pytest.mark.timing
def test_stop_after_batch(conn_cls, conn, dsn):
    conn.set_autocommit(True)
    conn.execute("listen foo")

    def notifier():
        with conn_cls.connect(dsn, autocommit=True) as nconn:
            with nconn.transaction():
                nconn.execute("notify foo, '1'")
                nconn.execute("notify foo, '2'")

    worker = spawn(notifier)
    try:
        ns = list(conn.notifies(timeout=1.0, stop_after=1))
        assert len(ns) == 2
        assert ns[0].payload == "1"
        assert ns[1].payload == "2"
    finally:
        gather(worker)


@pytest.mark.slow