fix e2e tests

Signed-off-by: Madhav Kandukuri <[email protected]>

Author: madhav165

tests/conftest.py

@@ -10,12 +10,15 @@
 # Standard
 import asyncio
 import os
+import tempfile
 from unittest.mock import AsyncMock, patch
 
 # Third-Party
+from _pytest.monkeypatch import MonkeyPatch
 import pytest
 from sqlalchemy import create_engine
 from sqlalchemy.orm import sessionmaker
+from sqlalchemy.pool import StaticPool
 
 # First-Party
 from mcpgateway.config import Settings
@@ -107,3 +110,57 @@ def mock_websocket():
 #     import mcpgateway.translate as translate
 #     translate._run_stdio_to_sse = AsyncMock(return_value=None)
 #     translate._run_sse_to_stdio = AsyncMock(return_value=None)
+
+
+@pytest.fixture(scope="module")  # one DB per test module is usually fine
+def app_with_temp_db():
+    """Return a FastAPI app wired to a fresh SQLite database."""
+    mp = MonkeyPatch()
+
+    # 1) create temp SQLite file
+    fd, path = tempfile.mkstemp(suffix=".db")
+    url = f"sqlite:///{path}"
+
+    # 2) patch settings
+    # First-Party
+    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)
+
+    # 4) patch the already‑imported main module **without reloading**
+    # First-Party
+    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)
+
+    # 5) reload main so routers, deps pick up new SessionLocal
+    # if "mcpgateway.main" in sys.modules:
+    #     import importlib
+
+    #     importlib.reload(sys.modules["mcpgateway.main"])
+    # else:
+    #     import mcpgateway.main  # noqa: F401
+
+    # First-Party
+    from mcpgateway.main import app
+
+    yield app
+
+    # 6) teardown
+    mp.undo()
+    engine.dispose()
+    os.close(fd)
+    os.unlink(path)

tests/e2e/test_admin_apis.py

@@ -35,24 +35,18 @@
 os.environ["MCPGATEWAY_UI_ENABLED"] = "true"
 
 # Standard
-import tempfile
-from typing import AsyncGenerator
+import logging
 from unittest.mock import patch
 from urllib.parse import quote
 import uuid
-import logging
 
 # Third-Party
 from httpx import AsyncClient
 import pytest
 import pytest_asyncio
-from sqlalchemy import create_engine
-from sqlalchemy.orm import sessionmaker
-from sqlalchemy.pool import StaticPool
 
-# First-Party
-from mcpgateway.db import Base
-from mcpgateway.main import app, get_db
+# from mcpgateway.db import Base
+# from mcpgateway.main import app, get_db
 
 
 # Configure logging for debugging
@@ -77,72 +71,23 @@ def setup_logging():
 # Fixtures
 # -------------------------
 @pytest_asyncio.fixture
-async def temp_db():
-    """
-    Create a temporary SQLite database for testing.
-
-    This fixture creates a fresh database for each test, ensuring complete
-    isolation between tests. The database is automatically cleaned up after
-    the test completes.
-    """
-    # Create temporary file for SQLite database
-    db_fd, db_path = tempfile.mkstemp(suffix=".db")
-
-    # Create engine with SQLite
-    engine = create_engine(
-        f"sqlite:///{db_path}",
-        connect_args={"check_same_thread": False},
-        poolclass=StaticPool,
-    )
-
-    # Log the database path for debugging
-    logging.debug(f"Using temporary database at: {db_path}")
-
-    # Create all tables
-    Base.metadata.create_all(bind=engine)
-
-    # Create session factory
-    TestSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
-
-    # Override the get_db dependency
-    def override_get_db():
-        db = TestSessionLocal()
-        try:
-            yield db
-        finally:
-            db.close()
-
-    app.dependency_overrides[get_db] = override_get_db
-
-    # Override authentication for all tests
+async def client(app_with_temp_db):
     # First-Party
     from mcpgateway.utils.verify_credentials import require_auth, require_basic_auth
 
-    def override_auth():
-        return TEST_USER
-
-    app.dependency_overrides[require_auth] = override_auth
-    app.dependency_overrides[require_basic_auth] = override_auth
-
-    yield engine
+    app_with_temp_db.dependency_overrides[require_auth] = lambda: TEST_USER
+    app_with_temp_db.dependency_overrides[require_basic_auth] = lambda: TEST_USER
 
-    # Cleanup
-    logging.debug("Cleaning up temporary database and overrides.")
-    app.dependency_overrides.clear()
-    os.close(db_fd)
-    os.unlink(db_path)
-
-
-@pytest_asyncio.fixture
-async def client(temp_db) -> AsyncGenerator[AsyncClient, None]:
-    """Create an async test client with the test database."""
     # Third-Party
     from httpx import ASGITransport, AsyncClient
 
-    transport = ASGITransport(app=app)
+    transport = ASGITransport(app=app_with_temp_db)
     async with AsyncClient(transport=transport, base_url="http://test") as ac:
         yield ac
 
+    app_with_temp_db.dependency_overrides.pop(require_auth, None)
+    app_with_temp_db.dependency_overrides.pop(require_basic_auth, None)
+
 
 @pytest_asyncio.fixture
 async def mock_settings():
@@ -351,7 +296,7 @@ async def test_admin_tool_name_conflict(self, client: AsyncClient, mock_settings
 class TestAdminResourceAPIs:
     """Test admin resource management endpoints."""
 
-    async def test_admin_add_resource(self, client: AsyncClient, mock_settings, temp_db):
+    async def test_admin_add_resource(self, client: AsyncClient, mock_settings):
         """Test adding a resource via the admin UI with new logic."""
         # Define valid form data
         valid_form_data = {

tests/security/test_rpc_api.py

@@ -16,6 +16,8 @@
 import json
 import os
 import sys
+
+# Third-Party
 import pytest
 
 try:
@@ -25,6 +27,7 @@
     print("Please install requests: pip install requests")
     sys.exit(1)
 
+
 @pytest.mark.skip(reason="Disabled temporarily as this requires a live MCP Gateway instance")
 def test_rpc_vulnerability():
     """Test the RPC endpoint with malicious method names."""