opensource/opentelemetry-python-contrib
1
0
Fork 0
mirror of https://github.com/open-telemetry/opentelemetry-python-contrib.git synced 2025-08-02 19:47:17 +08:00
Code Issues Packages Projects Releases Wiki Activity

Consistent way of not instrumenting multiple times (#549)

Browse Source
This commit is contained in:
Leighton Chen
2021-07-09 09:55:44 -07:00
committed by GitHub
parent bf97e172f0
commit 56da6d74df
17 changed files with 238 additions and 84 deletions
Download Patch File Download Diff File Expand all files Collapse all files

42
instrumentation/opentelemetry-instrumentation-fastapi/tests/test_fastapi_instrumentation.py
View File

@ -19,6 +19,7 @@ import fastapi
from fastapi.testclient import TestClient
import opentelemetry.instrumentation.fastapi as otel_fastapi
from opentelemetry.instrumentation.asgi import OpenTelemetryMiddleware
from opentelemetry.sdk.resources import Resource
from opentelemetry.semconv.trace import SpanAttributes
from opentelemetry.test.test_base import TestBase
@ -57,6 +58,47 @@ class TestFastAPIManualInstrumentation(TestBase):
super().tearDown()
self.env_patch.stop()
self.exclude_patch.stop()
with self.disable_logging():
self._instrumentor.uninstrument()
self._instrumentor.uninstrument_app(self._app)
def test_instrument_app_with_instrument(self):
if not isinstance(self, TestAutoInstrumentation):
self._instrumentor.instrument()
self._client.get("/foobar")
spans = self.memory_exporter.get_finished_spans()
self.assertEqual(len(spans), 3)
for span in spans:
self.assertIn("/foobar", span.name)
def test_uninstrument_app(self):
self._client.get("/foobar")
spans = self.memory_exporter.get_finished_spans()
self.assertEqual(len(spans), 3)
# pylint: disable=import-outside-toplevel
from fastapi.middleware.httpsredirect import HTTPSRedirectMiddleware
self._app.add_middleware(HTTPSRedirectMiddleware)
self._instrumentor.uninstrument_app(self._app)
print(self._app.user_middleware[0].cls)
self.assertFalse(
isinstance(
self._app.user_middleware[0].cls, OpenTelemetryMiddleware
)
)
self._client = TestClient(self._app)
resp = self._client.get("/foobar")
self.assertEqual(200, resp.status_code)
span_list = self.memory_exporter.get_finished_spans()
self.assertEqual(len(span_list), 3)
def test_uninstrument_app_after_instrument(self):
if not isinstance(self, TestAutoInstrumentation):
self._instrumentor.instrument()
self._instrumentor.uninstrument_app(self._app)
self._client.get("/foobar")
spans = self.memory_exporter.get_finished_spans()
self.assertEqual(len(spans), 0)
def test_basic_fastapi_call(self):
self._client.get("/foobar")