(waitfn, event, request)
| 220 | @pytest.mark.parametrize("event", ["W", "RW"]) |
| 221 | @pytest.mark.parametrize("waitfn", waitfns) |
| 222 | def test_wait_w(waitfn, event, request): |
| 223 | # Test that wait functions handle waiting and returning state correctly |
| 224 | waitfn = getattr(waiting, waitfn) |
| 225 | wait = getattr(waiting.Wait, event) |
| 226 | |
| 227 | rs, ws = socket.socketpair() # the w socket is already ready for writing |
| 228 | rs.setblocking(False) |
| 229 | ws.setblocking(False) |
| 230 | with rs, ws: |
| 231 | t0 = time.time() |
| 232 | r = waitfn(tgen(wait), ws.fileno(), 0.5) |
| 233 | dt = time.time() - t0 |
| 234 | # Check timing and received waiting state |
| 235 | assert r == waiting.Ready.W |
| 236 | if check_timing(request): |
| 237 | assert dt < 0.1 |
| 238 | |
| 239 | |
| 240 | @pytest.mark.parametrize("waitfn", waitfns) |
nothing calls this directly
no test coverage detected