(aconn_cls, aconn, dsn)
| 109 | @pytest.mark.slow |
| 110 | @pytest.mark.timing |
| 111 | async def test_notify_timeout(aconn_cls, aconn, dsn): |
| 112 | await aconn.set_autocommit(True) |
| 113 | await aconn.execute("listen foo") |
| 114 | |
| 115 | async def notifier(): |
| 116 | async with await aconn_cls.connect(dsn, autocommit=True) as nconn: |
| 117 | await asleep(0.25) |
| 118 | await nconn.execute("notify foo, '1'") |
| 119 | |
| 120 | worker = spawn(notifier) |
| 121 | try: |
| 122 | times = [time()] |
| 123 | async for n in aconn.notifies(timeout=0.5): |
| 124 | times.append(time()) |
| 125 | times.append(time()) |
| 126 | finally: |
| 127 | await gather(worker) |
| 128 | |
| 129 | assert len(times) == 3 |
| 130 | assert times[1] - times[0] == pytest.approx(0.25, 0.1) |
| 131 | assert times[2] - times[1] == pytest.approx(0.25, 0.1) |
| 132 | |
| 133 | |
| 134 | @pytest.mark.slow |
nothing calls this directly
no test coverage detected