Add unit tests for wait*() functions

We check that when a generator waits for a Wait value, it gets a Ready
value that matches. The socket we wait on is read- and write-ready.

These tests hang on non-Linux platform in CI, perhaps because some
socket operations (e.g. fileno()) are not portable, so we only run them
on Linux.

As is, wait_epoll() fails this test because it assumes that readiness is
either read or write, not both.

diff --git a/tests/test_waiting.py b/tests/test_waiting.py
index deb35ea6594302bb0453043ef8c6fc9a25ea215b..21d94b144a0ff1ed184754a49fb3a4ba414bdfae 100644 (file)
--- a/tests/test_waiting.py
+++ b/tests/test_waiting.py
@@ -1,4 +1,6 @@
 import select
+import socket
+import sys
 
 import pytest
 
@@ -8,9 +10,8 @@ from psycopg import generators
 from psycopg.pq import ConnStatus, ExecStatus
 
 
-skip_no_epoll = pytest.mark.skipif(
-    not hasattr(select, "epoll"), reason="epoll not available"
-)
+hasepoll = hasattr(select, "epoll")
+skip_no_epoll = pytest.mark.skipif(not hasepoll, reason="epoll not available")
 
 timeouts = [
     {},
@@ -21,6 +22,11 @@ timeouts = [
 ]
 
 
+skip_if_not_linux = pytest.mark.skipif(
+    not sys.platform.startswith("linux"), reason="non-Linux platform"
+)
+
+
 @pytest.mark.parametrize("timeout", timeouts)
 def test_wait_conn(dsn, timeout, retries):
     for retry in retries:
@@ -44,6 +50,31 @@ def test_wait(pgconn, timeout):
     assert res.status == ExecStatus.TUPLES_OK
 
 
+waits_and_ids = [
+    (waiting.wait, "wait"),
+    (waiting.wait_selector, "wait_selector"),
+]
+if hasepoll:
+    waits_and_ids.append((waiting.wait_epoll, "wait_epoll"))
+
+waits, wids = list(zip(*waits_and_ids))
+
+readys = [waiting.Ready.R, waiting.Ready.W, waiting.Ready.R | waiting.Ready.W]
+
+
+@pytest.mark.parametrize("waitfn", waits, ids=wids)
+@pytest.mark.parametrize("wait, ready", zip(waiting.Wait, readys))
+@skip_if_not_linux
+def test_wait_ready(waitfn, wait, ready):
+    def gen():
+        r = yield wait
+        return r
+
+    with socket.socket() as s:
+        r = waitfn(gen(), s.fileno())
+    assert r & ready
+
+
 @pytest.mark.parametrize("timeout", timeouts)
 def test_wait_selector(pgconn, timeout):
     pgconn.send_query(b"select 1")
@@ -100,6 +131,19 @@ async def test_wait_async(pgconn):
     assert res.status == ExecStatus.TUPLES_OK
 
 
+@pytest.mark.asyncio
+@pytest.mark.parametrize("wait, ready", zip(waiting.Wait, readys))
+@skip_if_not_linux
+async def test_wait_ready_async(wait, ready):
+    def gen():
+        r = yield wait
+        return r
+
+    with socket.socket() as s:
+        r = await waiting.wait_async(gen(), s.fileno())
+    assert r & ready
+
+
 @pytest.mark.asyncio
 async def test_wait_async_bad(pgconn):
     pgconn.send_query(b"select 1")