summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
authorJon Parise <jon@pinterest.com>2021-06-18 13:31:22 -0700
committerGitHub <noreply@github.com>2021-06-18 13:31:22 -0700
commit3506aa61d9e47f820461d88770998602c2cb5c2c (patch)
tree8e306907bdf3f3984d1af041033660414749f64d
parent7e243214c71661018ede3b927d11ef9172f84c4e (diff)
downloadpymemcache-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).
-rw-r--r--pymemcache/test/test_client.py24
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)