Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def span(self):
yield Span(tracer, 'some_span')
def span(self):
yield Span(tracer, 'some_span')
from ddtrace import tracer
if __name__ == '__main__':
assert tracer.enabled
print('Test success')
from ddtrace import tracer
if __name__ == '__main__':
assert tracer.writer.api.hostname == '172.10.0.1'
assert tracer.writer.api.port == 8120
print('Test success')
def test_unpatch_patch(self):
""" Tests unpatch-patch cycle """
unpatch()
self.assertIsNone(Pin.get_from(molten))
molten_client()
spans = self.tracer.writer.pop()
self.assertEqual(len(spans), 0)
patch()
# Need to override Pin here as we do in setUp
Pin.override(molten, tracer=self.tracer)
self.assertTrue(Pin.get_from(molten) is not None)
molten_client()
spans = self.tracer.writer.pop()
self.assertTrue(len(spans) > 0)
def test_pin_does_not_override_global(self):
# ensure that when a `Pin` is created from a class, the specific
# instance doesn't override the global one
class A(object):
pass
Pin.override(A, service='metrics')
global_pin = Pin.get_from(A)
global_pin._config['distributed_tracing'] = True
a = A()
pin = Pin.get_from(a)
assert pin is not None
assert pin._config['distributed_tracing'] is True
pin._config['distributed_tracing'] = False
assert global_pin._config['distributed_tracing'] is True
assert pin._config['distributed_tracing'] is False
def test_pin_does_not_override_global_with_new_instance(self):
# ensure that when a `Pin` is created from a class, the specific
# instance doesn't override the global one, even if only the
# `onto()` API has been used
class A(object):
pass
pin = Pin(service='metrics')
pin.onto(A)
global_pin = Pin.get_from(A)
global_pin._config['distributed_tracing'] = True
a = A()
pin = Pin.get_from(a)
assert pin is not None
assert pin._config['distributed_tracing'] is True
pin._config['distributed_tracing'] = False
assert global_pin._config['distributed_tracing'] is True
assert pin._config['distributed_tracing'] is False
def test_patch_app(self):
# When celery.App is patched it must include a `Pin` instance
app = celery.Celery()
assert Pin.get_from(app) is not None
def _get_conn_and_tracer(self):
conn = self._conn = yield from aiopg.connect(**POSTGRES_CONFIG)
Pin.get_from(conn).clone(tracer=self.tracer).onto(conn)
return conn, self.tracer
def test_patch_before_import(self):
from ddtrace import patch
patch(celery=True)
import celery
app = celery.Celery()
assert Pin.get_from(app) is not None