Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def make_client(self, mock_socket_values, **kwargs):
tracer = get_dummy_tracer()
Pin.override(pymemcache, tracer=tracer)
self.client = pymemcache.client.base.Client((TEST_HOST, TEST_PORT), **kwargs)
self.client.sock = MockSocket(list(mock_socket_values))
return self.client
def unpatch():
"""Remove pymemcache tracing"""
if not getattr(pymemcache.client, '_datadog_patch', False):
return
setattr(pymemcache.client, '_datadog_patch', False)
setattr(pymemcache.client.base, 'Client', _Client)
# Remove any pins that may exist on the pymemcache reference
setattr(pymemcache, _DD_PIN_NAME, None)
setattr(pymemcache, _DD_PIN_PROXY_NAME, None)
def _failure_translator():
"""Translates common pymemcache exceptions into tooz exceptions.
https://github.com/pinterest/pymemcache/blob/d995/pymemcache/client.py#L202
"""
try:
yield
except pymemcache_client.MemcacheUnexpectedCloseError as e:
utils.raise_with_cause(coordination.ToozConnectionError,
encodeutils.exception_to_unicode(e),
cause=e)
except (socket.timeout, socket.error,
socket.gaierror, socket.herror) as e:
# TODO(harlowja): get upstream pymemcache to produce a better
# exception for these, using socket (vs. a memcache specific
# error) seems sorta not right and/or the best approach...
msg = encodeutils.exception_to_unicode(e)
if e.errno is not None:
msg += " (with errno %s [%s])" % (errno.errorcode[e.errno],
e.errno)
utils.raise_with_cause(coordination.ToozConnectionError,
msg, cause=e)
except pymemcache_client.MemcacheError as e:
utils.raise_with_cause(tooz.ToozError,