fix: playwright admin_api fixture to stop duplicating JWT auth, fix linting

Signed-off-by: Marek Dano <Marek.Dano@ibm.com>

Author: Marek Dano

mcpgateway/alembic/versions/d3e4f5a6b7c8_add_binding_reference_id_to_tool_plugin_bindings.py

@@ -10,8 +10,8 @@
 from typing import Sequence, Union
 
 # Third-Party
-import sqlalchemy as sa
 from alembic import op
+import sqlalchemy as sa
 
 # revision identifiers, used by Alembic.
 revision: str = "d3e4f5a6b7c8"  # pragma: allowlist secret

mcpgateway/config.py

@@ -310,7 +310,7 @@ class Settings(BaseSettings):
     basic_auth_user: str = "admin"
     basic_auth_password: SecretStr = Field(default=SecretStr("changeme"))
     jwt_algorithm: str = "HS256"
-    jwt_secret_key: SecretStr = Field(default=SecretStr("my-test-key"))
+    jwt_secret_key: SecretStr = Field(default=SecretStr("my-test-key-but-now-longer-than-32-bytes"))
     jwt_public_key_path: str = ""
     jwt_private_key_path: str = ""
     jwt_audience: str = "mcpgateway-api"

mcpgateway/middleware/request_logging_middleware.py

@@ -62,8 +62,8 @@
 
 # First-Party
 from mcpgateway.auth import get_current_user
-from mcpgateway.config import settings
 from mcpgateway.common.validators import SecurityValidator
+from mcpgateway.config import settings
 from mcpgateway.middleware.path_filter import should_skip_request_logging
 from mcpgateway.services.logging_service import LoggingService
 from mcpgateway.services.structured_logger import get_structured_logger

mcpgateway/services/tool_plugin_binding_service.py

