opensource/opentelemetry-python-contrib
1
0
Fork 0
mirror of https://github.com/open-telemetry/opentelemetry-python-contrib.git synced 2025-07-29 21:23:55 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat(instrumentation-dbapi): add experimental sql commenter capability (#908)

Browse Source 
* feat(instrumentation-dbapi): add experimental sql commenter capability

* Update instrumentation/opentelemetry-instrumentation-dbapi/src/opentelemetry/instrumentation/dbapi/__init__.py

Co-authored-by: Diego Hurtado <ocelotl@users.noreply.github.com>

* Fix lint

* Add CHANGELOG entry

* Fix lint

* Fix lint again

Co-authored-by: Diego Hurtado <ocelotl@users.noreply.github.com>
This commit is contained in:
Srikanth Chekuri
2022-02-11 21:44:26 +05:30
committed by GitHub
parent 454f4b1511
commit d54b61e687
4 changed files with 109 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files

21
instrumentation/opentelemetry-instrumentation-dbapi/tests/test_dbapi_integration.py
View File

@ -228,6 +228,21 @@ class TestDBApiIntegration(TestBase):
span.attributes[SpanAttributes.DB_STATEMENT], "Test query"
)
def test_executemany_comment(self):
db_integration = dbapi.DatabaseApiIntegration(
"testname", "testcomponent", enable_commenter=True
)
mock_connection = db_integration.wrapped_connection(
mock_connect, {}, {}
)
cursor = mock_connection.cursor()
cursor.executemany("Test query")
spans_list = self.memory_exporter.get_finished_spans()
self.assertEqual(len(spans_list), 1)
span = spans_list[0]
comment = dbapi.CursorTracer._generate_comment(span)
self.assertIn(comment, cursor.query)
def test_callproc(self):
db_integration = dbapi.DatabaseApiIntegration(
"testname", "testcomponent"
@ -308,6 +323,10 @@ class MockConnection:
class MockCursor:
def __init__(self) -> None:
self.query = ""
self.params = None
# pylint: disable=unused-argument, no-self-use
def execute(self, query, params=None, throw_exception=False):
if throw_exception:
@ -317,6 +336,8 @@ class MockCursor:
def executemany(self, query, params=None, throw_exception=False):
if throw_exception:
raise Exception("Test Exception")
self.query = query
self.params = params
# pylint: disable=unused-argument, no-self-use
def callproc(self, query, params=None, throw_exception=False):