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_200(self):
# setup our test app
@self.app.route('/hi/')
def hi(name):
return 'hi %s' % name
self._trace_app(self.tracer)
# make a request
resp = self.app.get('/hi/dougie')
assert resp.status_int == 200
assert compat.to_unicode(resp.body) == u'hi dougie'
# validate it's traced
spans = self.tracer.writer.pop()
assert len(spans) == 1
s = spans[0]
assert s.name == 'bottle.request'
assert s.service == 'bottle-app'
assert s.resource == 'GET /hi/'
assert s.get_tag('http.status_code') == '200'
assert s.get_tag('http.method') == 'GET'
services = self.tracer.writer.pop_services()
assert services == {}
def test_to_unicode_string(self):
# Calling `compat.to_unicode` on a non-unicode string
res = to_unicode(b'test')
assert type(res) == unicode
assert res == 'test'
def test_to_unicode_unicode_string(self):
# Calling `compat.to_unicode` on a unicode string
res = to_unicode(u'ÿ')
assert type(res) == unicode
assert res == u'ÿ'
def _handle_server_exception(server_context, span):
if server_context is not None and \
hasattr(server_context, '_state') and \
server_context._state is not None:
code = to_unicode(server_context._state.code)
details = to_unicode(server_context._state.details)
span.error = 1
span.set_tag(errors.ERROR_MSG, details)
span.set_tag(errors.ERROR_TYPE, code)
def format_log_message(msg, *args, **kwargs):
if args:
msg = msg % args
pid = None
pname = None
tname = None
tid = None
try:
pid = os.getpid()
thread = threading.current_thread()
tname = to_unicode(thread.name)
tid = threading.get_ident()
process = multiprocessing.current_process()
pname = to_unicode(process.name)
except Exception:
pass
return to_unicode('({}:{}) ({}:{}) {}').format(pname, pid, tname, tid, msg)
def func(response):
try:
# pull out response code from gRPC response to use both for `grpc.status.code`
# tag and the error type tag if the response is an exception
response_code = response.code()
# cast code to unicode for tags
status_code = to_unicode(response_code)
span.set_tag(constants.GRPC_STATUS_CODE_KEY, status_code)
if response_code != grpc.StatusCode.OK:
_handle_error(span, response, status_code)
finally:
span.finish()
def _handle_server_exception(server_context, span):
if server_context is not None and \
hasattr(server_context, '_state') and \
server_context._state is not None:
code = to_unicode(server_context._state.code)
details = to_unicode(server_context._state.details)
span.error = 1
span.set_tag(errors.ERROR_MSG, details)
span.set_tag(errors.ERROR_TYPE, code)
# the request isn't guaranteed to exist here, so only use it carefully.
method = ''
endpoint = ''
url = ''
if request:
method = request.method
endpoint = request.endpoint or code
url = request.base_url or ''
# Let users specify their own resource in middleware if they so desire.
# See case https://github.com/DataDog/dd-trace-py/issues/353
if span.resource == SPAN_NAME:
resource = endpoint or code
span.resource = compat.to_unicode(resource).lower()
span.set_tag(http.URL, compat.to_unicode(url))
span.set_tag(http.STATUS_CODE, code)
span.set_tag(http.METHOD, method)
span.finish()