opentelemetry-instrumentation-mysql: add semconv stability opt-in test coverage (#4810)

* test: cover mysql instrumentation semconv opt-in modes

* test(mysql): use semconv constants instead of hardcoded attribute keys

* test(mysql): add changelog fragment for semconv opt-in mode tests

---------

Co-authored-by: Leighton Chen <lechen@microsoft.com>

Author: anneheartrecord

.changelog/4810.added

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

instrumentation/opentelemetry-instrumentation-mysql/tests/test_mysql_integration.py

@@ -1,20 +1,56 @@
 # Copyright The OpenTelemetry Authors
 # SPDX-License-Identifier: Apache-2.0
 
+import contextlib
 from unittest import mock
 
 import mysql.connector
 
 import opentelemetry.instrumentation.mysql
 from opentelemetry import trace as trace_api
+from opentelemetry.instrumentation._semconv import (
+    OTEL_SEMCONV_STABILITY_OPT_IN,
+    _OpenTelemetrySemanticConventionStability,
+)
 from opentelemetry.instrumentation.mysql import MySQLInstrumentor
 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 = mysql.connector.connect(database="test")
     cursor = cnx.cursor()
@@ -29,6 +65,7 @@ def make_mysql_connection_mock():
     cnx.database = "test"
     cnx.server_host = "localhost"
     cnx.server_port = 3306
+    cnx.user = "testuser"
 
     cursor = mock.MagicMock()
     cursor._cnx = cnx
@@ -441,3 +478,59 @@ def test_uninstrument_connection(self, mock_connect):
 
         spans_list = self.memory_exporter.get_finished_spans()
         self.assertEqual(len(spans_list), 1)
+
+    @mock.patch("mysql.connector.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_mysql_connection_mock()
+            MySQLInstrumentor().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("mysql.connector.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_mysql_connection_mock()
+            MySQLInstrumentor().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)