Testing lint checker3

Author: Johnnyl202

aws-opentelemetry-distro/src/amazon/opentelemetry/distro/mcpinstrumentor/mcpinstrumentor.py

@@ -29,12 +29,12 @@ class MCPInstrumentor(BaseInstrumentor):
     An instrumenter for MCP.
     """
 
-    def __init__(self):
+    def __init__(self): # pylint: disable=no-self-use
         super().__init__()
         self.tracer = None
 
     # Send Request Wrapper
-    def _wrap_send_request(self, wrapped, instance, args, kwargs):
+    def _wrap_send_request(self, wrapped, instance, args, kwargs):  # pylint: disable=no-self-use
         """
         Changes made:
             The wrapper intercepts the request before sending, injects distributed tracing context into the
@@ -43,7 +43,7 @@ def _wrap_send_request(self, wrapped, instance, args, kwargs):
             type and calling the original function with identical parameters.
         """
 
-        async def async_wrapper():
+        async def async_wrapper():  # pylint: disable=no-self-use
             with self.tracer.start_as_current_span("client.send_request", kind=trace.SpanKind.CLIENT) as span:
                 span_ctx = span.get_span_context()
                 request = args[0] if len(args) > 0 else kwargs.get("request")
@@ -68,7 +68,7 @@ async def async_wrapper():
         return async_wrapper()
 
     # Handle Request Wrapper
-    async def _wrap_handle_request(self, wrapped, instance, args, kwargs):
+    async def _wrap_handle_request(self, wrapped, instance, args, kwargs):  # pylint: disable=no-self-use
         """
         Changes made:
         This wrapper intercepts requests before processing, extracts distributed tracing context from
@@ -140,7 +140,7 @@ def _get_span_name(self, req):  # pylint: disable=no-self-use
                 span_name = "unknown"
         return span_name
 
-    def handle_attributes(self, span, request, is_client=True):
+    def handle_attributes(self, span, request, is_client=True): # pylint: disable=no-self-use
         import mcp.types as types  # pylint: disable=import-outside-toplevel,consider-using-from-import
 
         operation = self._get_span_name(request)
@@ -167,10 +167,10 @@ def _add_server_attributes(self, span, operation, request):  # pylint: disable=n
         if hasattr(request, "params") and hasattr(request.params, "name"):
             span.set_attribute("tool.name", request.params.name)
 
-    def instrumentation_dependencies(self) -> Collection[str]:
+    def instrumentation_dependencies(self) -> Collection[str]:  # pylint: disable=no-self-use
         return _instruments
 
-    def _instrument(self, **kwargs: Any) -> None:
+    def _instrument(self, **kwargs: Any) -> None:   # pylint: disable=no-self-use
         tracer_provider = kwargs.get("tracer_provider")
         if tracer_provider:
             self.tracer = tracer_provider.get_tracer("mcp")
@@ -193,6 +193,6 @@ def _instrument(self, **kwargs: Any) -> None:
             "mcp.server.lowlevel.server",
         )
 
-    def _uninstrument(self, **kwargs: Any) -> None:
+    def _uninstrument(self, **kwargs: Any) -> None: # pylint: disable=no-self-use
         unwrap("mcp.shared.session", "BaseSession.send_request")
         unwrap("mcp.server.lowlevel.server", "Server._handle_request")

aws-opentelemetry-distro/tests/amazon/opentelemetry/distro/test_mcpinstrumentor.py

@@ -23,19 +23,19 @@
 class SimpleSpanContext:
     """Simple mock span context without using MagicMock"""
 
-    def __init__(self, trace_id, span_id):
+    def __init__(self, trace_id, span_id):  # pylint: disable=no-self-use
         self.trace_id = trace_id
         self.span_id = span_id
 
 
 class SimpleTracerProvider:
     """Simple mock tracer provider without using MagicMock"""
 
-    def __init__(self):
+    def __init__(self): # pylint: disable=no-self-use
         self.get_tracer_called = False
         self.tracer_name = None
 
-    def get_tracer(self, name):
+    def get_tracer(self, name): # pylint: disable=no-self-use
         self.get_tracer_called = True
         self.tracer_name = name
         return "mock_tracer_from_provider"
@@ -44,10 +44,10 @@ def get_tracer(self, name):
 class TestInjectTraceContext(unittest.TestCase):
     """Test the _inject_trace_context method"""
 
-    def setUp(self):
+    def setUp(self):    # pylint: disable=no-self-use
         self.instrumentor = MCPInstrumentor()
 
-    def test_inject_trace_context_empty_dict(self):
+    def test_inject_trace_context_empty_dict(self): # pylint: disable=no-self-use
         """Test injecting trace context into empty dictionary"""
         # Setup
         request_data = {}
