Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_error(self):
with self.server_span:
with self.assertRaises(MemcacheClientError):
self.context.memcache.cas("key", b"value", b"whatever")
server_span_observer = self.baseplate_observer.get_only_child()
span_observer = server_span_observer.get_only_child()
self.assertEqual(span_observer.span.name, "memcache.cas")
self.assertTrue(span_observer.on_start_called)
self.assertTrue(span_observer.on_finish_called)
self.assertIsNotNone(span_observer.on_finish_exc_info)
def test_set_client_error(self):
client = self.make_client([b'CLIENT_ERROR some message\r\n'])
def _set():
client.set('key', 'value', noreply=False)
pytest.raises(MemcacheClientError, _set)
spans = self.check_spans(1, ['set'], ['set key'])
self.assertEqual(spans[0].error, 1)
def _raise_errors(self, line, name):
if line.startswith(b'ERROR'):
raise MemcacheUnknownCommandError(name)
if line.startswith(b'CLIENT_ERROR'):
error = line[line.find(b' ') + 1:]
raise MemcacheClientError(error)
if line.startswith(b'SERVER_ERROR'):
error = line[line.find(b' ') + 1:]
raise MemcacheServerError(error)
)
# try to set relevant tags, catch any exceptions so we don't mess
# with the application
try:
span.set_tags(p.tags)
vals = _get_query_string(args)
query = '{}{}{}'.format(method_name, ' ' if vals else '', vals)
span.set_tag(memcachedx.QUERY, query)
except Exception:
log.debug('Error setting relevant pymemcache tags')
try:
return method(*args, **kwargs)
except (
MemcacheClientError,
MemcacheServerError,
MemcacheUnknownCommandError,
MemcacheUnknownError,
MemcacheIllegalInputError,
):
(typ, val, tb) = sys.exc_info()
span.set_exc_info(typ, val, tb)
reraise(typ, val, tb)