Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def setup_class(self):
""" Clear all spans before a test run """
self.recorder = tracer.recorder
self.cluster = Cluster('couchbase://%s' % testenv['couchdb_host'])
self.bucket = Bucket('couchbase://%s/travel-sample' % testenv['couchdb_host'],
username=testenv['couchdb_username'], password=testenv['couchdb_password'])
def setUp(self):
""" Clear all spans before a test run """
self.recorder = tracer.recorder
self.recorder.clear_spans()
self.logger = logging.getLogger('unit test')
def setUp(self):
""" Clear all spans before a test run """
self.recorder = tracer.recorder
self.recorder.clear_spans()
self.channel = grpc.insecure_channel(testenv["grpc_server"])
self.server_stub = stan_pb2_grpc.StanStub(self.channel)
# The grpc client apparently needs a second to connect and initialize
time.sleep(1)
def setUp(self):
logger.warning("Connecting to MongoDB mongo://%s:@%s:%s",
testenv['mongodb_user'], testenv['mongodb_host'], testenv['mongodb_port'])
self.conn = pymongo.MongoClient(host=testenv['mongodb_host'], port=int(testenv['mongodb_port']),
username=testenv['mongodb_user'], password=testenv['mongodb_pw'])
self.conn.test.records.delete_many(filter={})
self.recorder = tracer.recorder
self.recorder.clear_spans()
def setUp(self):
""" Clear all spans before a test run """
self.http = urllib3.PoolManager()
self.recorder = tracer.recorder
self.recorder.clear_spans()
def setUp(self):
""" Clear all spans before a test run """
self.recorder = tracer.recorder
self.recorder.clear_spans()
self.session = Session()
def test_error_reporting(celery_app, celery_worker):
result = None
with tracer.start_active_span('test'):
result = will_raise_error.apply_async()
# Wait for jobs to finish
time.sleep(0.5)
spans = filter_out_ping_tasks(tracer.recorder.queued_spans())
assert len(spans) == 4
filter = lambda span: span.n == "sdk"
test_span = get_first_span_by_filter(spans, filter)
assert(test_span)
filter = lambda span: span.n == "celery-client"
client_span = get_first_span_by_filter(spans, filter)
assert(client_span)
filter = lambda span: span.n == "log"
log_span = get_first_span_by_filter(spans, filter)
assert(log_span)
filter = lambda span: span.n == "celery-worker"
worker_span = get_first_span_by_filter(spans, filter)
def setUp(self):
""" Clear all spans before a test run """
self.recorder = tracer.recorder
self.recorder.clear_spans()
self.http = urllib3.PoolManager()
def setUp(self):
logger.info("MySQL connecting: %s:@%s:3306/%s", testenv['mysql_user'], testenv['mysql_host'], testenv['mysql_db'])
self.db = MySQLdb.connect(host=testenv['mysql_host'], port=testenv['mysql_port'],
user=testenv['mysql_user'], passwd=testenv['mysql_pw'],
db=testenv['mysql_db'])
self.cursor = self.db.cursor()
self.recorder = tracer.recorder
self.recorder.clear_spans()
tracer.cur_ctx = None
def start(self, _):
"""
Starts the agent and required threads
This method is called after a successful announce. See fsm.py
"""
logger.debug("Spawning metric & span reporting threads")
self.should_threads_shutdown.clear()
self.sensor.start()
instana.singletons.tracer.recorder.start()