@@ -70,7 +70,7 @@ def test_inject_trace_context_empty_dict(self):
         self.assertEqual(int(parts[1], 16), 12345)  # trace_id
         self.assertEqual(int(parts[2], 16), 67890)  # span_id
 
-    def test_inject_trace_context_existing_params(self):
+    def test_inject_trace_context_existing_params(self):    # pylint: disable=no-self-use
         """Test injecting trace context when params already exist"""
         # Setup
         request_data = {"params": {"existing_field": "test_value"}}
@@ -94,13 +94,13 @@ def test_inject_trace_context_existing_params(self):
 class TestTracerProvider(unittest.TestCase):
     """Test the tracer provider kwargs logic in _instrument method"""
 
-    def setUp(self):
+    def setUp(self):    # pylint: disable=no-self-use
         self.instrumentor = MCPInstrumentor()
         # Reset tracer to ensure test isolation
         if hasattr(self.instrumentor, "tracer"):
             delattr(self.instrumentor, "tracer")
 
-    def test_instrument_without_tracer_provider_kwargs(self):
+    def test_instrument_without_tracer_provider_kwargs(self):   # pylint: disable=no-self-use
         """Test _instrument method when no tracer_provider in kwargs - should use default tracer"""
         # Execute - Actually test the mcpinstrumentor method
         with unittest.mock.patch("opentelemetry.trace.get_tracer") as mock_get_tracer:
@@ -112,7 +112,7 @@ def test_instrument_without_tracer_provider_kwargs(self):
         self.assertEqual(self.instrumentor.tracer, "default_tracer")
         mock_get_tracer.assert_called_with("mcp")
 
-    def test_instrument_with_tracer_provider_kwargs(self):
+    def test_instrument_with_tracer_provider_kwargs(self):  # pylint: disable=no-self-use
         """Test _instrument method when tracer_provider is in kwargs - should use provider's tracer"""
         # Setup
         provider = SimpleTracerProvider()
@@ -130,10 +130,10 @@ def test_instrument_with_tracer_provider_kwargs(self):
 class TestInstrumentationDependencies(unittest.TestCase):
     """Test the instrumentation_dependencies method"""
 
-    def setUp(self):
+    def setUp(self):    # pylint: disable=no-self-use
         self.instrumentor = MCPInstrumentor()
 
-    def test_instrumentation_dependencies(self):
+    def test_instrumentation_dependencies(self):    # pylint: disable=no-self-use
         """Test that instrumentation_dependencies method returns the expected dependencies"""
         # Execute - Actually test the mcpinstrumentor method
         dependencies = self.instrumentor.instrumentation_dependencies()
@@ -147,19 +147,19 @@ def test_instrumentation_dependencies(self):
 class TestTraceContextInjection(unittest.TestCase):
     """Test trace context injection using actual mcpinstrumentor methods"""
 
-    def setUp(self):
+    def setUp(self):    # pylint: disable=no-self-use
         self.instrumentor = MCPInstrumentor()
 
-    def test_trace_context_injection_with_realistic_request(self):
+    def test_trace_context_injection_with_realistic_request(self):  # pylint: disable=no-self-use
         """Test actual trace context injection using mcpinstrumentor._inject_trace_context with realistic MCP request"""
 
         # Create a realistic MCP request structure
         class CallToolRequest:
-            def __init__(self, tool_name, arguments=None):
+            def __init__(self, tool_name, arguments=None):  # pylint: disable=no-self-use
                 self.root = self
                 self.params = CallToolParams(tool_name, arguments)
 
-            def model_dump(self, by_alias=True, mode="json", exclude_none=True):
+            def model_dump(self, by_alias=True, mode="json", exclude_none=True):    # pylint: disable=no-self-use
                 result = {"method": "call_tool", "params": {"name": self.params.name}}
                 if self.params.arguments:
                     result["params"]["arguments"] = self.params.arguments
@@ -169,15 +169,15 @@ def model_dump(self, by_alias=True, mode="json", exclude_none=True):
                 return result
 
             @classmethod
-            def model_validate(cls, data):
+            def model_validate(cls, data):  # pylint: disable=no-self-use
                 instance = cls(data["params"]["name"], data["params"].get("arguments"))
                 # Restore _meta field if present
                 if "_meta" in data["params"]:
                     instance.params._meta = data["params"]["_meta"]
                 return instance
 
         class CallToolParams:
-            def __init__(self, name, arguments=None):
+            def __init__(self, name, arguments=None):   # pylint: disable=no-self-use
                 self.name = name
                 self.arguments = arguments
                 self._meta = None  # Will hold trace context
