test(pytest): add baseline pytest coverage and safe lint fixes

Author: francescobianco

pyproject.toml

@@ -24,5 +24,8 @@ dependencies = [
 [tool.setuptools]
 license-files = ["LICENSE"]
 
+[tool.pytest.ini_options]
+testpaths = ["tests/pytest"]
+
 [project.scripts]
 openapi-mcp-sdk = "openapi_mcp_sdk.cli:main"

requirements.txt

@@ -18,3 +18,4 @@ fastmcp
 requests
 pydantic
 uvicorn
+pytest

src/openapi_mcp_sdk/main.py

@@ -1,3 +1,5 @@
+"""ASGI entrypoint for the OpenAPI MCP server."""
+
 import os
 import re
 import sys
@@ -10,6 +12,23 @@
 from starlette.types import ASGIApp, Scope, Receive, Send, Message
 from .mcp_audit import McpAuditMiddleware
 from .storage_backend import read_file
+from .memory_store import get_callback_result, set_callback_result
+from .mcp_core import mcp
+from .apis import (
+    async_tool,
+    automotive,
+    cap,
+    company,
+    docuengine,
+    exchange,
+    geocoding,
+    info,
+    pec,
+    risk,
+    sms,
+    trust,
+    visurecamerali,
+)
 
 # ---------------------------------------------------------------------------
 # Bootstrap package logger early — before uvicorn configures its own logging.
@@ -59,12 +78,8 @@ def format(self, record: logging.LogRecord) -> str:
         # Fallback: at least mask tokens
         return self._TOKEN_RE.sub(r'\1token=***', result)
 
-    # Delegate everything else to the inner formatter
     def __getattr__(self, name):
         return getattr(self._inner, name)
-from .memory_store import get_callback_result, set_callback_result
-from .mcp_core import mcp # Import MCP instance with tools already registered in mcp_core.py
-from .apis import async_tool, company, cap, trust, visurecamerali, sms, risk, geocoding,automotive,exchange, pec, docuengine, info # Import tool modules (side-effect: triggers @mcp.tool registration)
 
 # Create the MCP ASGI app mounted at root
 mcp_app = mcp.http_app(path='/')
@@ -168,19 +183,31 @@ async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
             client_ip = (scope.get("client") or ("?", 0))[0]
             async with registration_lock:
                 if not getattr(app.state, "dynamic_tools_registered", False):
-                    _logger.info('%s %s "Initializing dynamic tools (token: %s...)"', "[JIT]", client_ip, token[:8])
+                    _logger.info(
+                        '%s %s "Initializing dynamic tools (token: %s...)"',
+                        "[JIT]",
+                        client_ip,
+                        token[:8],
+                    )
                     success = await asyncio.to_thread(docuengine.init_dynamic_tools, token)
                     if success:
                         app.state.dynamic_tools_registered = True
                     else:
-                        _logger.warning('%s %s "Registration failed — will retry on next request"', "[JIT]", client_ip)
+                        _logger.warning(
+                            '%s %s "Registration failed — will retry on next request"',
+                            "[JIT]",
+                            client_ip,
+                        )
 
         await self.app(scope, receive, send)
 
 
 _OAUTH_NOT_SUPPORTED_BODY = json.dumps({
     "error": "oauth_not_supported",
-    "message": "This server does not support OAuth. Use a pre-configured Bearer token in the Authorization header."
+    "message": (
+        "This server does not support OAuth. Use a pre-configured Bearer "
+        "token in the Authorization header."
+    ),
 }).encode()
 
 _OAUTH_DISCOVERY_PATHS = {
@@ -190,18 +217,25 @@ async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
 }
 
 class RejectOAuthDiscoveryMiddleware:
-    """Return a JSON 404 for OAuth discovery endpoints so MCP clients fall back to Bearer auth."""
+    """Return a JSON 404 for OAuth discovery endpoints."""
     def __init__(self, app: ASGIApp) -> None:
         self.app = app
 
     async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
-        if scope["type"] == "http" and scope["path"].rstrip("/") in {p.rstrip("/") for p in _OAUTH_DISCOVERY_PATHS}:
+        if (
+            scope["type"] == "http"
+            and scope["path"].rstrip("/") in {p.rstrip("/") for p in _OAUTH_DISCOVERY_PATHS}
+        ):
             await send({
                 "type": "http.response.start",
                 "status": 404,
                 "headers": [(b"content-type", b"application/json")],
             })
-            await send({"type": "http.response.body", "body": _OAUTH_NOT_SUPPORTED_BODY, "more_body": False})
+            await send({
+                "type": "http.response.body",
+                "body": _OAUTH_NOT_SUPPORTED_BODY,
+                "more_body": False,
+            })
             return
         await self.app(scope, receive, send)
 
@@ -214,7 +248,7 @@ async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
 app.add_middleware(McpAuditMiddleware)            # innermost — logs MCP JSON-RPC actions
 app.add_middleware(Enrich404Middleware)           # wraps 404s in JSON
 app.add_middleware(TokenQuerystringMiddleware)    # injects auth header + JIT registration
-app.add_middleware(RejectOAuthDiscoveryMiddleware) # short-circuits OAuth discovery paths
+app.add_middleware(RejectOAuthDiscoveryMiddleware)  # short-circuits OAuth discovery paths
 # CORS must be outermost so it runs before anything else on every request,
 # including pre-flight OPTIONS. expose_headers exposes Mcp-Session-Id to browsers.
 # allow_credentials must NOT be True when allow_origins=["*"].
@@ -243,7 +277,9 @@ async def callbacks_endpoint(request: Request):
         return {"status": "error", "message": "Body not a valid JSON"}
 
     cb_obj = callback.get("callback")
-    custom = callback.get("custom") or (cb_obj.get("data") if isinstance(cb_obj, dict) else None)
+    custom = callback.get("custom") or (
+        cb_obj.get("data") if isinstance(cb_obj, dict) else None
+    )
     if not custom:
         _logger.warning('%s %s "Missing callback.custom field"', "[CB]", client_ip)
         return {"status": "error", "message": "'callback.custom' missing from received data"}

src/openapi_mcp_sdk/mcp_audit.py

@@ -13,7 +13,7 @@
 
 import json
 import logging
-from starlette.types import ASGIApp, Scope, Receive, Send
+from starlette.types import ASGIApp, Scope, Receive, Send  # pylint: disable=import-error
 
 _log = logging.getLogger("openapi_mcp_sdk.audit")
 
@@ -42,7 +42,7 @@ def _emit(body: bytes, client_ip: str) -> None:
     """Parse a JSON-RPC body and emit a structured audit log line."""
     try:
         data = json.loads(body)
-    except Exception:
+    except json.JSONDecodeError:
         return
 
     if not isinstance(data, dict) or "method" not in data:
@@ -79,7 +79,7 @@ def _emit(body: bytes, client_ip: str) -> None:
         _log.info('%s %s "%s"', _TAG, client_ip, label)
 
 
-class McpAuditMiddleware:
+class McpAuditMiddleware:  # pylint: disable=too-few-public-methods
     """Non-destructive ASGI middleware that logs MCP JSON-RPC calls.
 
     Wraps the ``receive`` callable to buffer POST body chunks, emits the audit

src/openapi_mcp_sdk/memory_store.py

@@ -1,17 +1,19 @@
-# memory_store.py
-# Singleton in-memory store for callback results
+"""Shared storage for async callback results."""
+
+from __future__ import annotations
+
+import json
 import os
 import logging
 from typing import Dict
 
 logger = logging.getLogger(__name__)
 try:
     from pymemcache.client import base
-    _pymemcache_available = True
+    _PYMEMCACHE_AVAILABLE = True
 except ImportError:
     base = None  # type: ignore[assignment]
-    _pymemcache_available = False
-import json
+    _PYMEMCACHE_AVAILABLE = False
 
 callback_results = {}
 
@@ -37,11 +39,14 @@
         callbackUrl = callbackUrl.replace("alpha", "dev")
 
 MEMCACHED_HOST = os.getenv("MCP_CACHE_HOST", '0.0.0.0')
-MEMCACHED_PORT = int(os.getenv("MCP_CACHE_PORT", 11211))
+MEMCACHED_PORT = int(os.getenv("MCP_CACHE_PORT", "11211"))
 # connect_timeout / timeout = 1 s: when Memcached is unreachable (e.g. local dev,
 # Docker without the VPC network) the client fails fast and the except block
 # falls back to the in-process dict, keeping every endpoint responsive.
-memcached_client = base.Client((MEMCACHED_HOST, MEMCACHED_PORT), connect_timeout=1, timeout=1) if _pymemcache_available else None
+memcached_client = (
+    base.Client((MEMCACHED_HOST, MEMCACHED_PORT), connect_timeout=1, timeout=1)
+    if _PYMEMCACHE_AVAILABLE else None
+)
 
 # Funzioni aggiornate per supportare Memcached
 def get_callback_result(request_id: str):
@@ -52,11 +57,13 @@ def get_callback_result(request_id: str):
         if memcached_client is None:
             raise RuntimeError("pymemcache not available")
         return get_from_memcached(memcached_client, request_id)
-    except Exception as e:
+    except (RuntimeError, OSError, ValueError, TypeError) as e:
         # Fallback al dizionario in memoria
         logger.debug("memcached get failed, falling back to in-memory store: %s", e)
         if request_id not in callback_results:
-            raise KeyError(f"Result not found for request_id: {request_id}")
+            raise KeyError(
+                f"Result not found for request_id: {request_id}"
+            ) from e
         return callback_results[request_id]
 
 
@@ -72,20 +79,20 @@ def set_callback_result(request_id: str, data: Dict, custom: Dict):
         if memcached_client is None:
             raise RuntimeError("pymemcache not available")
         save_to_memcached(memcached_client, request_id, result)
-    except Exception as e:
+    except (RuntimeError, OSError, ValueError, TypeError) as e:
         # Fallback al dizionario in memoria
         logger.debug("memcached set failed, falling back to in-memory store: %s", e)
         callback_results[request_id] = result
 
 def save_to_memcached(client, key, value):
-    # Serialize to JSON and encode as UTF-8
+    """Serialize a callback payload and store it in Memcached."""
     binary_value = json.dumps(value).encode('utf-8')
     client.set(key, binary_value)
 
-# Retrieve data from Memcached
+
 def get_from_memcached(client, key):
+    """Read a callback payload from Memcached and decode it from JSON."""
     binary_value = client.get(key)
     if binary_value is not None:
-        # Decode from UTF-8 and deserialize from JSON
         return json.loads(binary_value.decode('utf-8'))
     return None

tests/pytest/conftest.py

@@ -0,0 +1,9 @@
+import sys
+from pathlib import Path
+
+
+ROOT = Path(__file__).resolve().parents[2]
+SRC = ROOT / "src"
+
+if str(SRC) not in sys.path:
+    sys.path.insert(0, str(SRC))

tests/pytest/test_storage_backend.py

@@ -0,0 +1,40 @@
+from pathlib import Path
+
+import pytest
+
+from openapi_mcp_sdk import storage_backend
+
+
+def test_save_and_read_file_with_local_backend(tmp_path, monkeypatch):
+    monkeypatch.setenv("MCP_STORAGE_BACKEND", "local")
+    monkeypatch.setenv("MCP_STORAGE_PATH", str(tmp_path))
+
+    storage_backend.save_file("nested/file.txt", b"hello", "text/plain")
+
+    content, content_type = storage_backend.read_file("nested/file.txt")
+
+    assert content == b"hello"
+    assert content_type == "application/octet-stream"
+    assert (tmp_path / "nested" / "file.txt").read_bytes() == b"hello"
+
+
+def test_read_file_raises_for_missing_local_file(tmp_path, monkeypatch):
+    monkeypatch.setenv("MCP_STORAGE_BACKEND", "local")
+    monkeypatch.setenv("MCP_STORAGE_PATH", str(tmp_path))
+
+    with pytest.raises(FileNotFoundError):
+        storage_backend.read_file("missing.txt")
+
+
+def test_cloud_backends_require_bucket(monkeypatch):
+    monkeypatch.setenv("MCP_STORAGE_BACKEND", "gcs")
+    monkeypatch.delenv("MCP_STORAGE_BUCKET", raising=False)
+
+    with pytest.raises(RuntimeError, match="MCP_STORAGE_BUCKET"):
+        storage_backend._storage_bucket()
+
+
+def test_storage_path_uses_mcp_storage_path(monkeypatch, tmp_path):
+    monkeypatch.setenv("MCP_STORAGE_PATH", str(tmp_path))
+
+    assert storage_backend._storage_path() == Path(tmp_path)