cpython: 038543d34166 (original) (raw)

Mercurial > cpython

changeset 85481:038543d34166

Switch the AF_* and SOCK_* constants in the socket module to IntEnum. Closes #18720. [#18720]

line wrap: on

line diff

--- a/Lib/socket.py +++ b/Lib/socket.py @@ -48,6 +48,7 @@ import _socket from _socket import import os, sys, io +from enum import IntEnum try: import errno @@ -60,6 +61,30 @@ EWOULDBLOCK = getattr(errno, 'EWOULDBLOC all = ["getfqdn", "create_connection"] all.extend(os._get_exports_list(socket)) +# Set up the socket.AF socket.SOCK_* constants as members of IntEnums for +# nicer string representations. +# Note that _socket only knows about the integer values. The public interface +# in this module understands the enums and translates them back from integers +# where needed (e.g. .family property of a socket object). +AddressFamily = IntEnum('AddressFamily',

•                    {name: value for name, value in globals().items()[](#l1.21)

•                     if name.isupper() and name.startswith('AF_')})[](#l1.22)

+globals().update(AddressFamily.members) + +SocketType = IntEnum('SocketType',

•                 {name: value for name, value in globals().items()[](#l1.26)

•                  if name.isupper() and name.startswith('SOCK_')})[](#l1.27)

+globals().update(SocketType.members) + +def _intenum_converter(value, enum_klass):

• """Convert a numeric family value to an IntEnum member.

+

• If it's not a known member, return the numeric value itself.
• """
• try:

•    return enum_klass(value)[](#l1.36)

• except ValueError:

•    return value[](#l1.38)

_realsocket = socket @@ -91,6 +116,10 @@ class socket(_socket.socket): slots = ["weakref", "_io_refs", "_closed"] def init(self, family=AF_INET, type=SOCK_STREAM, proto=0, fileno=None):

•    # For user code address family and type values are IntEnum members, but[](#l1.46)

•    # for the underlying _socket.socket they're just integers. The[](#l1.47)

•    # constructor of _socket.socket converts the given argument to an[](#l1.48)

•    # integer automatically.[](#l1.49)
     _socket.socket.__init__(self, family, type, proto, fileno)[](#l1.50)
     self._io_refs = 0[](#l1.51)
     self._closed = False[](#l1.52)

@@ -230,6 +259,18 @@ class socket(_socket.socket): self._closed = True return super().detach()

• @property
• def family(self):

•    """Read-only access to the address family for this socket.[](#l1.59)

•    """[](#l1.60)

•    return _intenum_converter(super().family, AddressFamily)[](#l1.61)

+

• @property
• def type(self):

•    """Read-only access to the socket type.[](#l1.65)

•    """[](#l1.66)

•    return _intenum_converter(super().type, SocketType)[](#l1.67)

+ if os.name == 'nt': def get_inheritable(self): return os.get_handle_inheritable(self.fileno()) @@ -243,7 +284,6 @@ class socket(socket.socket): get_inheritable.doc = "Get the inheritable flag of the socket" set_inheritable.doc = "Set the inheritable flag of the socket" - def fromfd(fd, family, type, proto=0): """ fromfd(fd, family, type[, proto]) -> socket object @@ -469,3 +509,27 @@ def create_connection(address, timeout= raise err else: raise error("getaddrinfo returns an empty list") + +def getaddrinfo(host, port, family=0, type=0, proto=0, flags=0):

• """Resolve host and port into list of address info entries.

+

• Translate the host/port argument into a sequence of 5-tuples that contain
• all the necessary arguments for creating a socket connected to that service.
• host is a domain name, a string representation of an IPv4/v6 address or
• None. port is a string service name such as 'http', a numeric port number or
• None. By passing None as the value of host and port, you can pass NULL to
• the underlying C API.

+

• The family, type and proto arguments can be optionally specified in order to
• narrow the list of addresses returned. Passing zero as a value for each of
• these arguments selects the full range of results.
• """

• We override this function since we want to translate the numeric family

• and socket type values to enum constants.

• addrlist = []
• for res in _socket.getaddrinfo(host, port, family, type, proto, flags):

•    af, socktype, proto, canonname, sa = res[](#l1.103)

•    addrlist.append((_intenum_converter(af, AddressFamily),[](#l1.104)

•                     _intenum_converter(socktype, SocketType),[](#l1.105)

•                     proto, canonname, sa))[](#l1.106)

• return addrlist

--- a/Lib/test/test_socket.py +++ b/Lib/test/test_socket.py @@ -1161,9 +1161,12 @@ class GeneralModuleTests(unittest.TestCa socket.getaddrinfo(HOST, 80) socket.getaddrinfo(HOST, None) # test family and socktype filters

•    infos = socket.getaddrinfo(HOST, None, socket.AF_INET)[](#l2.7)

•    for family, _, _, _, _ in infos:[](#l2.8)

•    infos = socket.getaddrinfo(HOST, 80, socket.AF_INET, socket.SOCK_STREAM)[](#l2.9)

•    for family, type, _, _, _ in infos:[](#l2.10)
         self.assertEqual(family, socket.AF_INET)[](#l2.11)

•        self.assertEqual(str(family), 'AddressFamily.AF_INET')[](#l2.12)

•        self.assertEqual(type, socket.SOCK_STREAM)[](#l2.13)

•        self.assertEqual(str(type), 'SocketType.SOCK_STREAM')[](#l2.14)
     infos = socket.getaddrinfo(HOST, None, 0, socket.SOCK_STREAM)[](#l2.15)
     for _, socktype, _, _, _ in infos:[](#l2.16)
         self.assertEqual(socktype, socket.SOCK_STREAM)[](#l2.17)

@@ -1321,6 +1324,27 @@ class GeneralModuleTests(unittest.TestCa with socket.socket(socket.AF_INET6, socket.SOCK_STREAM) as s: self.assertRaises(OverflowError, s.bind, (support.HOSTv6, 0, -10))

• def test_str_for_enums(self):

•    # Make sure that the AF_* and SOCK_* constants have enum-like string[](#l2.23)

•    # reprs.[](#l2.24)

•    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:[](#l2.25)

•        self.assertEqual(str(s.family), 'AddressFamily.AF_INET')[](#l2.26)

•        self.assertEqual(str(s.type), 'SocketType.SOCK_STREAM')[](#l2.27)

+

• @unittest.skipIf(os.name == 'nt', 'Will not work on Windows')
• def test_uknown_socket_family_repr(self):

•    # Test that when created with a family that's not one of the known[](#l2.31)

•    # AF_*/SOCK_* constants, socket.family just returns the number.[](#l2.32)

•    #[](#l2.33)

•    # To do this we fool socket.socket into believing it already has an[](#l2.34)

•    # open fd because on this path it doesn't actually verify the family and[](#l2.35)

•    # type and populates the socket object.[](#l2.36)

•    #[](#l2.37)

•    # On Windows this trick won't work, so the test is skipped.[](#l2.38)

•    fd, _ = tempfile.mkstemp()[](#l2.39)

•    with socket.socket(family=42424, type=13331, fileno=fd) as s:[](#l2.40)

•        self.assertEqual(s.family, 42424)[](#l2.41)

•        self.assertEqual(s.type, 13331)[](#l2.42)

@unittest.skipUnless(HAVE_SOCKET_CAN, 'SocketCan required for this test.') class BasicCANTest(unittest.TestCase):