Pymysql semconv optin tests

.changelog/4835.added

@@ -0,0 +1 @@
+`opentelemetry-instrumentation-pymysql`: add semantic convention stability opt-in mode tests

instrumentation/opentelemetry-instrumentation-pymysql/tests/test_pymysql_integration.py

@@ -1,20 +1,73 @@
 # Copyright The OpenTelemetry Authors
 # SPDX-License-Identifier: Apache-2.0
 
+import contextlib
 from unittest import mock
 
 import pymysql
 
 import opentelemetry.instrumentation.pymysql
 from opentelemetry import trace as trace_api
+from opentelemetry.instrumentation._semconv import (
+    OTEL_SEMCONV_STABILITY_OPT_IN,
+    _OpenTelemetrySemanticConventionStability,
+)
 from opentelemetry.instrumentation.pymysql import PyMySQLInstrumentor
 from opentelemetry.sdk import resources
 from opentelemetry.semconv._incubating.attributes.db_attributes import (
+    DB_NAME,
     DB_STATEMENT,
+    DB_SYSTEM,
+    DB_USER,
+)
+from opentelemetry.semconv._incubating.attributes.net_attributes import (
+    NET_PEER_NAME,
+    NET_PEER_PORT,
+)
+from opentelemetry.semconv.attributes.db_attributes import (
+    DB_NAMESPACE,
+    DB_QUERY_TEXT,
+    DB_SYSTEM_NAME,
+)
+from opentelemetry.semconv.attributes.server_attributes import (
+    SERVER_ADDRESS,
+    SERVER_PORT,
 )
 from opentelemetry.test.test_base import TestBase
 
 
+@contextlib.contextmanager
+def use_semconv_opt_in(sem_conv_mode):
+    env_patch = mock.patch.dict(
+        "os.environ",
+        {OTEL_SEMCONV_STABILITY_OPT_IN: sem_conv_mode},
+    )
+    _OpenTelemetrySemanticConventionStability._initialized = False
+    env_patch.start()
+    try:
+        yield
+    finally:
+        env_patch.stop()
+        _OpenTelemetrySemanticConventionStability._initialized = False
+
+
+def connect_and_execute_query():
+    cnx = pymysql.connect(database="test")
+    cursor = cnx.cursor()
+    query = "SELECT * FROM test"
+    cursor.execute(query)
+    return cnx, query
+
+
+def make_pymysql_connection_mock():
+    cnx = mock.MagicMock()
+    cnx.db = "test"
+    cnx.host = "localhost"
+    cnx.port = 3306
+    cnx.user = "testuser"
+    return cnx
+
+
 class TestPyMysqlIntegration(TestBase):
     # pylint: disable=invalid-name
     def tearDown(self):
@@ -477,3 +530,59 @@ def test_uninstrument_connection(self, mock_connect):
 
         spans_list = self.memory_exporter.get_finished_spans()
         self.assertEqual(len(spans_list), 1)
+
+    @mock.patch("pymysql.connect")
+    def test_semconv_stable(self, mock_connect):
+        """database,http opt-in emits only stable attributes."""
+        with use_semconv_opt_in("database,http"):
+            mock_connect.return_value = make_pymysql_connection_mock()
+            PyMySQLInstrumentor().instrument()
+
+            connect_and_execute_query()
+
+            spans_list = self.memory_exporter.get_finished_spans()
+            self.assertEqual(len(spans_list), 1)
+            span = spans_list[0]
+
+            self.assertEqual(span.attributes[DB_SYSTEM_NAME], "mysql")
+            self.assertEqual(span.attributes[DB_NAMESPACE], "test")
+            self.assertEqual(
+                span.attributes[DB_QUERY_TEXT], "SELECT * FROM test"
+            )
+            self.assertEqual(span.attributes[SERVER_ADDRESS], "localhost")
+            self.assertEqual(span.attributes[SERVER_PORT], 3306)
+            self.assertNotIn(DB_SYSTEM, span.attributes)
+            self.assertNotIn(DB_NAME, span.attributes)
+            self.assertNotIn(DB_STATEMENT, span.attributes)
+            self.assertNotIn(DB_USER, span.attributes)
+            self.assertNotIn(NET_PEER_NAME, span.attributes)
+            self.assertNotIn(NET_PEER_PORT, span.attributes)
+
+    @mock.patch("pymysql.connect")
+    def test_semconv_dup(self, mock_connect):
+        """database/dup,http/dup opt-in emits both legacy and stable attributes."""
+        with use_semconv_opt_in("database/dup,http/dup"):
+            mock_connect.return_value = make_pymysql_connection_mock()
+            PyMySQLInstrumentor().instrument()
+
+            connect_and_execute_query()
+
+            spans_list = self.memory_exporter.get_finished_spans()
+            self.assertEqual(len(spans_list), 1)
+            span = spans_list[0]
+
+            self.assertEqual(span.attributes[DB_SYSTEM], "mysql")
+            self.assertEqual(span.attributes[DB_SYSTEM_NAME], "mysql")
+            self.assertEqual(span.attributes[DB_NAME], "test")
+            self.assertEqual(span.attributes[DB_NAMESPACE], "test")
+            self.assertEqual(
+                span.attributes[DB_STATEMENT], "SELECT * FROM test"
+            )
+            self.assertEqual(
+                span.attributes[DB_QUERY_TEXT], "SELECT * FROM test"
+            )
+            self.assertEqual(span.attributes[DB_USER], "testuser")
+            self.assertEqual(span.attributes[NET_PEER_NAME], "localhost")
+            self.assertEqual(span.attributes[NET_PEER_PORT], 3306)
+            self.assertEqual(span.attributes[SERVER_ADDRESS], "localhost")
+            self.assertEqual(span.attributes[SERVER_PORT], 3306)