fixed failing testcases, pylint

Signed-off-by: Veeresh K <[email protected]>

Author: nmveeresh

mcpgateway/middleware/mcp_path_rewrite_middleware.py

@@ -73,12 +73,10 @@ async def __call__(self, scope, receive, send):
             '/tools'
         """
 
-        
         # Only handle HTTP requests, HTTPS uses scope["type"] == "http" in ASGI
         if scope["type"] != "http":
             await self.application(scope, receive, send)
             return
-    
         scope.setdefault("headers", [])
 
         # Call auth check first

mcpgateway/routers/__init__.py

@@ -0,0 +1 @@
+# pragma: no cover

mcpgateway/routers/oauth_router.py

@@ -19,6 +19,7 @@
 
 # First-Party
 from mcpgateway.db import Gateway, get_db
+from mcpgateway.services.gateway_service import GatewayService
 from mcpgateway.services.oauth_manager import OAuthError, OAuthManager
 from mcpgateway.services.token_storage_service import TokenStorageService
 
@@ -28,7 +29,7 @@
 
 
 @oauth_router.get("/authorize/{gateway_id}")
-async def initiate_oauth_flow(gateway_id: str, request: Request, db: Session = Depends(get_db)) -> RedirectResponse:
+async def initiate_oauth_flow(gateway_id: str, _request: Request, db: Session = Depends(get_db)) -> RedirectResponse:
     """Initiates the OAuth 2.0 Authorization Code flow for a specified gateway.
 
     This endpoint retrieves the OAuth configuration for the given gateway, validates that
