diff options
author | Jon Parise <jon@pinterest.com> | 2021-06-18 13:31:22 -0700 |
---|---|---|
committer | GitHub <noreply@github.com> | 2021-06-18 13:31:22 -0700 |
commit | 3506aa61d9e47f820461d88770998602c2cb5c2c (patch) | |
tree | 8e306907bdf3f3984d1af041033660414749f64d /pymemcache | |
parent | 7e243214c71661018ede3b927d11ef9172f84c4e (diff) | |
download | pymemcache-3506aa61d9e47f820461d88770998602c2cb5c2c.tar.gz |
Provide a mock implementation of socket.getaddrinfo (#327)
9551dfd0 introduced a call to socket.getaddrinfo() to support IPv6, but
we never added an implementation of that function to MockSocketModule.
This resulted in some tests making "live" socket.getaddrinfo() calls
because of the default MockSocketModule.__getattr__ implementation
(which we need to forward other module attribute lookups).
Diffstat (limited to 'pymemcache')
-rw-r--r-- | pymemcache/test/test_client.py | 24 |
1 files changed, 20 insertions, 4 deletions
diff --git a/pymemcache/test/test_client.py b/pymemcache/test/test_client.py index 24ecab2..93d327c 100644 --- a/pymemcache/test/test_client.py +++ b/pymemcache/test/test_client.py @@ -40,6 +40,11 @@ from pymemcache import pool from pymemcache.test.utils import MockMemcacheClient +# TODO: Use ipaddress module when dropping support for Python < 3.3 +def is_ipv6(address): + return re.match(r'^[0-9a-f:]+$', address) + + class MockSocket(object): def __init__(self, recv_bufs, connect_failure=None, close_failure=None): self.recv_bufs = collections.deque(recv_bufs) @@ -53,10 +58,8 @@ class MockSocket(object): @property def family(self): - # TODO: Use ipaddress module when dropping support for Python < 3.3 - ipv6_re = re.compile(r'^[0-9a-f:]+$') - is_ipv6 = any(ipv6_re.match(c[0]) for c in self.connections) - return socket.AF_INET6 if is_ipv6 else socket.AF_INET + any_ipv6 = any(is_ipv6(c[0]) for c in self.connections) + return socket.AF_INET6 if any_ipv6 else socket.AF_INET def sendall(self, value): self.send_bufs.append(value) @@ -115,6 +118,19 @@ class MockSocketModule(object): self.sockets.append(socket) return socket + def getaddrinfo(self, host, port, family=0, type=0, proto=0, flags=0): + family = family or ( + socket.AF_INET6 if is_ipv6(host) else socket.AF_INET + ) + type = type or socket.SOCK_STREAM + proto = proto or socket.IPPROTO_TCP + sockaddr = ( + ('::1', 11211, 0, 0) + if family == socket.AF_INET6 + else ('127.0.0.1', 11211) + ) + return [(family, type, proto, '', sockaddr)] + def __getattr__(self, name): return getattr(socket, name) |