Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/httpcore/_utils.py: 56%

1from __future__ import annotations 

2 

3import select 

4import socket 

5import sys 

6 

7 

8def is_socket_readable(sock: socket.socket | None) -> bool: 

9 """ 

10 Return whether a socket, as identifed by its file descriptor, is readable. 

11 "A socket is readable" means that the read buffer isn't empty, i.e. that calling 

12 .recv() on it would immediately return some data. 

13 """ 

14 # NOTE: we want check for readability without actually attempting to read, because 

15 # we don't want to block forever if it's not readable. 

16 

17 # In the case that the socket no longer exists, or cannot return a file 

18 # descriptor, we treat it as being readable, as if it the next read operation 

19 # on it is ready to return the terminating `b""`. 

20 sock_fd = None if sock is None else sock.fileno() 

21 if sock_fd is None or sock_fd < 0: # pragma: nocover 

22 return True 

23 

24 # The implementation below was stolen from: 

25 # https://github.com/python-trio/trio/blob/20ee2b1b7376db637435d80e266212a35837ddcc/trio/_socket.py#L471-L478 

26 # See also: https://github.com/encode/httpcore/pull/193#issuecomment-703129316 

27 

28 # Use select.select on Windows, and when poll is unavailable and select.poll 

29 # everywhere else. (E.g. When eventlet is in use. See #327) 

30 if ( 

31 sys.platform == "win32" or getattr(select, "poll", None) is None 

32 ): # pragma: nocover 

33 rready, _, _ = select.select([sock_fd], [], [], 0) 

34 return bool(rready) 

35 p = select.poll() 

36 p.register(sock_fd, select.POLLIN) 

37 return bool(p.poll(0))