Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
assert r.status_code == 200
assert r.text == "Hello, world!"
traces = tracer.writer.pop_traces()
assert len(traces) == 1
spans: List[Span] = traces[0]
assert len(spans) == 1
span = spans[0]
assert span.span_id
assert span.trace_id
assert span.parent_id is None
assert span.name == "asgi.request"
assert span.service == "test.asgi.service"
assert span.resource == "GET /"
assert span.get_tag(http_ext.STATUS_CODE) == "200"
assert span.get_tag(http_ext.URL) == "http://testserver/"
assert span.get_tag(http_ext.QUERY_STRING) is None
def test_route_failure(self):
app = molten.App(routes=[molten.Route('/hello/{name}/{age}', hello)])
client = TestClient(app)
response = client.get('/goodbye')
spans = self.tracer.writer.pop()
self.assertEqual(response.status_code, 404)
span = spans[0]
self.assertEqual(span.service, 'molten')
self.assertEqual(span.name, 'molten.request')
self.assertEqual(span.resource, 'GET 404')
self.assertEqual(span.get_tag(http.URL), 'http://127.0.0.1:8000/goodbye')
self.assertEqual(span.get_tag('http.method'), 'GET')
self.assertEqual(span.get_tag('http.status_code'), '404')
fqs = ''
response = self.fetch('/success/' + fqs)
assert 200 == response.code
traces = self.tracer.writer.pop_traces()
assert 1 == len(traces)
assert 1 == len(traces[0])
request_span = traces[0][0]
assert 'tornado-web' == request_span.service
assert 'tornado.request' == request_span.name
assert 'web' == request_span.span_type
assert 'tests.contrib.tornado.web.app.SuccessHandler' == request_span.resource
assert 'GET' == request_span.get_tag('http.method')
assert '200' == request_span.get_tag('http.status_code')
assert self.get_url('/success/') == request_span.get_tag(http.URL)
if config.tornado.trace_query_string:
assert query_string == request_span.get_tag(http.QUERY_STRING)
else:
assert http.QUERY_STRING not in request_span.meta
assert 0 == request_span.error
def test_nested_wrap_handler(self):
# it should trace a handler that calls a coroutine
response = self.fetch('/nested_wrap/')
assert 200 == response.code
traces = self.tracer.writer.pop_traces()
assert 1 == len(traces)
assert 2 == len(traces[0])
# check request span
request_span = traces[0][0]
assert 'tornado-web' == request_span.service
assert 'tornado.request' == request_span.name
assert 'web' == request_span.span_type
assert 'tests.contrib.tornado.web.app.NestedWrapHandler' == request_span.resource
assert 'GET' == request_span.get_tag('http.method')
assert '200' == request_span.get_tag('http.status_code')
assert self.get_url('/nested_wrap/') == request_span.get_tag(http.URL)
assert 0 == request_span.error
# check nested span
nested_span = traces[0][1]
assert 'tornado-web' == nested_span.service
assert 'tornado.coro' == nested_span.name
assert 0 == nested_span.error
# check durations because of the yield sleep
assert request_span.duration >= 0.05
assert nested_span.duration >= 0.05
assert rv.status_code == 200
assert rv.data == b'\xc3\xbc\xc5\x8b\xc3\xaf\xc4\x89\xc3\xb3\xc4\x91\xc4\x93'
# ensure trace worked
assert not self.tracer.current_span(), self.tracer.current_span().pprint()
spans = self.tracer.writer.pop()
assert len(spans) == 1
s = spans[0]
assert s.service == 'test.flask.service'
assert s.resource == u'üŋïĉóđē'
assert s.start >= start
assert s.duration <= end - start
assert s.error == 0
assert s.meta.get(http.STATUS_CODE) == '200'
assert s.meta.get(http.METHOD) == 'GET'
assert s.meta.get(http.URL) == u'http://localhost/üŋïĉóđē'
def test_middleware_trace_errors(self):
# ensures that the internals are properly traced
url = reverse('forbidden-view')
response = self.client.get(url)
assert response.status_code == 403
# check for spans
spans = self.tracer.writer.pop()
assert len(spans) == 1
span = spans[0]
assert span.get_tag('http.status_code') == '403'
assert span.get_tag(http.URL) == 'http://testserver/fail-view/'
assert span.resource == 'tests.contrib.django.app.views.ForbiddenView'
def test_list_no_match(self):
span = Span(name='Name', tracer=None)
span.set_tag(URL, r'http://cooldomain.example.com')
filtr = FilterRequestsOnUrl([r'http://domain\.example\.com', r'http://anotherdomain\.example\.com'])
trace = filtr.process_trace([span])
self.assertIsNotNone(trace)
assert response.status_code == 404
# check for spans
spans = self.tracer.writer.pop()
assert len(spans) == 2
sp_request = spans[0]
sp_template = spans[1]
# Template
# DEV: The template name is `unknown` because unless they define a `404.html`
# django generates the template from a string, which will not have a `Template.name` set
assert sp_template.get_tag('django.template_name') == 'unknown'
# Request
assert sp_request.get_tag('http.status_code') == '404'
assert sp_request.get_tag(http.URL) == 'http://testserver/unknown-url'
assert sp_request.get_tag('django.user.is_authenticated') == 'False'
assert sp_request.get_tag('http.method') == 'GET'
assert sp_request.span_type == 'http'
assert sp_request.resource == 'django.views.defaults.page_not_found'
try:
self.simulate_get('/exception')
except Exception:
pass
else:
assert 0
traces = self.tracer.writer.pop_traces()
assert len(traces) == 1
assert len(traces[0]) == 1
span = traces[0][0]
assert span.name == 'falcon.request'
assert span.service == self._service
assert span.resource == 'GET tests.contrib.falcon.app.resources.ResourceException'
assert span.get_tag(httpx.STATUS_CODE) == '500'
assert span.get_tag(httpx.URL) == 'http://falconframework.org/exception'
assert span.parent_id is None
if (
config.analytics_enabled and settings.ANALYTICS_ENABLED is not False
) or settings.ANALYTICS_ENABLED is True:
span.set_tag(
ANALYTICS_SAMPLE_RATE_KEY,
settings.ANALYTICS_SAMPLE_RATE
)
span.set_tag(http.METHOD, request.method)
# Do not fail if we cannot build an absoluate uri, use the request path as a fallback
try:
span.set_tag(http.URL, request.build_absolute_uri(request.path))
except Exception:
log.debug('failed to build absolute uri from %r', request.path, exc_info=True)
span.set_tag(http.URL, request.path)
_set_req_span(request, span)
except Exception:
log.debug('error tracing request', exc_info=True)