@@ -214,7 +214,7 @@ def __init__(self, name, arguments=None):
 class TestInstrumentedMCPServer(unittest.TestCase):
     """Test mcpinstrumentor with a mock MCP server to verify end-to-end functionality"""
 
-    def setUp(self):
+    def setUp(self):    # pylint: disable=no-self-use
         self.instrumentor = MCPInstrumentor()
         # Initialize tracer so the instrumentor can work
         mock_tracer = MagicMock()
@@ -224,15 +224,15 @@ def test_no_trace_context_fallback(self):  # pylint: disable=no-self-use
         """Test graceful handling when no trace context is present on server side"""
 
         class MockServerNoTrace:
-            async def _handle_request(self, session, request):
+            async def _handle_request(self, session, request): # pylint: disable=no-self-use
                 return {"success": True, "handled_without_trace": True}
 
         class MockServerRequestNoTrace:
-            def __init__(self, tool_name):
+            def __init__(self, tool_name): # pylint: disable=no-self-use
                 self.params = MockServerRequestParamsNoTrace(tool_name)
 
         class MockServerRequestParamsNoTrace:
-            def __init__(self, name):
+            def __init__(self, name): # pylint: disable=no-self-use
                 self.name = name
                 self.meta = None  # No trace context
 
@@ -269,12 +269,12 @@ def test_end_to_end_client_server_communication(self):  # pylint: disable=too-ma
 
         # Create realistic request/response classes
         class MCPRequest:
-            def __init__(self, tool_name, arguments=None, method="call_tool"):
+            def __init__(self, tool_name, arguments=None, method="call_tool"):  # pylint: disable=no-self-use
                 self.root = self
                 self.params = MCPRequestParams(tool_name, arguments)
                 self.method = method
 
-            def model_dump(self, by_alias=True, mode="json", exclude_none=True):
+            def model_dump(self, by_alias=True, mode="json", exclude_none=True):    # pylint: disable=no-self-use
                 result = {"method": self.method, "params": {"name": self.params.name}}
                 if self.params.arguments:
                     result["params"]["arguments"] = self.params.arguments
@@ -284,7 +284,7 @@ def model_dump(self, by_alias=True, mode="json", exclude_none=True):
                 return result
 
             @classmethod
-            def model_validate(cls, data):
+            def model_validate(cls, data):  # pylint: disable=no-self-use
                 method = data.get("method", "call_tool")
                 instance = cls(data["params"]["name"], data["params"].get("arguments"), method)
                 # Restore _meta field if present
@@ -293,19 +293,19 @@ def model_validate(cls, data):
                 return instance
 
         class MCPRequestParams:
-            def __init__(self, name, arguments=None):
+            def __init__(self, name, arguments=None):   # pylint: disable=no-self-use
                 self.name = name
                 self.arguments = arguments
                 self._meta = None
 
         class MCPServerRequest:
-            def __init__(self, client_request_data):
+            def __init__(self, client_request_data):    # pylint: disable=no-self-use
                 """Server request created from client's serialized data"""
                 self.method = client_request_data.get("method", "call_tool")
                 self.params = MCPServerRequestParams(client_request_data["params"])
 
         class MCPServerRequestParams:
-            def __init__(self, params_data):
+            def __init__(self, params_data):    # pylint: disable=no-self-use
                 self.name = params_data["name"]
                 self.arguments = params_data.get("arguments")
                 # Extract traceparent from _meta if present
@@ -315,16 +315,16 @@ def __init__(self, params_data):
                     self.meta = None
 
         class MCPServerRequestMeta:
-            def __init__(self, traceparent):
+            def __init__(self, traceparent):    # pylint: disable=no-self-use
                 self.traceparent = traceparent
 
         # Mock client and server that actually communicate
         class EndToEndMCPSystem:
-            def __init__(self):
+            def __init__(self): # pylint: disable=no-self-use
                 self.communication_log = []
                 self.last_sent_request = None
 
-            async def client_send_request(self, request):
+            async def client_send_request(self, request):   # pylint: disable=no-self-use
                 """Client sends request - captures what gets sent"""
                 self.communication_log.append("CLIENT: Preparing to send request")
                 self.last_sent_request = request  # Capture the modified request
@@ -336,7 +336,7 @@ async def client_send_request(self, request):
                 # Return client response
                 return {"success": True, "client_response": "Request sent successfully"}
 
-            async def server_handle_request(self, session, server_request):
+            async def server_handle_request(self, session, server_request): # pylint: disable=no-self-use
                 """Server handles the request it received"""
                 self.communication_log.append(f"SERVER: Received request for {server_request.params.name}")