diff --git a/tests/relay/test_receiver_admission.py b/tests/relay/test_receiver_admission.py index 389c2d4c..c136db0b 100644 --- a/tests/relay/test_receiver_admission.py +++ b/tests/relay/test_receiver_admission.py @@ -373,6 +373,12 @@ async def observed_handshake(channel): self.assert_empty() async def test_saturation_counters_and_diagnostics_are_bounded_and_sanitized(self): + # Waiting-room registration is not a barrier for client lifecycle + # callbacks. Retire this unrelated observer before capturing the exact + # admission-only warning count; keep product diagnostics unchanged. + self.relay_listener.cancel() + await asyncio.wait_for(asyncio.gather(self.relay_listener, return_exceptions=True), wire.TIMEOUT) + self.assertTrue(self.relay_listener.cancelled()) self.receiver.diagnostic_events = True async def blocked(raw): await asyncio.Future() @@ -420,6 +426,37 @@ async def blocked(raw): self.assertEqual(counters['abortFailed'], 100) self.assert_empty() + async def test_saturation_capture_waits_for_background_observer_to_retire(self): + # Replace the owned network listener with a controlled observer. A + # queued lifecycle callback must retire before exact admission-only + # log arithmetic starts, including when cancellation yields once. + self.relay_listener.cancel() + await asyncio.wait_for(asyncio.gather(self.relay_listener, return_exceptions=True), wire.TIMEOUT) + started, release, retired = asyncio.Event(), asyncio.Event(), asyncio.Event() + + async def observer(): + started.set() + try: + await release.wait() + finally: + await asyncio.sleep(0) + self.receiver.lifecycle_event('websocket_open', elapsedMs=0, attemptId=None) + retired.set() + + self.relay_listener = asyncio.create_task(observer()) + await asyncio.wait_for(started.wait(), wire.TIMEOUT) + original_close = self.receiver.close + + async def close_after_callback(): + release.set() + await asyncio.wait_for(retired.wait(), wire.TIMEOUT) + await original_close() + + with patch.object(self.receiver, 'close', new=close_after_callback): + await self.test_saturation_counters_and_diagnostics_are_bounded_and_sanitized() + self.assertTrue(retired.is_set()) + self.assertTrue(self.relay_listener.cancelled()) + async def test_close_rejects_new_spawn_during_and_after_shutdown(self): entered, finish = asyncio.Event(), asyncio.Event() raw = self.fake_raw()