fix: resolve lint and formatting issues in _semconv.py and tests

Author: RiyaChaturvedi37

opentelemetry-instrumentation/src/opentelemetry/instrumentation/_semconv.py

@@ -610,6 +610,26 @@ def _set_db_operation(
 
 
 # General
+def _set_span_status(span: Span, status: StatusCode) -> None:
+    status_priority = {
+        StatusCode.UNSET: 0,
+        StatusCode.OK: 1,
+        StatusCode.ERROR: 2,
+    }
+    current = getattr(span, "status", None)
+    if current is not None:
+        if status_priority.get(status, 0) < status_priority.get(
+            current.status_code, 0
+        ):
+            return
+        description = (
+            current.description
+            if current.description and status == current.status_code
+            else None
+        )
+        span.set_status(Status(status, description))
+    else:
+        span.set_status(Status(status))
 
 
 def _set_status(
@@ -650,28 +670,7 @@ def _set_status(
                     span.set_attribute(ERROR_TYPE, status_code_str)
                 metrics_attributes[ERROR_TYPE] = status_code_str
         if span.is_recording():
-            current_status = getattr(span, "status", None)
-            if current_status is not None:
-                _STATUS_PRIORITY = {
-                    StatusCode.UNSET: 0,
-                    StatusCode.OK: 1,
-                    StatusCode.ERROR: 2,
-                }
-                current_priority = _STATUS_PRIORITY.get(
-                    current_status.status_code, 0
-                )
-                incoming_priority = _STATUS_PRIORITY.get(status, 0)
-                if incoming_priority < current_priority:
-                    return
-                description = (
-                    current_status.description
-                    if current_status.description
-                    and status == current_status.status_code
-                    else None
-                )
-                span.set_status(Status(status, description))
-            else:
-                span.set_status(Status(status))
+            _set_span_status(span, status)
 
 
 def _get_schema_url(mode: _StabilityMode) -> str:

opentelemetry-instrumentation/tests/test_semconv_status.py

@@ -8,15 +8,15 @@
 
 import unittest
 from unittest.mock import MagicMock
-from opentelemetry.trace.status import Status, StatusCode
+
 from opentelemetry.instrumentation._semconv import (
     _set_status,
     _StabilityMode,
 )
+from opentelemetry.trace.status import Status, StatusCode
 
 
 class TestSetStatus(unittest.TestCase):
-
     def _make_span(self, status_code, description=None):
         span = MagicMock()
         span.is_recording.return_value = True
@@ -26,9 +26,14 @@ def _make_span(self, status_code, description=None):
     def test_does_not_downgrade_error_to_ok(self):
         """ERROR status should not be overridden by a lower priority OK"""
         span = self._make_span(StatusCode.ERROR, "original error")
-        _set_status(span, {}, 200, "200", server_span=True,
-                    sem_conv_opt_in_mode=_StabilityMode.DEFAULT)
-        # set_status should NOT have been called (no downgrade)
+        _set_status(
+            span,
+            {},
+            200,
+            "200",
+            server_span=True,
+            sem_conv_opt_in_mode=_StabilityMode.DEFAULT,
+        )
         for call in span.set_status.call_args_list:
             args = call[0]
             if args:
@@ -37,8 +42,14 @@ def test_does_not_downgrade_error_to_ok(self):
     def test_does_not_wipe_description_with_none(self):
         """Same ERROR status should preserve existing description"""
         span = self._make_span(StatusCode.ERROR, "keep this message")
-        _set_status(span, {}, 500, "500", server_span=True,
-                    sem_conv_opt_in_mode=_StabilityMode.DEFAULT)
+        _set_status(
+            span,
+            {},
+            500,
+            "500",
+            server_span=True,
+            sem_conv_opt_in_mode=_StabilityMode.DEFAULT,
+        )
         last_call = span.set_status.call_args
         if last_call:
             status_arg = last_call[0][0]
@@ -47,19 +58,27 @@ def test_does_not_wipe_description_with_none(self):
     def test_upgrades_unset_to_error(self):
         """UNSET status should be upgraded to ERROR"""
         span = self._make_span(StatusCode.UNSET)
-        _set_status(span, {}, 500, "500", server_span=True,
-                    sem_conv_opt_in_mode=_StabilityMode.DEFAULT)
+        _set_status(
+            span,
+            {},
+            500,
+            "500",
+            server_span=True,
+            sem_conv_opt_in_mode=_StabilityMode.DEFAULT,
+        )
         span.set_status.assert_called()
         last_call = span.set_status.call_args[0][0]
         self.assertEqual(last_call.status_code, StatusCode.ERROR)
 
     def test_unset_to_ok(self):
         """UNSET status should be upgraded to OK for 2xx"""
         span = self._make_span(StatusCode.UNSET)
-        _set_status(span, {}, 200, "200", server_span=False,
-                    sem_conv_opt_in_mode=_StabilityMode.DEFAULT)
+        _set_status(
+            span,
+            {},
+            200,
+            "200",
+            server_span=False,
+            sem_conv_opt_in_mode=_StabilityMode.DEFAULT,
+        )
         span.set_status.assert_called()
-
-
-if __name__ == "__main__":
-    unittest.main()