refactor: deduplicate E2E test helpers and export ROLE_LEVELS

- Unify _create_full_access_key, _create_scoped_key, and
  _create_api_key_with_scope into a single _create_api_key helper
  with optional scope and skip_on_error parameters
- Extract _outline_api helper to eliminate repeated httpx.post
  boilerplate in _get_user_id, _get_user_role, _set_user_role
- Rename _ROLE_LEVELS to ROLE_LEVELS and export from __init__.py

https://claude.ai/code/session_0122umEU4tP9VMzCTrV6SdZN

Author: claude

src/mcp_outline/features/dynamic_tools/__init__.py

@@ -11,11 +11,13 @@
     install_dynamic_tool_list,
 )
 from mcp_outline.features.dynamic_tools.introspect import (
+    ROLE_LEVELS,
     build_role_blocked_map,
     build_tool_endpoint_map,
 )
 
 __all__ = [
+    "ROLE_LEVELS",
     "build_role_blocked_map",
     "build_tool_endpoint_map",
     "get_blocked_tools",

src/mcp_outline/features/dynamic_tools/introspect.py

@@ -14,7 +14,7 @@
     from mcp.server.fastmcp import FastMCP
 
 # Outline workspace roles ordered by privilege level.
-_ROLE_LEVELS: dict[str, int] = {
+ROLE_LEVELS: dict[str, int] = {
     "viewer": 0,
     "member": 1,
     "admin": 2,
@@ -51,19 +51,19 @@ def build_role_blocked_map(
     (``collections.ts``, ``documents.ts``) and
     ``AuthenticationHelper.ts``.
     """
-    blocked: dict[str, set[str]] = {r: set() for r in _ROLE_LEVELS}
+    blocked: dict[str, set[str]] = {r: set() for r in ROLE_LEVELS}
     for name, tool in mcp._tool_manager._tools.items():
         min_role = (tool.meta or {}).get("min_role")
         if min_role is None:
             continue
-        if min_role not in _ROLE_LEVELS:
+        if min_role not in ROLE_LEVELS:
             raise ValueError(
                 f"Tool '{name}' has invalid min_role "
                 f"'{min_role}'; expected one of "
-                f"{set(_ROLE_LEVELS)}"
+                f"{set(ROLE_LEVELS)}"
             )
-        min_level = _ROLE_LEVELS[min_role]
-        for role, level in _ROLE_LEVELS.items():
+        min_level = ROLE_LEVELS[min_role]
+        for role, level in ROLE_LEVELS.items():
             if level < min_level:
                 blocked[role].add(name)
     return {r: frozenset(s) for r, s in blocked.items()}

tests/e2e/conftest.py

@@ -83,6 +83,20 @@ def _parse_set_cookies(response):
     return cookies
 
 
+def _outline_api(
+    token: str,
+    endpoint: str,
+    json: dict | None = None,
+) -> httpx.Response:
+    """POST to an Outline API endpoint with Bearer auth."""
+    return httpx.post(
+        f"{OUTLINE_URL}/api/{endpoint}",
+        headers={"Authorization": f"Bearer {token}"},
+        json=json or {},
+        timeout=30.0,
+    )
+
+
 def _require_redirect(resp, step_description: str) -> str:
     """Extract the ``Location`` header from a redirect response.
 
@@ -215,17 +229,30 @@ def _login_once(
     return access_token
 
 
-def _create_full_access_key(access_token: str) -> str:
-    """Create a full-access API key and return its value."""
-    resp = httpx.post(
-        f"{OUTLINE_URL}/api/apiKeys.create",
-        headers={
-            "Authorization": f"Bearer {access_token}",
-        },
-        json={"name": "e2e-test"},
-        timeout=30.0,
-    )
-    resp.raise_for_status()
+def _create_api_key(
+    access_token: str,
+    name: str = "e2e-test",
+    scope: list | None = None,
+    *,
+    skip_on_error: bool = False,
+) -> str:
+    """Create an Outline API key and return its value.
+
+    Args:
+        access_token: OIDC session token (Bearer).
+        name: Key name shown in Outline Settings.
+        scope: Scope array, or ``None`` for full access.
+        skip_on_error: If ``True``, call ``pytest.skip``
+            instead of raising on non-200 responses.
+    """
+    json_body: dict = {"name": name}
+    if scope is not None:
+        json_body["scope"] = scope
+    resp = _outline_api(access_token, "apiKeys.create", json_body)
+    if resp.status_code != 200:
+        if skip_on_error:
+            pytest.skip(f"apiKeys.create returned {resp.status_code}")
+        resp.raise_for_status()
     return resp.json()["data"]["value"]
 
 
@@ -237,19 +264,13 @@ def _login_and_create_api_key():
     token.
     """
     token = _login("admin@example.com", "admin")
-    key = _create_full_access_key(token)
+    key = _create_api_key(token)
     return key, token
 
 
 def _get_user_id(access_token: str) -> str:
     """Return the user ID for the given session token."""
-    resp = httpx.post(
-        f"{OUTLINE_URL}/api/auth.info",
-        headers={
-            "Authorization": f"Bearer {access_token}",
-        },
-        timeout=30.0,
-    )
+    resp = _outline_api(access_token, "auth.info")
     resp.raise_for_status()
     return resp.json()["data"]["user"]["id"]
 
@@ -259,14 +280,7 @@ def _get_user_role(
     user_id: str,
 ) -> str | None:
     """Return the current role for *user_id* via ``users.info``."""
-    resp = httpx.post(
-        f"{OUTLINE_URL}/api/users.info",
-        headers={
-            "Authorization": f"Bearer {admin_token}",
-        },
-        json={"id": user_id},
-        timeout=30.0,
-    )
+    resp = _outline_api(admin_token, "users.info", {"id": user_id})
     if resp.status_code != 200:
         return None
     return resp.json()["data"].get("role")
@@ -283,13 +297,10 @@ def _set_user_role(
     handles profile fields like name/avatar).  Skips dependent
     tests if the endpoint is not available.
     """
-    resp = httpx.post(
-        f"{OUTLINE_URL}/api/users.update_role",
-        headers={
-            "Authorization": f"Bearer {admin_token}",
-        },
-        json={"id": user_id, "role": role},
-        timeout=30.0,
+    resp = _outline_api(
+        admin_token,
+        "users.update_role",
+        {"id": user_id, "role": role},
     )
     if resp.status_code != 200:
         pytest.skip(
@@ -451,7 +462,7 @@ def _viewer_credentials(outline_stack, _outline_credentials):
     user_id = _get_user_id(viewer_token)
 
     # Create all keys while user is still a member
-    full_key = _create_full_access_key(viewer_token)
+    full_key = _create_api_key(viewer_token)
     scoped_keys = _create_viewer_scoped_keys(viewer_token)
 
     # Now demote to viewer — keys remain valid
@@ -468,11 +479,7 @@ def _viewer_credentials(outline_stack, _outline_credentials):
     # Verify the viewer's API key still works after demotion.
     # auth.info requires no specific role, so this confirms
     # the key wasn't invalidated by the role change.
-    resp = httpx.post(
-        f"{OUTLINE_URL}/api/auth.info",
-        headers={"Authorization": f"Bearer {full_key}"},
-        timeout=30.0,
-    )
+    resp = _outline_api(full_key, "auth.info")
     if resp.status_code != 200:
         pytest.skip(
             f"Viewer API key invalid after role change "
@@ -492,42 +499,19 @@ def _create_viewer_scoped_keys(
     Returns a dict keyed by test name.
     """
     return {
-        "with_auth_info": _create_scoped_key(
+        "with_auth_info": _create_api_key(
             access_token,
             "e2e-viewer-with-auth-info",
             ["apiKeys.list", "auth.info", "documents:write"],
         ),
-        "without_auth_info": _create_scoped_key(
+        "without_auth_info": _create_api_key(
             access_token,
             "e2e-viewer-no-auth-info",
             ["apiKeys.list", "documents:write"],
         ),
     }
 
 
-def _create_scoped_key(
-    access_token: str,
-    name: str,
-    scope: list,
-) -> str:
-    """Create a scoped API key via the Outline admin API.
-
-    Like ``_create_api_key_with_scope`` in the test file but
-    usable from conftest (no pytest.skip on failure — raises
-    instead).
-    """
-    resp = httpx.post(
-        f"{OUTLINE_URL}/api/apiKeys.create",
-        headers={
-            "Authorization": f"Bearer {access_token}",
-        },
-        json={"name": name, "scope": scope},
-        timeout=30.0,
-    )
-    resp.raise_for_status()
-    return resp.json()["data"]["value"]
-
-
 @pytest.fixture(scope="session")
 def viewer_api_key(_viewer_credentials):
     """Full-access API key for the viewer user."""

tests/e2e/test_dynamic_tool_list.py

@@ -46,6 +46,7 @@
     build_tool_endpoint_map,
 )
 
+from .conftest import _create_api_key
 from .helpers import OUTLINE_URL
 
 HTTP_PORT = 3997
@@ -114,37 +115,6 @@ async def _list_tools_stdio(api_key: str) -> set[str]:
             return {t.name for t in result.tools}
 
 
-def _create_api_key_with_scope(
-    access_token: str,
-    name: str,
-    scope: list,
-) -> str:
-    """Create a scoped API key via the Outline admin API.
-
-    Uses the OIDC *access_token* (session token) to call
-    ``apiKeys.create``.  This endpoint requires a session
-    token -- API keys cannot create other API keys.
-
-    Raises ``pytest.skip`` if the Outline instance does not
-    support scoped API keys.
-
-    Returns the API key value string.
-    """
-    resp = httpx.post(
-        f"{OUTLINE_URL}/api/apiKeys.create",
-        headers={"Authorization": f"Bearer {access_token}"},
-        json={"name": name, "scope": scope},
-        timeout=30.0,
-    )
-    if resp.status_code != 200:
-        pytest.skip(
-            "Outline does not support scoped API keys "
-            f"(apiKeys.create returned {resp.status_code})"
-        )
-
-    return resp.json()["data"]["value"]
-
-
 def _assert_tools(
     actual: set[str],
     expected: set[str],
@@ -287,7 +257,7 @@ async def test_route_scoped_read_only(
     Scope: explicit ``namespace.method`` entries for every
     read-only endpoint in the TOOL_ENDPOINT_MAP.
     """
-    key = _create_api_key_with_scope(
+    key = _create_api_key(
         outline_access_token,
         "e2e-stdio-route-read-only",
         [
@@ -306,6 +276,7 @@ async def test_route_scoped_read_only(
             "documents.archived",
             "documents.deleted",
         ],
+        skip_on_error=True,
     )
 
     expected = {
@@ -351,7 +322,7 @@ async def test_namespace_read_scope(
     route scope.  ``collections.export_all`` likewise defaults
     to ``write`` and needs ``collections:write`` or a route scope.
     """
-    key = _create_api_key_with_scope(
+    key = _create_api_key(
         outline_access_token,
         "e2e-stdio-namespace-read",
         [
@@ -360,6 +331,7 @@ async def test_namespace_read_scope(
             "collections:read",
             "comments:read",
         ],
+        skip_on_error=True,
     )
 
     expected = {
@@ -393,10 +365,11 @@ async def test_namespace_write_documents_only(
     stay blocked (attachment tools now map to the ``attachments``
     namespace).
     """
-    key = _create_api_key_with_scope(
+    key = _create_api_key(
         outline_access_token,
         "e2e-stdio-namespace-write-docs",
         ["apiKeys.list", "documents:write"],
+        skip_on_error=True,
     )
 
     expected = {
@@ -442,7 +415,7 @@ async def test_mixed_namespace_and_route_scope(
     Attachment, export, comment, and other write tools are
     blocked.
     """
-    key = _create_api_key_with_scope(
+    key = _create_api_key(
         outline_access_token,
         "e2e-stdio-mixed-scope",
         [
@@ -451,6 +424,7 @@ async def test_mixed_namespace_and_route_scope(
             "collections.create",
             "collections.list",
         ],
+        skip_on_error=True,
     )
 
     expected = {
@@ -480,10 +454,11 @@ async def test_namespace_create_scope(
     matches methods whose ``methodToScope`` is ``create``
     (i.e. the ``create`` method itself).
     """
-    key = _create_api_key_with_scope(
+    key = _create_api_key(
         outline_access_token,
         "e2e-stdio-namespace-create-docs",
         ["apiKeys.list", "documents:create"],
+        skip_on_error=True,
     )
 
     expected = {
@@ -516,10 +491,11 @@ async def test_http_header_filters_tools(
     _assert_tools(admin_names, ALL_TOOLS, "http header admin key")
 
     # Scoped key via header -> subset
-    scoped_key = _create_api_key_with_scope(
+    scoped_key = _create_api_key(
         outline_access_token,
         "e2e-http-header-scoped",
         ["apiKeys.list", "documents:read"],
+        skip_on_error=True,
     )
     scoped_names = await _list_tools_http(scoped_key)
     assert "read_document" in scoped_names