diff --git a/tests/frameworks/test_starlette.py b/tests/frameworks/test_starlette.py index 6df31c996..ad67f4aa4 100644 --- a/tests/frameworks/test_starlette.py +++ b/tests/frameworks/test_starlette.py @@ -1,272 +1,275 @@ # (c) Copyright IBM Corp. 2021 # (c) Copyright Instana Inc. 2020 +import multiprocessing import time import pytest import requests -import multiprocessing +import unittest from ..helpers import testenv from instana.singletons import tracer from ..helpers import get_first_span_by_filter -@pytest.fixture(scope="module") -def server(): - from tests.apps.starlette_app import launch_starlette - proc = multiprocessing.Process(target=launch_starlette, args=(), daemon=True) - proc.start() - time.sleep(2) - yield - proc.kill() # Kill server after tests - -def test_vanilla_get(server): - result = requests.get(testenv["starlette_server"] + '/') - assert(result) - spans = tracer.recorder.queued_spans() - # Starlette instrumentation (like all instrumentation) _always_ traces unless told otherwise - assert len(spans) == 1 - assert spans[0].n == 'asgi' - - assert "X-INSTANA-T" in result.headers - assert "X-INSTANA-S" in result.headers - assert "X-INSTANA-L" in result.headers - assert result.headers["X-INSTANA-L"] == '1' - assert "Server-Timing" in result.headers - -def test_basic_get(server): - result = None - with tracer.start_active_span('test'): - result = requests.get(testenv["starlette_server"] + '/') - assert(result) - - spans = tracer.recorder.queued_spans() - assert len(spans) == 3 - - span_filter = lambda span: span.n == "sdk" and span.data['sdk']['name'] == 'test' - test_span = get_first_span_by_filter(spans, span_filter) - assert(test_span) - - span_filter = lambda span: span.n == "urllib3" - urllib3_span = get_first_span_by_filter(spans, span_filter) - assert(urllib3_span) - - span_filter = lambda span: span.n == 'asgi' - asgi_span = get_first_span_by_filter(spans, span_filter) - assert(asgi_span) - - assert(test_span.t == urllib3_span.t == asgi_span.t) - assert(asgi_span.p == urllib3_span.s) - assert(urllib3_span.p == test_span.s) - - assert "X-INSTANA-T" in result.headers - assert result.headers["X-INSTANA-T"] == asgi_span.t - assert "X-INSTANA-S" in result.headers - assert result.headers["X-INSTANA-S"] == asgi_span.s - assert "X-INSTANA-L" in result.headers - assert result.headers["X-INSTANA-L"] == '1' - assert "Server-Timing" in result.headers - assert result.headers["Server-Timing"] == ("intid;desc=%s" % asgi_span.t) - - assert(asgi_span.ec == None) - assert (asgi_span.data['http']['host'] == '127.0.0.1') - assert (asgi_span.data['http']['path'] == '/') - assert (asgi_span.data['http']['path_tpl'] == '/') - assert (asgi_span.data['http']['method'] == 'GET') - assert (asgi_span.data['http']['status'] == 200) - assert (asgi_span.data['http']['error'] is None) - assert (asgi_span.data['http']['params'] is None) - -def test_path_templates(server): - result = None - with tracer.start_active_span('test'): - result = requests.get(testenv["starlette_server"] + '/users/1') - - assert(result) - - spans = tracer.recorder.queued_spans() - assert len(spans) == 3 - - span_filter = lambda span: span.n == "sdk" and span.data['sdk']['name'] == 'test' - test_span = get_first_span_by_filter(spans, span_filter) - assert(test_span) - - span_filter = lambda span: span.n == "urllib3" - urllib3_span = get_first_span_by_filter(spans, span_filter) - assert(urllib3_span) - - span_filter = lambda span: span.n == 'asgi' - asgi_span = get_first_span_by_filter(spans, span_filter) - assert(asgi_span) - - assert(test_span.t == urllib3_span.t == asgi_span.t) - assert(asgi_span.p == urllib3_span.s) - assert(urllib3_span.p == test_span.s) - - assert "X-INSTANA-T" in result.headers - assert result.headers["X-INSTANA-T"] == asgi_span.t - assert "X-INSTANA-S" in result.headers - assert result.headers["X-INSTANA-S"] == asgi_span.s - assert "X-INSTANA-L" in result.headers - assert result.headers["X-INSTANA-L"] == '1' - assert "Server-Timing" in result.headers - assert result.headers["Server-Timing"] == ("intid;desc=%s" % asgi_span.t) - - assert(asgi_span.ec == None) - assert (asgi_span.data['http']['host'] == '127.0.0.1') - assert (asgi_span.data['http']['path'] == '/users/1') - assert (asgi_span.data['http']['path_tpl'] == '/users/{user_id}') - assert (asgi_span.data['http']['method'] == 'GET') - assert (asgi_span.data['http']['status'] == 200) - assert (asgi_span.data['http']['error'] is None) - assert (asgi_span.data['http']['params'] is None) - -def test_secret_scrubbing(server): - result = None - with tracer.start_active_span('test'): - result = requests.get(testenv["starlette_server"] + '/?secret=shhh') - - assert(result) - - spans = tracer.recorder.queued_spans() - assert len(spans) == 3 - - span_filter = lambda span: span.n == "sdk" and span.data['sdk']['name'] == 'test' - test_span = get_first_span_by_filter(spans, span_filter) - assert(test_span) - - span_filter = lambda span: span.n == "urllib3" - urllib3_span = get_first_span_by_filter(spans, span_filter) - assert(urllib3_span) - - span_filter = lambda span: span.n == 'asgi' - asgi_span = get_first_span_by_filter(spans, span_filter) - assert(asgi_span) - - assert(test_span.t == urllib3_span.t == asgi_span.t) - assert(asgi_span.p == urllib3_span.s) - assert(urllib3_span.p == test_span.s) - - assert "X-INSTANA-T" in result.headers - assert result.headers["X-INSTANA-T"] == asgi_span.t - assert "X-INSTANA-S" in result.headers - assert result.headers["X-INSTANA-S"] == asgi_span.s - assert "X-INSTANA-L" in result.headers - assert result.headers["X-INSTANA-L"] == '1' - assert "Server-Timing" in result.headers - assert result.headers["Server-Timing"] == ("intid;desc=%s" % asgi_span.t) - - assert(asgi_span.ec == None) - assert (asgi_span.data['http']['host'] == '127.0.0.1') - assert (asgi_span.data['http']['path'] == '/') - assert (asgi_span.data['http']['path_tpl'] == '/') - assert (asgi_span.data['http']['method'] == 'GET') - assert (asgi_span.data['http']['status'] == 200) - assert (asgi_span.data['http']['error'] is None) - assert (asgi_span.data['http']['params'] == 'secret=') - -def test_synthetic_request(server): - request_headers = { - 'X-INSTANA-SYNTHETIC': '1' - } - with tracer.start_active_span('test'): - result = requests.get(testenv["starlette_server"] + '/', headers=request_headers) - - assert(result) - - spans = tracer.recorder.queued_spans() - assert len(spans) == 3 - - span_filter = lambda span: span.n == "sdk" and span.data['sdk']['name'] == 'test' - test_span = get_first_span_by_filter(spans, span_filter) - assert(test_span) - - span_filter = lambda span: span.n == "urllib3" - urllib3_span = get_first_span_by_filter(spans, span_filter) - assert(urllib3_span) - - span_filter = lambda span: span.n == 'asgi' - asgi_span = get_first_span_by_filter(spans, span_filter) - assert(asgi_span) - - assert(test_span.t == urllib3_span.t == asgi_span.t) - assert(asgi_span.p == urllib3_span.s) - assert(urllib3_span.p == test_span.s) - - assert "X-INSTANA-T" in result.headers - assert result.headers["X-INSTANA-T"] == asgi_span.t - assert "X-INSTANA-S" in result.headers - assert result.headers["X-INSTANA-S"] == asgi_span.s - assert "X-INSTANA-L" in result.headers - assert result.headers["X-INSTANA-L"] == '1' - assert "Server-Timing" in result.headers - assert result.headers["Server-Timing"] == ("intid;desc=%s" % asgi_span.t) - - assert(asgi_span.ec == None) - assert (asgi_span.data['http']['host'] == '127.0.0.1') - assert (asgi_span.data['http']['path'] == '/') - assert (asgi_span.data['http']['path_tpl'] == '/') - assert (asgi_span.data['http']['method'] == 'GET') - assert (asgi_span.data['http']['status'] == 200) - assert (asgi_span.data['http']['error'] is None) - assert (asgi_span.data['http']['params'] is None) - - assert(asgi_span.sy) - assert(urllib3_span.sy is None) - assert(test_span.sy is None) - -def test_custom_header_capture(server): - from instana.singletons import agent - - # The background Starlette server is pre-configured with custom headers to capture - - request_headers = { - 'X-Capture-This': 'this', - 'X-Capture-That': 'that' - } - with tracer.start_active_span('test'): - result = requests.get(testenv["starlette_server"] + '/', headers=request_headers) - - assert(result) - - spans = tracer.recorder.queued_spans() - assert len(spans) == 3 - - span_filter = lambda span: span.n == "sdk" and span.data['sdk']['name'] == 'test' - test_span = get_first_span_by_filter(spans, span_filter) - assert(test_span) - - span_filter = lambda span: span.n == "urllib3" - urllib3_span = get_first_span_by_filter(spans, span_filter) - assert(urllib3_span) - - span_filter = lambda span: span.n == 'asgi' - asgi_span = get_first_span_by_filter(spans, span_filter) - assert(asgi_span) - - assert(test_span.t == urllib3_span.t == asgi_span.t) - assert(asgi_span.p == urllib3_span.s) - assert(urllib3_span.p == test_span.s) - - assert "X-INSTANA-T" in result.headers - assert result.headers["X-INSTANA-T"] == asgi_span.t - assert "X-INSTANA-S" in result.headers - assert result.headers["X-INSTANA-S"] == asgi_span.s - assert "X-INSTANA-L" in result.headers - assert result.headers["X-INSTANA-L"] == '1' - assert "Server-Timing" in result.headers - assert result.headers["Server-Timing"] == ("intid;desc=%s" % asgi_span.t) - - assert(asgi_span.ec == None) - assert (asgi_span.data['http']['host'] == '127.0.0.1') - assert (asgi_span.data['http']['path'] == '/') - assert (asgi_span.data['http']['path_tpl'] == '/') - assert (asgi_span.data['http']['method'] == 'GET') - assert (asgi_span.data['http']['status'] == 200) - assert (asgi_span.data['http']['error'] is None) - assert (asgi_span.data['http']['params'] is None) - - assert ("X-Capture-This" in asgi_span.data["http"]["header"]) - assert ("this" == asgi_span.data["http"]["header"]["X-Capture-This"]) - assert ("X-Capture-That" in asgi_span.data["http"]["header"]) - assert ("that" == asgi_span.data["http"]["header"]["X-Capture-That"]) +class TestStarlette(unittest.TestCase): + def setUp(self): + from tests.apps.starlette_app import launch_starlette + self.proc = multiprocessing.Process(target=launch_starlette, args=(), daemon=True) + self.proc.start() + time.sleep(2) + + def tearDown(self): + self.proc.kill() # Kill server after tests + + def test_vanilla_get(self): + result = requests.get(testenv["starlette_server"] + '/') + self.assertTrue(result) + spans = tracer.recorder.queued_spans() + # Starlette instrumentation (like all instrumentation) _always_ traces unless told otherwise + self.assertEqual(len(spans), 1) + self.assertEqual(spans[0].n, 'asgi') + + self.assertIn("X-INSTANA-T", result.headers) + self.assertIn("X-INSTANA-S", result.headers) + self.assertIn("X-INSTANA-L", result.headers) + self.assertEqual(result.headers["X-INSTANA-L"], '1') + self.assertIn("Server-Timing", result.headers) + + def test_basic_get(self): + result = None + with tracer.start_active_span('test'): + result = requests.get(testenv["starlette_server"] + '/') + + self.assertTrue(result) + + spans = tracer.recorder.queued_spans() + self.assertEqual(len(spans), 3) + + span_filter = lambda span: span.n == "sdk" and span.data['sdk']['name'] == 'test' + test_span = get_first_span_by_filter(spans, span_filter) + self.assertTrue(test_span) + + span_filter = lambda span: span.n == "urllib3" + urllib3_span = get_first_span_by_filter(spans, span_filter) + self.assertTrue(urllib3_span) + + span_filter = lambda span: span.n == 'asgi' + asgi_span = get_first_span_by_filter(spans, span_filter) + self.assertTrue(asgi_span) + + self.assertTrue(test_span.t == urllib3_span.t == asgi_span.t) + self.assertEqual(asgi_span.p, urllib3_span.s) + self.assertEqual(urllib3_span.p, test_span.s) + + self.assertIn("X-INSTANA-T", result.headers) + self.assertEqual(result.headers["X-INSTANA-T"], asgi_span.t) + self.assertIn("X-INSTANA-S", result.headers) + self.assertEqual(result.headers["X-INSTANA-S"], asgi_span.s) + self.assertIn("X-INSTANA-L", result.headers) + self.assertEqual(result.headers["X-INSTANA-L"], '1') + self.assertIn("Server-Timing", result.headers) + self.assertEqual(result.headers["Server-Timing"], ("intid;desc=%s" % asgi_span.t)) + + self.assertIsNone(asgi_span.ec) + self.assertEqual(asgi_span.data['http']['host'], '127.0.0.1') + self.assertEqual(asgi_span.data['http']['path'], '/') + self.assertEqual(asgi_span.data['http']['path_tpl'], '/') + self.assertEqual(asgi_span.data['http']['method'], 'GET') + self.assertEqual(asgi_span.data['http']['status'], 200) + self.assertIsNone(asgi_span.data['http']['error']) + self.assertIsNone(asgi_span.data['http']['params']) + + def test_path_templates(self): + result = None + with tracer.start_active_span('test'): + result = requests.get(testenv["starlette_server"] + '/users/1') + + self.assertTrue(result) + + spans = tracer.recorder.queued_spans() + self.assertEqual(len(spans), 3) + + span_filter = lambda span: span.n == "sdk" and span.data['sdk']['name'] == 'test' + test_span = get_first_span_by_filter(spans, span_filter) + self.assertTrue(test_span) + + span_filter = lambda span: span.n == "urllib3" + urllib3_span = get_first_span_by_filter(spans, span_filter) + self.assertTrue(urllib3_span) + + span_filter = lambda span: span.n == 'asgi' + asgi_span = get_first_span_by_filter(spans, span_filter) + self.assertTrue(asgi_span) + + self.assertTrue(test_span.t == urllib3_span.t == asgi_span.t) + self.assertEqual(asgi_span.p, urllib3_span.s) + self.assertEqual(urllib3_span.p, test_span.s) + + self.assertIn("X-INSTANA-T", result.headers) + self.assertEqual(result.headers["X-INSTANA-T"], asgi_span.t) + self.assertIn("X-INSTANA-S", result.headers) + self.assertEqual( result.headers["X-INSTANA-S"], asgi_span.s) + self.assertIn("X-INSTANA-L", result.headers) + self.assertEqual(result.headers["X-INSTANA-L"], '1') + self.assertIn("Server-Timing", result.headers) + self.assertEqual(result.headers["Server-Timing"], ("intid;desc=%s" % asgi_span.t)) + + self.assertIsNone(asgi_span.ec) + self.assertEqual(asgi_span.data['http']['host'], '127.0.0.1') + self.assertEqual(asgi_span.data['http']['path'], '/users/1') + self.assertEqual(asgi_span.data['http']['path_tpl'], '/users/{user_id}') + self.assertEqual(asgi_span.data['http']['method'], 'GET') + self.assertEqual(asgi_span.data['http']['status'], 200) + self.assertIsNone(asgi_span.data['http']['error']) + self.assertIsNone(asgi_span.data['http']['params']) + + def test_secret_scrubbing(self): + result = None + with tracer.start_active_span('test'): + result = requests.get(testenv["starlette_server"] + '/?secret=shhh') + + self.assertTrue(result) + + spans = tracer.recorder.queued_spans() + assert len(spans) == 3 + + span_filter = lambda span: span.n == "sdk" and span.data['sdk']['name'] == 'test' + test_span = get_first_span_by_filter(spans, span_filter) + self.assertTrue(test_span) + + span_filter = lambda span: span.n == "urllib3" + urllib3_span = get_first_span_by_filter(spans, span_filter) + self.assertTrue(urllib3_span) + + span_filter = lambda span: span.n == 'asgi' + asgi_span = get_first_span_by_filter(spans, span_filter) + self.assertTrue(asgi_span) + + self.assertTrue(test_span.t == urllib3_span.t == asgi_span.t) + self.assertEqual(asgi_span.p, urllib3_span.s) + self.assertEqual(urllib3_span.p, test_span.s) + + self.assertIn("X-INSTANA-T", result.headers) + self.assertEqual(result.headers["X-INSTANA-T"], asgi_span.t) + self.assertIn("X-INSTANA-S", result.headers) + self.assertEqual(result.headers["X-INSTANA-S"], asgi_span.s) + self.assertIn("X-INSTANA-L", result.headers) + self.assertEqual(result.headers["X-INSTANA-L"], '1') + self.assertIn("Server-Timing", result.headers) + self.assertEqual(result.headers["Server-Timing"], ("intid;desc=%s" % asgi_span.t)) + + self.assertIsNone(asgi_span.ec) + self.assertEqual(asgi_span.data['http']['host'], '127.0.0.1') + self.assertEqual(asgi_span.data['http']['path'], '/') + self.assertEqual(asgi_span.data['http']['path_tpl'], '/') + self.assertEqual(asgi_span.data['http']['method'], 'GET') + self.assertEqual(asgi_span.data['http']['status'], 200) + self.assertIsNone(asgi_span.data['http']['error']) + self.assertEqual(asgi_span.data['http']['params'], 'secret=') + + def test_synthetic_request(self): + request_headers = { + 'X-INSTANA-SYNTHETIC': '1' + } + with tracer.start_active_span('test'): + result = requests.get(testenv["starlette_server"] + '/', headers=request_headers) + + self.assertTrue(result) + + spans = tracer.recorder.queued_spans() + assert len(spans) == 3 + + span_filter = lambda span: span.n == "sdk" and span.data['sdk']['name'] == 'test' + test_span = get_first_span_by_filter(spans, span_filter) + self.assertTrue(test_span) + + span_filter = lambda span: span.n == "urllib3" + urllib3_span = get_first_span_by_filter(spans, span_filter) + self.assertTrue(urllib3_span) + + span_filter = lambda span: span.n == 'asgi' + asgi_span = get_first_span_by_filter(spans, span_filter) + self.assertTrue(asgi_span) + + self.assertTrue(test_span.t == urllib3_span.t == asgi_span.t) + self.assertEqual(asgi_span.p, urllib3_span.s) + self.assertEqual(urllib3_span.p, test_span.s) + + self.assertIn("X-INSTANA-T", result.headers) + self.assertEqual(result.headers["X-INSTANA-T"], asgi_span.t) + self.assertIn("X-INSTANA-S", result.headers) + self.assertEqual(result.headers["X-INSTANA-S"], asgi_span.s) + self.assertIn("X-INSTANA-L", result.headers) + self.assertEqual(result.headers["X-INSTANA-L"], '1') + self.assertIn("Server-Timing", result.headers) + self.assertEqual(result.headers["Server-Timing"], ("intid;desc=%s" % asgi_span.t)) + + self.assertIsNone(asgi_span.ec) + self.assertEqual(asgi_span.data['http']['host'], '127.0.0.1') + self.assertEqual(asgi_span.data['http']['path'], '/') + self.assertEqual(asgi_span.data['http']['path_tpl'], '/') + self.assertEqual(asgi_span.data['http']['method'], 'GET') + self.assertEqual(asgi_span.data['http']['status'], 200) + self.assertIsNone(asgi_span.data['http']['error']) + self.assertIsNone(asgi_span.data['http']['params']) + + self.assertTrue(asgi_span.sy) + self.assertIsNone(urllib3_span.sy) + self.assertIsNone(test_span.sy) + + def test_custom_header_capture(self): + from instana.singletons import agent + + # The background Starlette server is pre-configured with custom headers to capture + + request_headers = { + 'X-Capture-This': 'this', + 'X-Capture-That': 'that' + } + with tracer.start_active_span('test'): + result = requests.get(testenv["starlette_server"] + '/', headers=request_headers) + + self.assertTrue(result) + + spans = tracer.recorder.queued_spans() + self.assertEqual(len(spans), 3) + + span_filter = lambda span: span.n == "sdk" and span.data['sdk']['name'] == 'test' + test_span = get_first_span_by_filter(spans, span_filter) + self.assertTrue(test_span) + + span_filter = lambda span: span.n == "urllib3" + urllib3_span = get_first_span_by_filter(spans, span_filter) + self.assertTrue(urllib3_span) + + span_filter = lambda span: span.n == 'asgi' + asgi_span = get_first_span_by_filter(spans, span_filter) + self.assertTrue(asgi_span) + + self.assertTrue(test_span.t == urllib3_span.t == asgi_span.t) + self.assertEqual(asgi_span.p, urllib3_span.s) + self.assertEqual(urllib3_span.p, test_span.s) + + self.assertIn("X-INSTANA-T", result.headers) + self.assertEqual(result.headers["X-INSTANA-T"], asgi_span.t) + self.assertIn("X-INSTANA-S", result.headers) + self.assertEqual( result.headers["X-INSTANA-S"], asgi_span.s) + self.assertIn("X-INSTANA-L", result.headers) + self.assertEqual( result.headers["X-INSTANA-L"], '1') + self.assertIn("Server-Timing", result.headers) + self.assertEqual(result.headers["Server-Timing"], ("intid;desc=%s" % asgi_span.t)) + + self.assertIsNone(asgi_span.ec) + self.assertEqual(asgi_span.data['http']['host'], '127.0.0.1') + self.assertEqual(asgi_span.data['http']['path'], '/') + self.assertEqual(asgi_span.data['http']['path_tpl'], '/') + self.assertEqual(asgi_span.data['http']['method'], 'GET') + self.assertEqual(asgi_span.data['http']['status'], 200) + self.assertIsNone(asgi_span.data['http']['error']) + self.assertIsNone(asgi_span.data['http']['params']) + + self.assertIn("X-Capture-This", asgi_span.data["http"]["header"]) + self.assertEqual("this", asgi_span.data["http"]["header"]["X-Capture-This"]) + self.assertIn("X-Capture-That", asgi_span.data["http"]["header"]) + self.assertEqual("that", asgi_span.data["http"]["header"]["X-Capture-That"]) diff --git a/tests/opentracing/test_ot_propagators.py b/tests/opentracing/test_ot_propagators.py index 01626e2cf..de26f4714 100644 --- a/tests/opentracing/test_ot_propagators.py +++ b/tests/opentracing/test_ot_propagators.py @@ -2,6 +2,7 @@ # (c) Copyright Instana Inc. 2020 import inspect +import unittest import opentracing as ot @@ -12,297 +13,297 @@ from instana.tracer import InstanaTracer -def test_http_basics(): - inspect.isclass(ihp.HTTPPropagator) +class TestOTSpan(unittest.TestCase): + def test_http_basics(self): + inspect.isclass(ihp.HTTPPropagator) - inject_func = getattr(ihp.HTTPPropagator, "inject", None) - assert inject_func - assert callable(inject_func) + inject_func = getattr(ihp.HTTPPropagator, "inject", None) + self.assertTrue(inject_func) + self.assertTrue(callable(inject_func)) - extract_func = getattr(ihp.HTTPPropagator, "extract", None) - assert extract_func - assert callable(extract_func) + extract_func = getattr(ihp.HTTPPropagator, "extract", None) + self.assertTrue(extract_func) + self.assertTrue(callable(extract_func)) -def test_http_inject_with_dict(): - ot.tracer = InstanaTracer() + def test_http_inject_with_dict(self): + ot.tracer = InstanaTracer() - carrier = {} - span = ot.tracer.start_span("unittest") - ot.tracer.inject(span.context, ot.Format.HTTP_HEADERS, carrier) + carrier = {} + span = ot.tracer.start_span("unittest") + ot.tracer.inject(span.context, ot.Format.HTTP_HEADERS, carrier) - assert 'X-INSTANA-T' in carrier - assert carrier['X-INSTANA-T'] == span.context.trace_id - assert 'X-INSTANA-S' in carrier - assert carrier['X-INSTANA-S'] == span.context.span_id - assert 'X-INSTANA-L' in carrier - assert carrier['X-INSTANA-L'] == "1" + self.assertIn('X-INSTANA-T', carrier) + self.assertEqual(carrier['X-INSTANA-T'], span.context.trace_id) + self.assertIn('X-INSTANA-S', carrier) + self.assertEqual(carrier['X-INSTANA-S'], span.context.span_id) + self.assertIn('X-INSTANA-L', carrier) + self.assertEqual(carrier['X-INSTANA-L'], "1") -def test_http_inject_with_list(): - ot.tracer = InstanaTracer() + def test_http_inject_with_list(self): + ot.tracer = InstanaTracer() - carrier = [] - span = ot.tracer.start_span("unittest") - ot.tracer.inject(span.context, ot.Format.HTTP_HEADERS, carrier) + carrier = [] + span = ot.tracer.start_span("unittest") + ot.tracer.inject(span.context, ot.Format.HTTP_HEADERS, carrier) - assert ('X-INSTANA-T', span.context.trace_id) in carrier - assert ('X-INSTANA-S', span.context.span_id) in carrier - assert ('X-INSTANA-L', "1") in carrier + self.assertIn(('X-INSTANA-T', span.context.trace_id), carrier) + self.assertIn(('X-INSTANA-S', span.context.span_id), carrier) + self.assertIn(('X-INSTANA-L', "1"), carrier) -def test_http_basic_extract(): - ot.tracer = InstanaTracer() + def test_http_basic_extract(self): + ot.tracer = InstanaTracer() - carrier = {'X-INSTANA-T': '1', 'X-INSTANA-S': '1', 'X-INSTANA-L': '1', 'X-INSTANA-SYNTHETIC': '1'} - ctx = ot.tracer.extract(ot.Format.HTTP_HEADERS, carrier) + carrier = {'X-INSTANA-T': '1', 'X-INSTANA-S': '1', 'X-INSTANA-L': '1', 'X-INSTANA-SYNTHETIC': '1'} + ctx = ot.tracer.extract(ot.Format.HTTP_HEADERS, carrier) - assert isinstance(ctx, SpanContext) - assert ctx.trace_id == '0000000000000001' - assert ctx.span_id == '0000000000000001' - assert ctx.synthetic + self.assertIsInstance(ctx, SpanContext) + self.assertEqual(ctx.trace_id, '0000000000000001') + self.assertEqual(ctx.span_id, '0000000000000001') + self.assertTrue(ctx.synthetic) -def test_http_extract_with_byte_keys(): - ot.tracer = InstanaTracer() + def test_http_extract_with_byte_keys(self): + ot.tracer = InstanaTracer() - carrier = {b'X-INSTANA-T': '1', b'X-INSTANA-S': '1', b'X-INSTANA-L': '1', b'X-INSTANA-SYNTHETIC': '1'} - ctx = ot.tracer.extract(ot.Format.HTTP_HEADERS, carrier) + carrier = {b'X-INSTANA-T': '1', b'X-INSTANA-S': '1', b'X-INSTANA-L': '1', b'X-INSTANA-SYNTHETIC': '1'} + ctx = ot.tracer.extract(ot.Format.HTTP_HEADERS, carrier) - assert isinstance(ctx, SpanContext) - assert ctx.trace_id == '0000000000000001' - assert ctx.span_id == '0000000000000001' - assert ctx.synthetic + self.assertIsInstance(ctx, SpanContext) + self.assertEqual(ctx.trace_id, '0000000000000001') + self.assertEqual(ctx.span_id, '0000000000000001') + self.assertTrue(ctx.synthetic) -def test_http_extract_from_list_of_tuples(): - ot.tracer = InstanaTracer() + def test_http_extract_from_list_of_tuples(self): + ot.tracer = InstanaTracer() - carrier = [(b'user-agent', b'python-requests/2.23.0'), (b'accept-encoding', b'gzip, deflate'), - (b'accept', b'*/*'), (b'connection', b'keep-alive'), - (b'x-instana-t', b'1'), (b'x-instana-s', b'1'), (b'x-instana-l', b'1'), (b'X-INSTANA-SYNTHETIC', '1')] - ctx = ot.tracer.extract(ot.Format.HTTP_HEADERS, carrier) + carrier = [(b'user-agent', b'python-requests/2.23.0'), (b'accept-encoding', b'gzip, deflate'), + (b'accept', b'*/*'), (b'connection', b'keep-alive'), + (b'x-instana-t', b'1'), (b'x-instana-s', b'1'), (b'x-instana-l', b'1'), (b'X-INSTANA-SYNTHETIC', '1')] + ctx = ot.tracer.extract(ot.Format.HTTP_HEADERS, carrier) - assert isinstance(ctx, SpanContext) - assert ctx.trace_id == '0000000000000001' - assert ctx.span_id == '0000000000000001' - assert ctx.synthetic + self.assertIsInstance(ctx, SpanContext) + self.assertEqual(ctx.trace_id, '0000000000000001') + self.assertEqual(ctx.span_id, '0000000000000001') + self.assertTrue(ctx.synthetic) -def test_http_mixed_case_extract(): - ot.tracer = InstanaTracer() + def test_http_mixed_case_extract(self): + ot.tracer = InstanaTracer() - carrier = {'x-insTana-T': '1', 'X-inSTANa-S': '1', 'X-INstana-l': '1'} - ctx = ot.tracer.extract(ot.Format.HTTP_HEADERS, carrier) + carrier = {'x-insTana-T': '1', 'X-inSTANa-S': '1', 'X-INstana-l': '1'} + ctx = ot.tracer.extract(ot.Format.HTTP_HEADERS, carrier) - assert isinstance(ctx, SpanContext) - assert ctx.trace_id == '0000000000000001' - assert ctx.span_id == '0000000000000001' - assert not ctx.synthetic + self.assertIsInstance(ctx, SpanContext) + self.assertEqual(ctx.trace_id, '0000000000000001') + self.assertEqual(ctx.span_id, '0000000000000001') + self.assertFalse(ctx.synthetic) -def test_http_extract_synthetic_only(): - ot.tracer = InstanaTracer() + def test_http_extract_synthetic_only(self): + ot.tracer = InstanaTracer() - carrier = {'X-INSTANA-SYNTHETIC': '1'} - ctx = ot.tracer.extract(ot.Format.HTTP_HEADERS, carrier) + carrier = {'X-INSTANA-SYNTHETIC': '1'} + ctx = ot.tracer.extract(ot.Format.HTTP_HEADERS, carrier) - assert isinstance(ctx, SpanContext) - assert ctx.trace_id is None - assert ctx.span_id is None - assert ctx.synthetic + self.assertIsInstance(ctx, SpanContext) + self.assertIsNone(ctx.trace_id) + self.assertIsNone(ctx.span_id) + self.assertTrue(ctx.synthetic) -def test_http_default_context_extract(): - ot.tracer = InstanaTracer() + def test_http_default_context_extract(self): + ot.tracer = InstanaTracer() - carrier = {} - ctx = ot.tracer.extract(ot.Format.HTTP_HEADERS, carrier) + carrier = {} + ctx = ot.tracer.extract(ot.Format.HTTP_HEADERS, carrier) - assert isinstance(ctx, SpanContext) - assert ctx.trace_id is None - assert ctx.span_id is None - assert ctx.synthetic is False + self.assertIsInstance(ctx, SpanContext) + self.assertIsNone(ctx.trace_id) + self.assertIsNone(ctx.span_id) + self.assertFalse(ctx.synthetic) -def test_http_128bit_headers(): - ot.tracer = InstanaTracer() + def test_http_128bit_headers(self): + ot.tracer = InstanaTracer() - carrier = {'X-INSTANA-T': '0000000000000000b0789916ff8f319f', - 'X-INSTANA-S': '0000000000000000b0789916ff8f319f', 'X-INSTANA-L': '1'} - ctx = ot.tracer.extract(ot.Format.HTTP_HEADERS, carrier) + carrier = {'X-INSTANA-T': '0000000000000000b0789916ff8f319f', + 'X-INSTANA-S': '0000000000000000b0789916ff8f319f', 'X-INSTANA-L': '1'} + ctx = ot.tracer.extract(ot.Format.HTTP_HEADERS, carrier) - assert isinstance(ctx, SpanContext) - assert ctx.trace_id == 'b0789916ff8f319f' - assert ctx.span_id == 'b0789916ff8f319f' + self.assertIsInstance(ctx, SpanContext) + self.assertEqual(ctx.trace_id, 'b0789916ff8f319f') + self.assertEqual(ctx.span_id, 'b0789916ff8f319f') -def test_text_basics(): - inspect.isclass(itp.TextPropagator) + def test_text_basics(self): + inspect.isclass(itp.TextPropagator) - inject_func = getattr(itp.TextPropagator, "inject", None) - assert inject_func - assert callable(inject_func) + inject_func = getattr(itp.TextPropagator, "inject", None) + self.assertTrue(inject_func) + self.assertTrue(callable(inject_func)) - extract_func = getattr(itp.TextPropagator, "extract", None) - assert extract_func - assert callable(extract_func) + extract_func = getattr(itp.TextPropagator, "extract", None) + self.assertTrue(extract_func) + self.assertTrue(callable(extract_func)) -def test_text_inject_with_dict(): - ot.tracer = InstanaTracer() + def test_text_inject_with_dict(self): + ot.tracer = InstanaTracer() - carrier = {} - span = ot.tracer.start_span("unittest") - ot.tracer.inject(span.context, ot.Format.TEXT_MAP, carrier) + carrier = {} + span = ot.tracer.start_span("unittest") + ot.tracer.inject(span.context, ot.Format.TEXT_MAP, carrier) - assert 'x-instana-t' in carrier - assert carrier['x-instana-t'] == span.context.trace_id - assert 'x-instana-s' in carrier - assert carrier['x-instana-s'] == span.context.span_id - assert 'x-instana-l' in carrier - assert carrier['x-instana-l'] == "1" + self.assertIn('x-instana-t', carrier) + self.assertEqual(carrier['x-instana-t'], span.context.trace_id) + self.assertIn('x-instana-s', carrier) + self.assertEqual(carrier['x-instana-s'], span.context.span_id) + self.assertIn('x-instana-l', carrier) + self.assertEqual(carrier['x-instana-l'], "1") -def test_text_inject_with_list(): - ot.tracer = InstanaTracer() + def test_text_inject_with_list(self): + ot.tracer = InstanaTracer() - carrier = [] - span = ot.tracer.start_span("unittest") - ot.tracer.inject(span.context, ot.Format.TEXT_MAP, carrier) + carrier = [] + span = ot.tracer.start_span("unittest") + ot.tracer.inject(span.context, ot.Format.TEXT_MAP, carrier) - assert ('x-instana-t', span.context.trace_id) in carrier - assert ('x-instana-s', span.context.span_id) in carrier - assert ('x-instana-l', "1") in carrier + self.assertIn(('x-instana-t', span.context.trace_id), carrier) + self.assertIn(('x-instana-s', span.context.span_id), carrier) + self.assertIn(('x-instana-l', "1"), carrier) -def test_text_basic_extract(): - ot.tracer = InstanaTracer() + def test_text_basic_extract(self): + ot.tracer = InstanaTracer() - carrier = {'x-instana-t': '1', 'x-instana-s': '1', 'x-instana-l': '1'} - ctx = ot.tracer.extract(ot.Format.TEXT_MAP, carrier) + carrier = {'x-instana-t': '1', 'x-instana-s': '1', 'x-instana-l': '1'} + ctx = ot.tracer.extract(ot.Format.TEXT_MAP, carrier) - assert isinstance(ctx, SpanContext) - assert ctx.trace_id == '0000000000000001' - assert ctx.span_id == '0000000000000001' + self.assertIsInstance(ctx, SpanContext) + self.assertEqual(ctx.trace_id, '0000000000000001') + self.assertEqual(ctx.span_id, '0000000000000001') -def test_text_mixed_case_extract(): - ot.tracer = InstanaTracer() + def test_text_mixed_case_extract(self): + ot.tracer = InstanaTracer() - carrier = {'x-insTana-T': '1', 'X-inSTANa-S': '1', 'X-INstana-l': '1'} - ctx = ot.tracer.extract(ot.Format.TEXT_MAP, carrier) + carrier = {'x-insTana-T': '1', 'X-inSTANa-S': '1', 'X-INstana-l': '1'} + ctx = ot.tracer.extract(ot.Format.TEXT_MAP, carrier) - assert isinstance(ctx, SpanContext) - assert ctx.trace_id == '0000000000000001' - assert ctx.span_id == '0000000000000001' + self.assertIsInstance(ctx, SpanContext) + self.assertEqual(ctx.trace_id, '0000000000000001') + self.assertEqual(ctx.span_id, '0000000000000001') -def test_text_default_context_extract(): - ot.tracer = InstanaTracer() + def test_text_default_context_extract(self): + ot.tracer = InstanaTracer() - carrier = {} - ctx = ot.tracer.extract(ot.Format.TEXT_MAP, carrier) + carrier = {} + ctx = ot.tracer.extract(ot.Format.TEXT_MAP, carrier) - assert isinstance(ctx, SpanContext) - assert ctx.trace_id is None - assert ctx.span_id is None - assert ctx.synthetic is False + self.assertIsInstance(ctx, SpanContext) + self.assertIsNone(ctx.trace_id) + self.assertIsNone(ctx.span_id) + self.assertFalse(ctx.synthetic) -def test_text_128bit_headers(): - ot.tracer = InstanaTracer() + def test_text_128bit_headers(self): + ot.tracer = InstanaTracer() - carrier = {'x-instana-t': '0000000000000000b0789916ff8f319f', - 'x-instana-s': ' 0000000000000000b0789916ff8f319f', 'X-INSTANA-L': '1'} - ctx = ot.tracer.extract(ot.Format.TEXT_MAP, carrier) + carrier = {'x-instana-t': '0000000000000000b0789916ff8f319f', + 'x-instana-s': ' 0000000000000000b0789916ff8f319f', 'X-INSTANA-L': '1'} + ctx = ot.tracer.extract(ot.Format.TEXT_MAP, carrier) - assert isinstance(ctx, SpanContext) - assert('b0789916ff8f319f' == ctx.span_id) - assert ctx.trace_id == 'b0789916ff8f319f' - assert ctx.span_id == 'b0789916ff8f319f' + self.assertIsInstance(ctx, SpanContext) + self.assertEqual(ctx.trace_id, 'b0789916ff8f319f') + self.assertEqual(ctx.span_id, 'b0789916ff8f319f') -def test_binary_basics(): - inspect.isclass(ibp.BinaryPropagator) + def test_binary_basics(self): + inspect.isclass(ibp.BinaryPropagator) - inject_func = getattr(ibp.BinaryPropagator, "inject", None) - assert inject_func - assert callable(inject_func) + inject_func = getattr(ibp.BinaryPropagator, "inject", None) + self.assertTrue(inject_func) + self.assertTrue(callable(inject_func)) - extract_func = getattr(ibp.BinaryPropagator, "extract", None) - assert extract_func - assert callable(extract_func) + extract_func = getattr(ibp.BinaryPropagator, "extract", None) + self.assertTrue(extract_func) + self.assertTrue(callable(extract_func)) -def test_binary_inject_with_dict(): - ot.tracer = InstanaTracer() + def test_binary_inject_with_dict(self): + ot.tracer = InstanaTracer() - carrier = {} - span = ot.tracer.start_span("unittest") - ot.tracer.inject(span.context, ot.Format.BINARY, carrier) + carrier = {} + span = ot.tracer.start_span("unittest") + ot.tracer.inject(span.context, ot.Format.BINARY, carrier) - assert b'x-instana-t' in carrier - assert carrier[b'x-instana-t'] == str.encode(span.context.trace_id) - assert b'x-instana-s' in carrier - assert carrier[b'x-instana-s'] == str.encode(span.context.span_id) - assert b'x-instana-l' in carrier - assert carrier[b'x-instana-l'] == b'1' + self.assertIn(b'x-instana-t', carrier) + self.assertEqual(carrier[b'x-instana-t'], str.encode(span.context.trace_id)) + self.assertIn(b'x-instana-s', carrier) + self.assertEqual(carrier[b'x-instana-s'], str.encode(span.context.span_id)) + self.assertIn(b'x-instana-l', carrier) + self.assertEqual(carrier[b'x-instana-l'], b'1') -def test_binary_inject_with_list(): - ot.tracer = InstanaTracer() + def test_binary_inject_with_list(self): + ot.tracer = InstanaTracer() - carrier = [] - span = ot.tracer.start_span("unittest") - ot.tracer.inject(span.context, ot.Format.BINARY, carrier) + carrier = [] + span = ot.tracer.start_span("unittest") + ot.tracer.inject(span.context, ot.Format.BINARY, carrier) - assert (b'x-instana-t', str.encode(span.context.trace_id)) in carrier - assert (b'x-instana-s', str.encode(span.context.span_id)) in carrier - assert (b'x-instana-l', b'1') in carrier + self.assertIn((b'x-instana-t', str.encode(span.context.trace_id)), carrier) + self.assertIn((b'x-instana-s', str.encode(span.context.span_id)), carrier) + self.assertIn((b'x-instana-l', b'1'), carrier) -def test_binary_basic_extract(): - ot.tracer = InstanaTracer() + def test_binary_basic_extract(self): + ot.tracer = InstanaTracer() - carrier = {b'X-INSTANA-T': b'1', b'X-INSTANA-S': b'1', b'X-INSTANA-L': b'1', b'X-INSTANA-SYNTHETIC': b'1'} - ctx = ot.tracer.extract(ot.Format.BINARY, carrier) + carrier = {b'X-INSTANA-T': b'1', b'X-INSTANA-S': b'1', b'X-INSTANA-L': b'1', b'X-INSTANA-SYNTHETIC': b'1'} + ctx = ot.tracer.extract(ot.Format.BINARY, carrier) - assert isinstance(ctx, SpanContext) - assert ctx.trace_id == '0000000000000001' - assert ctx.span_id == '0000000000000001' - assert ctx.synthetic + self.assertIsInstance(ctx, SpanContext) + self.assertEqual(ctx.trace_id, '0000000000000001') + self.assertEqual(ctx.span_id, '0000000000000001') + self.assertTrue(ctx.synthetic) -def test_binary_mixed_case_extract(): - ot.tracer = InstanaTracer() + def test_binary_mixed_case_extract(self): + ot.tracer = InstanaTracer() - carrier = {'x-insTana-T': '1', 'X-inSTANa-S': '1', 'X-INstana-l': '1', b'X-inStaNa-SYNtheTIC': b'1'} - ctx = ot.tracer.extract(ot.Format.BINARY, carrier) + carrier = {'x-insTana-T': '1', 'X-inSTANa-S': '1', 'X-INstana-l': '1', b'X-inStaNa-SYNtheTIC': b'1'} + ctx = ot.tracer.extract(ot.Format.BINARY, carrier) - assert isinstance(ctx, SpanContext) - assert ctx.trace_id == '0000000000000001' - assert ctx.span_id == '0000000000000001' - assert ctx.synthetic + self.assertIsInstance(ctx, SpanContext) + self.assertEqual(ctx.trace_id, '0000000000000001') + self.assertEqual(ctx.span_id, '0000000000000001') + self.assertTrue(ctx.synthetic) -def test_binary_default_context_extract(): - ot.tracer = InstanaTracer() + def test_binary_default_context_extract(self): + ot.tracer = InstanaTracer() - carrier = {} - ctx = ot.tracer.extract(ot.Format.BINARY, carrier) + carrier = {} + ctx = ot.tracer.extract(ot.Format.BINARY, carrier) - assert isinstance(ctx, SpanContext) - assert ctx.trace_id is None - assert ctx.span_id is None - assert ctx.synthetic is False + self.assertIsInstance(ctx, SpanContext) + self.assertIsNone(ctx.trace_id) + self.assertIsNone(ctx.span_id) + self.assertFalse(ctx.synthetic) -def test_binary_128bit_headers(): - ot.tracer = InstanaTracer() + def test_binary_128bit_headers(self): + ot.tracer = InstanaTracer() - carrier = {'X-INSTANA-T': '0000000000000000b0789916ff8f319f', - 'X-INSTANA-S': ' 0000000000000000b0789916ff8f319f', 'X-INSTANA-L': '1'} - ctx = ot.tracer.extract(ot.Format.BINARY, carrier) + carrier = {'X-INSTANA-T': '0000000000000000b0789916ff8f319f', + 'X-INSTANA-S': ' 0000000000000000b0789916ff8f319f', 'X-INSTANA-L': '1'} + ctx = ot.tracer.extract(ot.Format.BINARY, carrier) - assert isinstance(ctx, SpanContext) - assert ctx.trace_id == 'b0789916ff8f319f' - assert ctx.span_id == 'b0789916ff8f319f' + self.assertIsInstance(ctx, SpanContext) + self.assertEqual(ctx.trace_id, 'b0789916ff8f319f') + self.assertEqual(ctx.span_id, 'b0789916ff8f319f') diff --git a/tests/opentracing/test_ot_span.py b/tests/opentracing/test_ot_span.py index 0196aa0ce..9280df76d 100644 --- a/tests/opentracing/test_ot_span.py +++ b/tests/opentracing/test_ot_span.py @@ -29,14 +29,14 @@ def tearDown(self): def test_span_interface(self): span = opentracing.tracer.start_span("blah") - assert hasattr(span, "finish") - assert hasattr(span, "set_tag") - assert hasattr(span, "tags") - assert hasattr(span, "operation_name") - assert hasattr(span, "set_baggage_item") - assert hasattr(span, "get_baggage_item") - assert hasattr(span, "context") - assert hasattr(span, "log") + self.assertTrue(hasattr(span, "finish")) + self.assertTrue(hasattr(span, "set_tag")) + self.assertTrue(hasattr(span, "tags")) + self.assertTrue(hasattr(span, "operation_name")) + self.assertTrue(hasattr(span, "set_baggage_item")) + self.assertTrue(hasattr(span, "get_baggage_item")) + self.assertTrue(hasattr(span, "context")) + self.assertTrue(hasattr(span, "log")) def test_span_ids(self): count = 0 @@ -44,8 +44,8 @@ def test_span_ids(self): count += 1 span = opentracing.tracer.start_span("test_span_ids") context = span.context - assert 0 <= int(context.span_id, 16) <= 18446744073709551615 - assert 0 <= int(context.trace_id, 16) <= 18446744073709551615 + self.assertTrue(0 <= int(context.span_id, 16) <= 18446744073709551615) + self.assertTrue(0 <= int(context.trace_id, 16) <= 18446744073709551615) # Python 3.11 support is incomplete yet # TODO: Remove this once we find a workaround or DROP opentracing! @@ -53,20 +53,20 @@ def test_span_ids(self): def test_stacks(self): # Entry spans have no stack attached by default wsgi_span = opentracing.tracer.start_span("wsgi") - assert wsgi_span.stack is None + self.assertIsNone(wsgi_span.stack) # SDK spans have no stack attached by default sdk_span = opentracing.tracer.start_span("unregistered_span_type") - assert sdk_span.stack is None + self.assertIsNone(sdk_span.stack) # Exit spans are no longer than 30 frames exit_span = opentracing.tracer.start_span("urllib3") - assert len(exit_span.stack) == 30 + self.assertLessEqual(len(exit_span.stack), 30) def test_span_fields(self): span = opentracing.tracer.start_span("mycustom") self.assertEqual("mycustom", span.operation_name) - assert span.context + self.assertTrue(span.context) span.set_tag("tagone", "string") span.set_tag("tagtwo", 150) @@ -99,23 +99,23 @@ def test_sdk_spans(self): span.finish() spans = recorder.queued_spans() - assert 1, len(spans) + self.assertEqual(1, len(spans)) sdk_span = spans[0] self.assertEqual('sdk', sdk_span.n) self.assertEqual(None, sdk_span.p) self.assertEqual(sdk_span.s, sdk_span.t) - assert sdk_span.ts - assert sdk_span.ts > 0 - assert sdk_span.d - assert sdk_span.d > 0 + self.assertTrue(sdk_span.ts) + self.assertGreater(sdk_span.ts, 0) + self.assertTrue(sdk_span.d) + self.assertGreater(sdk_span.d, 0) - assert sdk_span.data - assert sdk_span.data["sdk"] + self.assertTrue(sdk_span.data) + self.assertTrue(sdk_span.data["sdk"]) self.assertEqual('entry', sdk_span.data["sdk"]["type"]) self.assertEqual('custom_sdk_span', sdk_span.data["sdk"]["name"]) - assert sdk_span.data["sdk"]["custom"] - assert sdk_span.data["sdk"]["custom"]["tags"] + self.assertTrue(sdk_span.data["sdk"]["custom"]) + self.assertTrue(sdk_span.data["sdk"]["custom"]["tags"]) def test_span_kind(self): recorder = opentracing.tracer.recorder @@ -141,7 +141,7 @@ def test_span_kind(self): span.finish() spans = recorder.queued_spans() - assert 5, len(spans) + self.assertEqual(5, len(spans)) span = spans[0] self.assertEqual('entry', span.data["sdk"]["type"]) @@ -185,31 +185,29 @@ def test_tag_values(self): scope.span.set_tag('myset', {"one", 2}) spans = tracer.recorder.queued_spans() - assert len(spans) == 1 + self.assertEqual(1, len(spans)) test_span = spans[0] - assert(test_span) - assert(len(test_span.data['sdk']['custom']['tags']) == 5) - assert(test_span.data['sdk']['custom']['tags']['uuid'] == "UUID('12345678-1234-5678-1234-567812345678')") - assert(test_span.data['sdk']['custom']['tags']['tracer']) - assert(test_span.data['sdk']['custom']['tags']['none'] == 'None') - assert(test_span.data['sdk']['custom']['tags']['mylist'] == [1, 2, 3]) - set_regexp = re.compile(r"\{.*,.*\}") - assert(set_regexp.search(test_span.data['sdk']['custom']['tags']['myset'])) + self.assertTrue(test_span) + self.assertEqual(len(test_span.data['sdk']['custom']['tags']), 5) + self.assertEqual(test_span.data['sdk']['custom']['tags']['uuid'], "UUID('12345678-1234-5678-1234-567812345678')") + self.assertTrue(test_span.data['sdk']['custom']['tags']['tracer']) + self.assertEqual(test_span.data['sdk']['custom']['tags']['none'], 'None') + self.assertListEqual(test_span.data['sdk']['custom']['tags']['mylist'], [1, 2, 3]) + self.assertRegex(test_span.data['sdk']['custom']['tags']['myset'], r"\{.*,.*\}") # Convert to JSON json_data = to_json(test_span) - assert(json_data) + self.assertTrue(json_data) # And back span_dict = json.loads(json_data) - assert(len(span_dict['data']['sdk']['custom']['tags']) == 5) - assert(span_dict['data']['sdk']['custom']['tags']['uuid'] == "UUID('12345678-1234-5678-1234-567812345678')") - assert(span_dict['data']['sdk']['custom']['tags']['tracer']) - assert(span_dict['data']['sdk']['custom']['tags']['none'] == 'None') - assert(span_dict['data']['sdk']['custom']['tags']['mylist'] == [1, 2, 3]) - set_regexp = re.compile(r"{.*,.*}") - assert(set_regexp.search(test_span.data['sdk']['custom']['tags']['myset'])) + self.assertEqual(len(span_dict['data']['sdk']['custom']['tags']), 5) + self.assertEqual(span_dict['data']['sdk']['custom']['tags']['uuid'], "UUID('12345678-1234-5678-1234-567812345678')") + self.assertTrue(span_dict['data']['sdk']['custom']['tags']['tracer']) + self.assertEqual(span_dict['data']['sdk']['custom']['tags']['none'], 'None') + self.assertListEqual(span_dict['data']['sdk']['custom']['tags']['mylist'], [1, 2, 3]) + self.assertRegex(test_span.data['sdk']['custom']['tags']['myset'], r"{.*,.*}") def test_tag_names(self): with tracer.start_active_span('test') as scope: @@ -219,15 +217,15 @@ def test_tag_names(self): scope.span.set_tag(u'asdf', 'This should be ok') spans = tracer.recorder.queued_spans() - assert len(spans) == 1 + self.assertEqual(len(spans), 1) test_span = spans[0] - assert(test_span) - assert(len(test_span.data['sdk']['custom']['tags']) == 1) - assert(test_span.data['sdk']['custom']['tags']['asdf'] == 'This should be ok') + self.assertTrue(test_span) + self.assertEqual(len(test_span.data['sdk']['custom']['tags']), 1) + self.assertEqual(test_span.data['sdk']['custom']['tags']['asdf'], 'This should be ok') json_data = to_json(test_span) - assert(json_data) + self.assertTrue(json_data) def test_custom_service_name(self): # Set a custom service name @@ -245,37 +243,37 @@ def test_custom_service_name(self): exit_scope.span.set_tag(u'type', 'exit_span') spans = tracer.recorder.queued_spans() - assert len(spans) == 3 + self.assertEqual(len(spans), 3) filter = lambda span: span.n == "sdk" and span.data['sdk']['name'] == "entry_span" entry_span = get_first_span_by_filter(spans, filter) - assert (entry_span) + self.assertTrue(entry_span) filter = lambda span: span.n == "sdk" and span.data['sdk']['name'] == "intermediate_span" intermediate_span = get_first_span_by_filter(spans, filter) - assert (intermediate_span) + self.assertTrue(intermediate_span) filter = lambda span: span.n == "sdk" and span.data['sdk']['name'] == "exit_span" exit_span = get_first_span_by_filter(spans, filter) - assert (exit_span) - - assert(entry_span) - assert(len(entry_span.data['sdk']['custom']['tags']) == 2) - assert(entry_span.data['sdk']['custom']['tags']['type'] == 'entry_span') - assert(entry_span.data['service'] == 'custom_service_name') - assert(entry_span.k == 1) - - assert(intermediate_span) - assert(len(intermediate_span.data['sdk']['custom']['tags']) == 1) - assert(intermediate_span.data['sdk']['custom']['tags']['type'] == 'intermediate_span') - assert(intermediate_span.data['service'] == 'custom_service_name') - assert(intermediate_span.k == 3) - - assert(exit_span) - assert(len(exit_span.data['sdk']['custom']['tags']) == 2) - assert(exit_span.data['sdk']['custom']['tags']['type'] == 'exit_span') - assert(exit_span.data['service'] == 'custom_service_name') - assert(exit_span.k == 2) + self.assertTrue(exit_span) + + self.assertTrue(entry_span) + self.assertEqual(len(entry_span.data['sdk']['custom']['tags']), 2) + self.assertEqual(entry_span.data['sdk']['custom']['tags']['type'], 'entry_span') + self.assertEqual(entry_span.data['service'], 'custom_service_name') + self.assertEqual(entry_span.k, 1) + + self.assertTrue(intermediate_span) + self.assertEqual(len(intermediate_span.data['sdk']['custom']['tags']), 1) + self.assertEqual(intermediate_span.data['sdk']['custom']['tags']['type'], 'intermediate_span') + self.assertEqual(intermediate_span.data['service'], 'custom_service_name') + self.assertEqual(intermediate_span.k, 3) + + self.assertTrue(exit_span) + self.assertEqual(len(exit_span.data['sdk']['custom']['tags']), 2) + self.assertEqual(exit_span.data['sdk']['custom']['tags']['type'], 'exit_span') + self.assertEqual(exit_span.data['service'], 'custom_service_name') + self.assertEqual(exit_span.k, 2) def test_span_log(self): with tracer.start_active_span('mylogspan') as scope: @@ -283,14 +281,10 @@ def test_span_log(self): scope.span.log_kv({'Elton John': 'Your Song'}) spans = tracer.recorder.queued_spans() - assert len(spans) == 1 + self.assertEqual(len(spans), 1) my_log_span = spans[0] - assert my_log_span.n == 'sdk' + self.assertEqual(my_log_span.n, 'sdk') log_data = my_log_span.data['sdk']['custom']['logs'] - assert len(log_data) == 2 - - - - + self.assertEqual(len(log_data), 2) diff --git a/tests/test_id_management.py b/tests/test_id_management.py index 61ba5ad3f..fb3badbbc 100644 --- a/tests/test_id_management.py +++ b/tests/test_id_management.py @@ -1,54 +1,56 @@ # (c) Copyright IBM Corp. 2021 # (c) Copyright Instana Inc. 2017 +import unittest import instana -def test_id_generation(): - count = 0 - while count <= 10000: - id = instana.util.ids.generate_id() - base10_id = int(id, 16) - assert base10_id >= 0 - assert base10_id <= 18446744073709551615 - count += 1 +class TestIdManagement(unittest.TestCase): + def test_id_generation(self): + count = 0 + while count <= 10000: + id = instana.util.ids.generate_id() + base10_id = int(id, 16) + self.assertGreaterEqual(base10_id, 0) + self.assertLessEqual(base10_id, 18446744073709551615) + count += 1 -def test_various_header_to_id_conversion(): - # Get a hex string to test against & convert - header_id = instana.util.ids.generate_id() - converted_id = instana.util.ids.header_to_long_id(header_id) - assert(header_id == converted_id) + def test_various_header_to_id_conversion(self): + # Get a hex string to test against & convert + header_id = instana.util.ids.generate_id() + converted_id = instana.util.ids.header_to_long_id(header_id) + self.assertEqual(header_id, converted_id) - # Hex value - result should be left padded - result = instana.util.ids.header_to_long_id('abcdef') - assert('0000000000abcdef' == result) + # Hex value - result should be left padded + result = instana.util.ids.header_to_long_id('abcdef') + self.assertEqual('0000000000abcdef', result) - # Hex value - result = instana.util.ids.header_to_long_id('0123456789abcdef') - assert('0123456789abcdef' == result) + # Hex value + result = instana.util.ids.header_to_long_id('0123456789abcdef') + self.assertEqual('0123456789abcdef', result) - # Very long incoming header should just return the rightmost 16 bytes - result = instana.util.ids.header_to_long_id('0x0123456789abcdef0123456789abcdef') - assert('0x0123456789abcdef0123456789abcdef' == result) + # Very long incoming header should just return the rightmost 16 bytes + result = instana.util.ids.header_to_long_id('0x0123456789abcdef0123456789abcdef') + self.assertEqual('0x0123456789abcdef0123456789abcdef', result) -def test_header_to_id_conversion_with_bogus_header(): - # Bogus nil arg - bogus_result = instana.util.ids.header_to_long_id(None) - assert(instana.util.ids.BAD_ID == bogus_result) + def test_header_to_id_conversion_with_bogus_header(self): + # Bogus nil arg + bogus_result = instana.util.ids.header_to_long_id(None) + self.assertEqual(instana.util.ids.BAD_ID, bogus_result) - # Bogus Integer arg - bogus_result = instana.util.ids.header_to_long_id(1234) - assert(instana.util.ids.BAD_ID == bogus_result) + # Bogus Integer arg + bogus_result = instana.util.ids.header_to_long_id(1234) + self.assertEqual(instana.util.ids.BAD_ID, bogus_result) - # Bogus Array arg - bogus_result = instana.util.ids.header_to_long_id([1234]) - assert(instana.util.ids.BAD_ID == bogus_result) + # Bogus Array arg + bogus_result = instana.util.ids.header_to_long_id([1234]) + self.assertEqual(instana.util.ids.BAD_ID, bogus_result) - # Bogus Hex Values in String - bogus_result = instana.util.ids.header_to_long_id('0xZZZZZZ') - assert(instana.util.ids.BAD_ID == bogus_result) + # Bogus Hex Values in String + bogus_result = instana.util.ids.header_to_long_id('0xZZZZZZ') + self.assertEqual(instana.util.ids.BAD_ID, bogus_result) - bogus_result = instana.util.ids.header_to_long_id('ZZZZZZ') - assert(instana.util.ids.BAD_ID == bogus_result) + bogus_result = instana.util.ids.header_to_long_id('ZZZZZZ') + self.assertEqual(instana.util.ids.BAD_ID, bogus_result) diff --git a/tests/test_util.py b/tests/test_util.py new file mode 100644 index 000000000..3aa2db6a6 --- /dev/null +++ b/tests/test_util.py @@ -0,0 +1,21 @@ +# (c) Copyright IBM Corp. 2024 + +import unittest +from instana.util import validate_url + + +class TestUtil(unittest.TestCase): + def test_validate_url(self): + self.assertTrue(validate_url("http://localhost:3000")) + self.assertTrue(validate_url("http://localhost:3000/")) + self.assertTrue(validate_url("https://localhost:3000/path/item")) + self.assertTrue(validate_url("http://localhost")) + self.assertTrue(validate_url("https://localhost/")) + self.assertTrue(validate_url("https://localhost/path/item")) + self.assertTrue(validate_url("http://127.0.0.1")) + self.assertTrue(validate_url("https://10.0.12.221/")) + self.assertTrue(validate_url("http://[2001:db8:85a3:8d3:1319:8a2e:370:7348]/")) + self.assertTrue(validate_url("https://[2001:db8:85a3:8d3:1319:8a2e:370:7348]:443/")) + self.assertFalse(validate_url("boligrafo")) + self.assertFalse(validate_url("http:boligrafo")) + self.assertFalse(validate_url(None)) diff --git a/tests/test_utils.py b/tests/test_utils.py index 439436952..57fc5cf64 100644 --- a/tests/test_utils.py +++ b/tests/test_utils.py @@ -1,29 +1,6 @@ # (c) Copyright IBM Corp. 2021 # (c) Copyright Instana Inc. 2020 -from instana.util import validate_url - - -def setup_method(): - pass - - -def test_validate_url(): - assert(validate_url("http://localhost:3000")) - assert(validate_url("http://localhost:3000/")) - assert(validate_url("https://localhost:3000/path/item")) - assert(validate_url("http://localhost")) - assert(validate_url("https://localhost/")) - assert(validate_url("https://localhost/path/item")) - assert(validate_url("http://127.0.0.1")) - assert(validate_url("https://10.0.12.221/")) - assert(validate_url("http://[2001:db8:85a3:8d3:1319:8a2e:370:7348]/")) - assert(validate_url("https://[2001:db8:85a3:8d3:1319:8a2e:370:7348]:443/")) - assert(validate_url("boligrafo") is False) - assert(validate_url("http:boligrafo") is False) - assert(validate_url(None) is False) - - class _TraceContextMixin: def assertTraceContextPropagated(self, parent_span, child_span): self.assertEqual(parent_span.t, child_span.t)