@@ -37,7 +38,7 @@ async def initiate_oauth_flow(gateway_id: str, request: Request, db: Session = D
 
     Args:
         gateway_id: The unique identifier of the gateway to authorize.
-        request: The FastAPI request object.
+        _request: The FastAPI request object.
         db: The database session dependency.
 
     Returns:
@@ -80,8 +81,7 @@ async def initiate_oauth_flow(gateway_id: str, request: Request, db: Session = D
 async def oauth_callback(
     code: str = Query(..., description="Authorization code from OAuth provider"),
     state: str = Query(..., description="State parameter for CSRF protection"),
-    # Remove the gateway_id parameter requirement
-    request: Request = None,
+    _request: Request = None,
     db: Session = Depends(get_db),
 ) -> HTMLResponse:
     """Handle the OAuth callback and complete the authorization process.
@@ -93,7 +93,7 @@ async def oauth_callback(
     Args:
         code (str): The authorization code returned by the OAuth provider.
         state (str): The state parameter for CSRF protection, which encodes the gateway ID.
-        request (Request): The incoming HTTP request object.
+        _request (Request): The incoming HTTP request object.
         db (Session): The database session dependency.
 
     Returns:
@@ -349,14 +349,14 @@ async def get_oauth_status(gateway_id: str, db: Session = Depends(get_db)) -> di
                 "redirect_uri": oauth_config.get("redirect_uri"),
                 "message": "Gateway configured for Authorization Code flow",
             }
-        else:
-            return {
-                "oauth_enabled": True,
-                "grant_type": grant_type,
-                "client_id": oauth_config.get("client_id"),
-                "scopes": oauth_config.get("scopes", []),
-                "message": f"Gateway configured for {grant_type} flow",
-            }
+
+        return {
+            "oauth_enabled": True,
+            "grant_type": grant_type,
+            "client_id": oauth_config.get("client_id"),
+            "scopes": oauth_config.get("scopes", []),
+            "message": f"Gateway configured for {grant_type} flow",
+        }
 
     except HTTPException:
         raise
@@ -380,9 +380,6 @@ async def fetch_tools_after_oauth(gateway_id: str, db: Session = Depends(get_db)
         HTTPException: If fetching tools fails
     """
     try:
-        # First-Party
-        from mcpgateway.services.gateway_service import GatewayService
-
         gateway_service = GatewayService()
         result = await gateway_service.fetch_tools_after_oauth(db, gateway_id)
         tools_count = len(result.get("tools", []))

mcpgateway/routers/reverse_proxy.py

@@ -150,13 +150,13 @@ def list_sessions(self) -> list[Dict[str, Any]]:
 @reverse_proxy_router.websocket("/ws")
 async def websocket_endpoint(
     websocket: WebSocket,
-    db: Session = Depends(get_db),
+    _db: Session = Depends(get_db),
 ):
     """WebSocket endpoint for reverse proxy connections.
 
     Args:
         websocket: WebSocket connection.
-        db: Database session.
+        _db: Database session.
     """
     await websocket.accept()
 
@@ -230,13 +230,13 @@ async def websocket_endpoint(
 
 @reverse_proxy_router.get("/sessions")
 async def list_sessions(
-    request: Request,
+    _request: Request,
     _: str | dict = Depends(require_auth),
 ):
     """List all active reverse proxy sessions.
 
     Args:
-        request: HTTP request.
+        _request: HTTP request.
         _: Authenticated user info (used for auth check).
 
     Returns:
@@ -248,14 +248,14 @@ async def list_sessions(
 @reverse_proxy_router.delete("/sessions/{session_id}")
 async def disconnect_session(
     session_id: str,
-    request: Request,
+    _request: Request,
     _: str | dict = Depends(require_auth),
 ):
     """Disconnect a reverse proxy session.
 
     Args:
         session_id: Session ID to disconnect.
-        request: HTTP request.
+        _request: HTTP request.
         _: Authenticated user info (used for auth check).
 
     Returns:
@@ -279,15 +279,15 @@ async def disconnect_session(
 async def send_request_to_session(
     session_id: str,
     mcp_request: Dict[str, Any],
-    request: Request,
+    _request: Request,
     _: str | dict = Depends(require_auth),
 ):
     """Send an MCP request to a reverse proxy session.
 
     Args:
         session_id: Session ID to send request to.
         mcp_request: MCP request to send.
-        request: HTTP request.
+        _request: HTTP request.
         _: Authenticated user info (used for auth check).
 
     Returns:

mcpgateway/routers/setup_routes.py

@@ -25,7 +25,6 @@
     tag_router,
     tool_router,
     utility_router,
-    version_router,
     well_known_router,
 )
 

mcpgateway/routers/v1/__init__.py

@@ -0,0 +1 @@
+# pragma: no cover

mcpgateway/routers/well_known.py

@@ -75,7 +75,7 @@ def validate_security_txt(content: str) -> Optional[str]:
 
 
 @well_known_router.get("/.well-known/{filename:path}", include_in_schema=False)
-async def get_well_known_file(filename: str, response: Response, request: Request):
+async def get_well_known_file(filename: str, _response: Response, _request: Request):
     """
     Serve well-known URI files.
 
@@ -86,8 +86,8 @@ async def get_well_known_file(filename: str, response: Response, request: Reques
 
     Args:
         filename: The well-known filename requested
-        response: FastAPI response object for headers
-        request: FastAPI request object for logging
+        _response: FastAPI response object for headers
+        _request: FastAPI request object for logging
 
     Returns:
         Plain text content of the requested file
@@ -110,7 +110,7 @@ async def get_well_known_file(filename: str, response: Response, request: Reques
         return PlainTextResponse(content=settings.well_known_robots_txt, media_type="text/plain; charset=utf-8", headers=headers)
 
     # Handle security.txt
-    elif filename == "security.txt":
+    if filename == "security.txt":
         if not settings.well_known_security_txt_enabled:
             raise HTTPException(status_code=404, detail="security.txt not configured")
 
@@ -121,7 +121,7 @@ async def get_well_known_file(filename: str, response: Response, request: Reques
         return PlainTextResponse(content=content, media_type="text/plain; charset=utf-8", headers=common_headers)
 
     # Handle custom files
-    elif filename in settings.custom_well_known_files:
+    if filename in settings.custom_well_known_files:
         content = settings.custom_well_known_files[filename]
 
         # Determine content type
@@ -131,22 +131,20 @@ async def get_well_known_file(filename: str, response: Response, request: Reques
 
         return PlainTextResponse(content=content, media_type=content_type, headers=common_headers)
 
-    # File not found
-    else:
-        # Provide helpful error for known well-known URIs
-        if filename in WELL_KNOWN_REGISTRY:
-            raise HTTPException(status_code=404, detail=f"{filename} is not configured. This is a {WELL_KNOWN_REGISTRY[filename]['description']} file.")
-        else:
-            raise HTTPException(status_code=404, detail="Not found")
+    # File not found - provide helpful error for known well-known URIs
+    if filename in WELL_KNOWN_REGISTRY:
+        raise HTTPException(status_code=404, detail=f"{filename} is not configured. This is a {WELL_KNOWN_REGISTRY[filename]['description']} file.")
+
+    raise HTTPException(status_code=404, detail="Not found")
 
 
 @well_known_router.get("/admin/well-known", response_model=dict)
-async def get_well_known_status(user: str = Depends(require_auth)):
+async def get_well_known_status(_user: str = Depends(require_auth)):
     """
     Get status of well-known URI configuration.
 
     Args:
-        user: Authenticated user from dependency injection.
+        _user: Authenticated user from dependency injection.
 
     Returns:
         Dict containing well-known configuration status and available files.

mcpgateway/utils/url_utils.py

@@ -21,6 +21,7 @@ def get_protocol_from_request(request: Request) -> str:
     if forwarded:
         # may be a comma-separated list; take the first
         return forwarded.split(",")[0].strip()
+
     return request.url.scheme
 
 
@@ -37,5 +38,6 @@ def update_url_protocol(request: Request) -> str:
     parsed = urlparse(str(request.base_url))
     proto = get_protocol_from_request(request)
     new_parsed = parsed._replace(scheme=proto)
+
     # urlunparse keeps netloc and path intact
     return urlunparse(new_parsed).rstrip("/")

tests/conftest.py

@@ -1,10 +1,8 @@
 # -*- coding: utf-8 -*-
-"""
-
+"""Location: ./tests/conftest.py
 Copyright 2025
 SPDX-License-Identifier: Apache-2.0
 Authors: Mihai Criveti
-
 """
 
 # Standard
@@ -36,7 +34,7 @@ def event_loop():
 @pytest.fixture(scope="session")
 def test_db_url():
     """Return the URL for the test database."""
-    return "sqlite:///./test.db"
+    return "sqlite:///:memory:"
 
 
 @pytest.fixture(scope="session")
@@ -74,13 +72,46 @@ def test_settings():
 
 
 @pytest.fixture
-def app(test_settings):
-    """Create a FastAPI test application."""
-    with patch("mcpgateway.config.get_settings", return_value=test_settings):
-        # First-Party
-        from mcpgateway.main import app
+def app():
+    """Create a FastAPI test application with proper database setup."""
+    # Use the existing app_with_temp_db fixture logic which works correctly
+    mp = MonkeyPatch()
+
+    # 1) create temp SQLite file
+    fd, path = tempfile.mkstemp(suffix=".db")
+    url = f"sqlite:///{path}"
+
+    # 2) patch settings
+    from mcpgateway.config import settings
+    mp.setattr(settings, "database_url", url, raising=False)
+
+    # First-Party
+    import mcpgateway.db as db_mod
+
+    engine = create_engine(url, connect_args={"check_same_thread": False}, poolclass=StaticPool)
+    TestSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
+    mp.setattr(db_mod, "engine", engine, raising=False)
+    mp.setattr(db_mod, "SessionLocal", TestSessionLocal, raising=False)
 
-        yield app
+    # 4) patch the already‑imported main module **without reloading**
+    import mcpgateway.main as main_mod
+    mp.setattr(main_mod, "SessionLocal", TestSessionLocal, raising=False)
+    # (patch engine too if your code references it)
+    mp.setattr(main_mod, "engine", engine, raising=False)
+
+    # 4) create schema
+    db_mod.Base.metadata.create_all(bind=engine)
+
+    # First-Party
+    from mcpgateway.main import app
+
+    yield app
+
+    # 6) teardown
+    mp.undo()
+    engine.dispose()
+    os.close(fd)
+    os.unlink(path)
 
 
 @pytest.fixture
@@ -164,4 +195,4 @@ def app_with_temp_db():
     mp.undo()
     engine.dispose()
     os.close(fd)
-    os.unlink(path)
+    os.unlink(path)