File size: 3,463 Bytes
4879fc7 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 | """Direct unit coverage of app/notify.py's Notifier internals.
This file is deliberately narrow. Most of the Notifier's behaviour is already
proven end-to-end through the real app by test_longpoll_api.py and
test_updates_api.py. Three behaviours, though, are subtle enough — and internal
enough — that the API-level tests only exercise them indirectly, as a side
effect of some other assertion, rather than pinning them directly:
- waking a parked waiter from a foreign OS thread, which is the exact bridge
(``loop.call_soon_threadsafe`` on the future captured at ``register`` time)
production relies on when the verifier's worker thread announces a verdict
from outside the event loop;
- a wake's latch absorbing a signal that arrives before ``wait`` is ever
called, so a wake racing a client's reconnect is never lost;
- double-wake idempotency, so two writers touching the same key in quick
succession don't double-count or raise.
Kept here rather than folded into an API test because sinking a bug in any of
these three would surface only as an intermittent, hard-to-reproduce timing
flake three layers up (a long-poll that occasionally holds for its full
timeout instead of waking promptly) — a direct unit test turns that into a
deterministic, immediate failure at the primitive itself.
"""
from __future__ import annotations
import asyncio
import threading
import time
from app.notify import Notifier
def _notifier(per_owner: int = 4, total: int = 256) -> Notifier:
# Mirrors production Settings defaults (config.py) for the two spread
# knobs: a threshold of 20 keeps every wake in these tests (at most a
# couple of subscriptions) on the instant path, never the spread-out-over-
# wake_spread_s path meant for large broadcasts.
return Notifier(
max_waiters_per_owner=per_owner,
max_waiters_total=total,
wake_spread_s=8.0,
wake_spread_threshold=20,
)
def test_latch_absorbs_wake_before_wait():
n = _notifier()
async def scenario():
sub = n.register("a", {"k"})
n.wake({"k"}) # arrives before the first wait()
first = await sub.wait(0.01) # consumes the latch immediately
second = await sub.wait(0.05) # latch cleared -> times out
return first, second
first, second = asyncio.run(scenario())
assert first is True # not lost despite arriving between/around waits
assert second is False
def test_double_wake_is_idempotent():
n = _notifier()
async def scenario():
sub = n.register("a", {"k"})
n.wake({"k"})
n.wake({"k"}) # second wake must not error or double-count
first = await sub.wait(0.5)
second = await sub.wait(0.05) # only one latch was pending
return first, second
first, second = asyncio.run(scenario())
assert first is True
assert second is False
def test_wake_from_foreign_thread():
n = _notifier()
async def scenario():
sub = n.register("a", {"k"})
# Fire the wake from a plain OS thread while the coroutine is parked;
# the bridge is loop.call_soon_threadsafe on the captured future.
def waker():
time.sleep(0.05)
n.wake({"k"})
t = threading.Thread(target=waker)
t.start()
result = await sub.wait(2.0)
t.join(1.0)
return result
assert asyncio.run(scenario()) is True
|