@@ -173,15 +173,9 @@ def upsert_bindings(
                         # different external references are claiming the same (team, tool, plugin)
                         # triple.  The new reference_id wins (last-caller-wins), but the old
                         # caller's DELETE by reference will now be a no-op.
-                        if (
-                            existing.binding_reference_id
-                            and policy.binding_reference_id
-                            and existing.binding_reference_id != policy.binding_reference_id
-                        ):
+                        if existing.binding_reference_id and policy.binding_reference_id and existing.binding_reference_id != policy.binding_reference_id:
                             logger.warning(
-                                "binding_reference_id ownership transfer: "
-                                "team=%s tool=%s plugin=%s old_ref=%s new_ref=%s — "
-                                "DELETE by old_ref will now be a no-op",
+                                "binding_reference_id ownership transfer: " "team=%s tool=%s plugin=%s old_ref=%s new_ref=%s — " "DELETE by old_ref will now be a no-op",
                                 team_id,
                                 tool_name,
                                 policy.plugin_id.value,
@@ -284,8 +278,7 @@ def list_bindings(
         if binding_reference_id:
             if team_id:
                 logger.warning(
-                    "Both team_id=%r and binding_reference_id=%r supplied to list_bindings; "
-                    "team_id will be ignored. Omit team_id when filtering by binding_reference_id.",
+                    "Both team_id=%r and binding_reference_id=%r supplied to list_bindings; " "team_id will be ignored. Omit team_id when filtering by binding_reference_id.",
                     team_id,
                     binding_reference_id,
                 )

tests/playwright/conftest.py

@@ -330,6 +330,55 @@ def api_request_context(playwright: Playwright) -> Generator[APIRequestContext,
     request_context.dispose()
 
 
+@pytest.fixture(scope="session")
+def admin_api(playwright: Playwright) -> Generator[APIRequestContext, None, None]:
+    """Consolidated admin-authenticated API context for all Playwright tests.
+
+    Token resolution priority (fail-closed):
+    1. MCP_AUTH env var (from Makefile / compose bootstrap) — preferred
+    2. MCPGATEWAY_BEARER_TOKEN env var (legacy name used by other tools)
+    3. Locally-signed JWT using Settings().jwt_secret_key (fallback)
+
+    This fixture replaces the four duplicate admin_api/owasp_admin_api fixtures
+    previously scattered across entities, OWASP, operations, and teams conftests.
+    Per-directory non-admin fixtures (viewer_api, owasp_user_a_api, etc.) remain
+    local since they intentionally mint throwaway users.
+    """
+    headers = {"Accept": "application/json"}
+
+    # Priority 1: MCP_AUTH (Makefile-generated token signed with gateway's secret)
+    token = os.getenv("MCP_AUTH", "")
+    if not token:
+        # Priority 2: MCPGATEWAY_BEARER_TOKEN (legacy name)
+        token = os.getenv("MCPGATEWAY_BEARER_TOKEN", "")
+    if not token and not DISABLE_JWT_FALLBACK:
+        # Priority 3: Locally-signed JWT fallback
+        try:
+            token = _create_jwt_token(
+                {"sub": ADMIN_EMAIL},
+                user_data={
+                    "email": ADMIN_EMAIL,
+                    "full_name": "Test Admin",
+                    "is_admin": True,
+                    "auth_provider": "test",
+                },
+                teams=None,  # Admin bypass: null teams with is_admin=true
+            )
+        except Exception:
+            pass  # Use empty if generation fails
+
+    auth_header = _format_auth_header(token)
+    if auth_header:
+        headers["Authorization"] = auth_header
+
+    request_context = playwright.request.new_context(
+        base_url=BASE_URL,
+        extra_http_headers=headers,
+    )
+    yield request_context
+    request_context.dispose()
+
+
 @pytest.fixture(scope="session")
 def browser_context_args(
     pytestconfig,

tests/playwright/entities/test_entity_lifecycle.py

@@ -36,23 +36,6 @@ def _make_jwt(email: str, is_admin: bool = False, teams=None) -> str:
     )
 
 
-@pytest.fixture(scope="module")
-def admin_api(playwright: Playwright):
-    """Admin-authenticated API context.
-
-    Prefers the ``MCP_AUTH`` env var (set by the Makefile from a token signed with
-    the running gateway's secret) so signatures match the deployed instance. Falls
-    back to a locally-signed JWT only when ``MCP_AUTH`` is unset.
-    """
-    token = os.getenv("MCP_AUTH", "") or _make_jwt("admin@example.com", is_admin=True)
-    ctx = playwright.request.new_context(
-        base_url=BASE_URL,
-        extra_http_headers={"Authorization": f"Bearer {token}", "Accept": "application/json"},
-    )
-    yield ctx
-    ctx.dispose()
-
-
 @pytest.fixture(scope="module")
 def viewer_api(playwright: Playwright):
     """Viewer (non-admin, no RBAC roles) API context for permission checks."""

tests/playwright/operations/conftest.py

@@ -29,23 +29,6 @@ def _make_jwt(email: str, is_admin: bool = False, teams=None) -> str:
     )
 
 
-@pytest.fixture(scope="module")
-def admin_api(playwright: Playwright) -> Generator[APIRequestContext, None, None]:
-    """Admin-authenticated API context.
-
-    Prefers the ``MCP_AUTH`` env var (set by the Makefile from a token signed with
-    the running gateway's secret) so signatures match the deployed instance. Falls
-    back to a locally-signed JWT only when ``MCP_AUTH`` is unset.
-    """
-    token = os.getenv("MCP_AUTH", "") or _make_jwt("admin@example.com", is_admin=True)
-    ctx = playwright.request.new_context(
-        base_url=BASE_URL,
-        extra_http_headers={"Authorization": f"Bearer {token}", "Accept": "application/json"},
-    )
-    yield ctx
-    ctx.dispose()
-
-
 @pytest.fixture(scope="module")
 def non_admin_api(playwright: Playwright) -> Generator[APIRequestContext, None, None]:
     """Non-admin API context for permission checks."""

tests/playwright/security/owasp/conftest.py

@@ -53,28 +53,11 @@ def owasp_anon_api(playwright: Playwright) -> Generator[APIRequestContext, None,
     ctx.dispose()
 
 
-@pytest.fixture(scope="module")
-def owasp_admin_api(playwright: Playwright) -> Generator[APIRequestContext, None, None]:
-    """Admin-authenticated API context for OWASP A01 tests (admin bypass via teams=null).
-
-    Prefers the ``MCP_AUTH`` env var (set by the Makefile from a token signed with
-    the running gateway's secret) so signatures match the deployed instance. Falls
-    back to a locally-signed JWT only when ``MCP_AUTH`` is unset.
-    """
-    token = os.getenv("MCP_AUTH", "") or _make_jwt("admin@example.com", is_admin=True)
-    ctx = playwright.request.new_context(
-        base_url=BASE_URL,
-        extra_http_headers={"Authorization": f"Bearer {token}", "Accept": "application/json"},
-    )
-    yield ctx
-    ctx.dispose()
-
-
 @pytest.fixture
-def owasp_user_a_api(owasp_admin_api: APIRequestContext, playwright: Playwright):
+def owasp_user_a_api(admin_api: APIRequestContext, playwright: Playwright):
     """Non-admin API context for User A, registered in the system. Cleans up after test."""
     email = f"owasp-a-{uuid.uuid4().hex[:8]}@example.com"
-    create_resp = owasp_admin_api.post(
+    create_resp = admin_api.post(
         "/auth/email/admin/users",
         data={"email": email, "password": TEST_PASSWORD, "full_name": "OWASP User A"},
     )
@@ -85,14 +68,14 @@ def owasp_user_a_api(owasp_admin_api: APIRequestContext, playwright: Playwright)
     yield {"ctx": ctx, "email": email}
     ctx.dispose()
     with suppress(Exception):
-        owasp_admin_api.delete(f"/auth/email/admin/users/{email}")
+        admin_api.delete(f"/auth/email/admin/users/{email}")
 
 
 @pytest.fixture
-def owasp_user_b_api(owasp_admin_api: APIRequestContext, playwright: Playwright):
+def owasp_user_b_api(admin_api: APIRequestContext, playwright: Playwright):
     """Non-admin API context for User B, registered in the system. Cleans up after test."""
     email = f"owasp-b-{uuid.uuid4().hex[:8]}@example.com"
-    create_resp = owasp_admin_api.post(
+    create_resp = admin_api.post(
         "/auth/email/admin/users",
         data={"email": email, "password": TEST_PASSWORD, "full_name": "OWASP User B"},
     )
@@ -103,11 +86,11 @@ def owasp_user_b_api(owasp_admin_api: APIRequestContext, playwright: Playwright)
     yield {"ctx": ctx, "email": email}
     ctx.dispose()
     with suppress(Exception):
-        owasp_admin_api.delete(f"/auth/email/admin/users/{email}")
+        admin_api.delete(f"/auth/email/admin/users/{email}")
 
 
 @pytest.fixture
-def two_teams_setup(owasp_admin_api: APIRequestContext, playwright: Playwright):
+def two_teams_setup(admin_api: APIRequestContext, playwright: Playwright):
     """Create two distinct teams with separate scoped tokens. Cleans up after test."""
     suffix = uuid.uuid4().hex[:8]
     team_a_id: str | None = None
@@ -118,25 +101,25 @@ def two_teams_setup(owasp_admin_api: APIRequestContext, playwright: Playwright):
     ctx_b = None
     try:
         # Team A
-        resp_a = owasp_admin_api.post("/teams/", data={"name": f"owasp-team-a-{suffix}", "description": "OWASP Team A", "visibility": "private"})
+        resp_a = admin_api.post("/teams/", data={"name": f"owasp-team-a-{suffix}", "description": "OWASP Team A", "visibility": "private"})
         assert resp_a.status in (200, 201), f"Failed creating Team A: {resp_a.status} {resp_a.text()}"
         team_a_id = resp_a.json()["id"]
 
         # Team B
-        resp_b = owasp_admin_api.post("/teams/", data={"name": f"owasp-team-b-{suffix}", "description": "OWASP Team B", "visibility": "private"})
+        resp_b = admin_api.post("/teams/", data={"name": f"owasp-team-b-{suffix}", "description": "OWASP Team B", "visibility": "private"})
         assert resp_b.status in (200, 201), f"Failed creating Team B: {resp_b.status} {resp_b.text()}"
         team_b_id = resp_b.json()["id"]
 
         # Server owned by Team A
-        srv_a = owasp_admin_api.post(
+        srv_a = admin_api.post(
             "/servers",
             data={"server": {"name": f"owasp-srv-a-{suffix}", "description": "Team A server"}, "team_id": team_a_id, "visibility": "team"},
         )
         assert srv_a.status in (200, 201), f"Failed creating Team A server: {srv_a.status} {srv_a.text()}"
         server_a_id = srv_a.json()["id"]
 
         # Server owned by Team B
-        srv_b = owasp_admin_api.post(
+        srv_b = admin_api.post(
             "/servers",
             data={"server": {"name": f"owasp-srv-b-{suffix}", "description": "Team B server"}, "team_id": team_b_id, "visibility": "team"},
         )
@@ -163,24 +146,24 @@ def two_teams_setup(owasp_admin_api: APIRequestContext, playwright: Playwright):
             ctx_b.dispose()
         if server_a_id:
             with suppress(Exception):
-                owasp_admin_api.delete(f"/servers/{server_a_id}")
+                admin_api.delete(f"/servers/{server_a_id}")
         if server_b_id:
             with suppress(Exception):
-                owasp_admin_api.delete(f"/servers/{server_b_id}")
+                admin_api.delete(f"/servers/{server_b_id}")
         if team_a_id:
             with suppress(Exception):
-                owasp_admin_api.delete(f"/teams/{team_a_id}")
+                admin_api.delete(f"/teams/{team_a_id}")
         if team_b_id:
             with suppress(Exception):
-                owasp_admin_api.delete(f"/teams/{team_b_id}")
+                admin_api.delete(f"/teams/{team_b_id}")
 
 
 @pytest.fixture
-def private_server_owned_by_user_a(owasp_admin_api: APIRequestContext, owasp_user_a_api: dict):
+def private_server_owned_by_user_a(admin_api: APIRequestContext, owasp_user_a_api: dict):
     """Create a private server via admin on behalf of User A and return its ID. Cleans up after test."""
     server_id: str | None = None
     try:
-        resp = owasp_admin_api.post(
+        resp = admin_api.post(
             "/servers",
             data={
                 "server": {"name": f"owasp-priv-{uuid.uuid4().hex[:8]}", "description": "User A private server"},
@@ -195,4 +178,4 @@ def private_server_owned_by_user_a(owasp_admin_api: APIRequestContext, owasp_use
     finally:
         if server_id:
             with suppress(Exception):
-                owasp_admin_api.delete(f"/servers/{server_id}")
+                admin_api.delete(f"/servers/{server_id}")

tests/playwright/security/owasp/test_a01_broken_access_control.py

@@ -213,10 +213,10 @@ class TestVerticalPrivilegeEscalation:
     """CWE-269, CWE-285 – Non-admin authenticated users cannot call admin-only endpoints."""
 
     @pytest.fixture(scope="class")
-    def non_admin_ctx(self, owasp_admin_api: APIRequestContext, playwright: Playwright) -> APIRequestContext:
-        """Register a non-admin user via owasp_admin_api, then yield an API context for them."""
+    def non_admin_ctx(self, admin_api: APIRequestContext, playwright: Playwright) -> APIRequestContext:
+        """Register a non-admin user via admin_api, then yield an API context for them."""
         email = f"nonadmin-a01-{uuid.uuid4().hex[:8]}@example.com"
-        create_resp = owasp_admin_api.post(
+        create_resp = admin_api.post(
             "/auth/email/admin/users",
             data={"email": email, "password": TEST_PASSWORD, "full_name": "Non-Admin A01"},
         )
@@ -226,7 +226,7 @@ def non_admin_ctx(self, owasp_admin_api: APIRequestContext, playwright: Playwrig
         yield ctx
         ctx.dispose()
         with suppress(Exception):
-            owasp_admin_api.delete(f"/auth/email/admin/users/{email}")
+            admin_api.delete(f"/auth/email/admin/users/{email}")
 
     def test_non_admin_cannot_list_all_users(self, non_admin_ctx: APIRequestContext) -> None:
         resp = non_admin_ctx.get("/auth/email/admin/users")

tests/playwright/teams/conftest.py

@@ -71,23 +71,6 @@ def invite_and_accept(admin_api: APIRequestContext, playwright: Playwright, team
     return inv_data
 
 
-@pytest.fixture(scope="module")
-def admin_api(playwright: Playwright) -> Generator[APIRequestContext, None, None]:
-    """Admin-authenticated API context for team tests.
-
-    Prefers the ``MCP_AUTH`` env var (set by the Makefile from a token signed with
-    the running gateway's secret) so signatures match the deployed instance. Falls
-    back to a locally-signed JWT only when ``MCP_AUTH`` is unset.
-    """
-    token = os.getenv("MCP_AUTH", "") or _make_jwt("admin@example.com", is_admin=True)
-    ctx = playwright.request.new_context(
-        base_url=BASE_URL,
-        extra_http_headers={"Authorization": f"Bearer {token}", "Accept": "application/json"},
-    )
-    yield ctx
-    ctx.dispose()
-
-
 @pytest.fixture(scope="module")
 def private_team(admin_api: APIRequestContext):
     """Create a private team for invitation tests, cleanup after module."""