Add test for wait_for_socket EINTR handling w/ infinite timeout
This covers an otherwise uncovered branch in _retry_on_intr.
As requested here:
import signal
import socket
+import threading
try:
from time import monotonic
except ImportError:
from time import time as monotonic
+import time
import pytest
@@ -132,3 +134,44 @@ def test_eintr(wfs, spair):
signal.signal(signal.SIGALRM, old_handler)
assert interrupt_count[0] > 0
+
+
+@pytest.mark.skipif(
+ not hasattr(signal, "setitimer"),
+ reason="need setitimer() support"
+)
+@pytest.mark.parametrize("wfs", variants)
+def test_eintr_infinite_timeout(wfs, spair):
+ a, b = spair
+ interrupt_count = [0]
+
+ def handler(sig, frame):
+ assert sig == signal.SIGALRM
+ interrupt_count[0] += 1
+
+ def make_a_readable_after_one_second():
+ time.sleep(1)
+ b.send(b"x")
+
+ old_handler = signal.signal(signal.SIGALRM, handler)
+ try:
+ assert not wfs(a, read=True, timeout=0)
+ start = monotonic()
+ try:
+ # Start delivering SIGALRM 10 times per second
+ signal.setitimer(signal.ITIMER_REAL, 0.1, 0.1)
+ # Sleep for 1 second (we hope!)
+ thread = threading.Thread(target=make_a_readable_after_one_second)
+ thread.start()
+ wfs(a, read=True)
+ finally:
+ # Stop delivering SIGALRM
+ signal.setitimer(signal.ITIMER_REAL, 0)
+ thread.join()
+ end = monotonic()
+ dur = end - start
+ assert 0.9 < dur < 3
+ finally:
+ signal.signal(signal.SIGALRM, old_handler)
+
+ assert interrupt_count[0] > 0
