test: add comprehensive E2E tests for Streamable HTTP & list change notifications

Add two layers of test coverage for the Streamable HTTP transport:

Rust integration tests (gateway_notifications.rs):
- Gateway capabilities advertisement (listChanged: true)
- Tools/prompts/resources list_changed forwarding to clients
- Server disconnect notification propagation
- Grant change notification delivery
- Content-based deduplication preventing spurious notifications
- Time-based throttling coalescing rapid notifications
- Server features refresh triggering notifications
- Full re-fetch cycle after notification

Tauri E2E tests (streamable-http.wdio.ts):
- Gateway Streamable HTTP endpoint serving
- Backend HTTP transport connection
- Stub server control endpoint verification
- All three notification types from backend
- Dynamic tool add/remove with notifications
- Rapid successive notification handling
- Server disable/re-enable reconnection cycle
- OAuth DCR registration and approval
- PKCE token exchange flow
- Authenticated MCP initialize with capabilities check
- Session management via Mcp-Session-Id header

Supporting infrastructure:
- Stub MCP server control endpoints for triggering notifications
- E2E helpers for OAuth flow and stub server control
- Enhanced transport-level notification tests (prompts, resources)

Signed-off-by: Myko <myko@mcpmux.com>

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
Signed-off-by: Mohammod Al Amin Ashik <maa.ashik00@gmail.com>

Author: its-mash

Cargo.lock

@@ -0,0 +1,172 @@
+/**
+ * MCP Client Helper for E2E Tests
+ *
+ * Provides functions to perform the full OAuth 2.1 + PKCE flow
+ * against the McpMux gateway programmatically and obtain an access token.
+ *
+ * This enables tests to connect an MCP client to the gateway
+ * without going through the browser-based consent flow.
+ */
+
+import crypto from 'node:crypto';
+
+const DEFAULT_GATEWAY_PORT = 45818;
+
+function gatewayUrl(path: string, port?: number): string {
+  return `http://localhost:${port ?? DEFAULT_GATEWAY_PORT}${path}`;
+}
+
+/** Generate PKCE code verifier + challenge pair (S256) */
+function generatePkce(): { codeVerifier: string; codeChallenge: string } {
+  const codeVerifier = crypto.randomBytes(32).toString('base64url');
+  const hash = crypto.createHash('sha256').update(codeVerifier).digest();
+  const codeChallenge = hash.toString('base64url');
+  return { codeVerifier, codeChallenge };
+}
+
+/**
+ * Register a new OAuth client via Dynamic Client Registration (DCR).
+ * Returns the client_id assigned by the gateway.
+ */
+export async function registerOAuthClient(
+  clientName: string,
+  redirectUri: string = 'http://localhost:0/callback',
+  port?: number,
+): Promise<string> {
+  const res = await fetch(gatewayUrl('/oauth/register', port), {
+    method: 'POST',
+    headers: { 'Content-Type': 'application/json' },
+    body: JSON.stringify({
+      client_name: clientName,
+      redirect_uris: [redirectUri],
+      grant_types: ['authorization_code'],
+      response_types: ['code'],
+      token_endpoint_auth_method: 'none',
+    }),
+  });
+
+  if (!res.ok) {
+    throw new Error(`DCR failed: ${res.status} ${await res.text()}`);
+  }
+
+  const data = (await res.json()) as { client_id: string };
+  return data.client_id;
+}
+
+/**
+ * Perform the full OAuth 2.1 + PKCE flow to obtain a JWT access token.
+ *
+ * Prerequisites:
+ * - Client must be registered (via registerOAuthClient or DCR)
+ * - Client must be approved (via Tauri API approveOAuthClient)
+ *
+ * Steps:
+ * 1. GET /oauth/authorize with PKCE challenge → extracts request_id from deep link
+ * 2. POST /oauth/consent/approve with request_id → extracts auth code from redirect
+ * 3. POST /oauth/token with auth code + code verifier → returns JWT
+ *
+ * @returns JWT access token string
+ */
+export async function obtainAccessToken(
+  clientId: string,
+  redirectUri: string = 'http://localhost:0/callback',
+  port?: number,
+): Promise<string> {
+  const { codeVerifier, codeChallenge } = generatePkce();
+  const state = crypto.randomUUID();
+
+  // Step 1: Authorization request
+  const authorizeUrl = new URL(gatewayUrl('/oauth/authorize', port));
+  authorizeUrl.searchParams.set('client_id', clientId);
+  authorizeUrl.searchParams.set('response_type', 'code');
+  authorizeUrl.searchParams.set('redirect_uri', redirectUri);
+  authorizeUrl.searchParams.set('code_challenge', codeChallenge);
+  authorizeUrl.searchParams.set('code_challenge_method', 'S256');
+  authorizeUrl.searchParams.set('state', state);
+
+  // The authorize endpoint returns an HTML page with a deep link.
+  // We need to extract the request_id from the HTML content.
+  const authorizeRes = await fetch(authorizeUrl.toString(), { redirect: 'manual' });
+  const html = await authorizeRes.text();
+
+  // Extract request_id from the deep link in the HTML
+  // Format: mcpmux://authorize?request_id=<id>
+  const requestIdMatch = html.match(/request_id=([^"&\s]+)/);
+  if (!requestIdMatch) {
+    throw new Error(`Could not extract request_id from authorize response. HTML: ${html.substring(0, 500)}`);
+  }
+  const requestId = decodeURIComponent(requestIdMatch[1]);
+
+  // Step 2: Consent approval
+  const consentRes = await fetch(gatewayUrl('/oauth/consent/approve', port), {
+    method: 'POST',
+    headers: { 'Content-Type': 'application/json' },
+    body: JSON.stringify({
+      request_id: requestId,
+      approved: true,
+    }),
+  });
+
+  if (!consentRes.ok) {
+    throw new Error(`Consent approval failed: ${consentRes.status} ${await consentRes.text()}`);
+  }
+
+  const consentData = (await consentRes.json()) as {
+    success: boolean;
+    redirect_url: string;
+    error?: string;
+  };
+
+  if (!consentData.success) {
+    throw new Error(`Consent not successful: ${consentData.error}`);
+  }
+
+  // Extract auth code from redirect URL
+  const redirectUrl = new URL(consentData.redirect_url);
+  const code = redirectUrl.searchParams.get('code');
+  if (!code) {
+    throw new Error(`No auth code in redirect: ${consentData.redirect_url}`);
+  }
+
+  // Step 3: Token exchange
+  const tokenRes = await fetch(gatewayUrl('/oauth/token', port), {
+    method: 'POST',
+    headers: { 'Content-Type': 'application/x-www-form-urlencoded' },
+    body: new URLSearchParams({
+      grant_type: 'authorization_code',
+      code,
+      client_id: clientId,
+      redirect_uri: redirectUri,
+      code_verifier: codeVerifier,
+    }).toString(),
+  });
+
+  if (!tokenRes.ok) {
+    throw new Error(`Token exchange failed: ${tokenRes.status} ${await tokenRes.text()}`);
+  }
+
+  const tokenData = (await tokenRes.json()) as {
+    access_token: string;
+    token_type: string;
+    expires_in?: number;
+  };
+
+  return tokenData.access_token;
+}
+
+/**
+ * Wait for the gateway to be ready by polling /health.
+ */
+export async function waitForGateway(port?: number, timeoutMs: number = 10000): Promise<void> {
+  const start = Date.now();
+  while (Date.now() - start < timeoutMs) {
+    try {
+      const res = await fetch(gatewayUrl('/health', port));
+      if (res.ok) return;
+    } catch {
+      // Not ready yet
+    }
+    await new Promise((r) => setTimeout(r, 200));
+  }
+  throw new Error(`Gateway not ready after ${timeoutMs}ms`);
+}

tests/e2e/helpers/mcp-client.ts

@@ -0,0 +1,66 @@
+/**
+ * Stub MCP Server Control Helper
+ *
+ * Provides functions to trigger list_changed notifications and manage
+ * dynamic tools on the stub MCP HTTP server via its control endpoints.
+ */
+
+const DEFAULT_STUB_PORT = 3457;
+
+function controlUrl(path: string, port?: number): string {
+  return `http://localhost:${port ?? DEFAULT_STUB_PORT}${path}`;
+}
+
+/** Trigger tools/list_changed notification on all connected sessions */
+export async function triggerToolsChanged(port?: number): Promise<{ ok: boolean; sessions_notified: number }> {
+  const res = await fetch(controlUrl('/control/notify-tools-changed', port), {
+    method: 'POST',
+    headers: { 'Content-Type': 'application/json' },
+  });
+  return res.json() as Promise<{ ok: boolean; sessions_notified: number }>;
+}
+
+/** Trigger prompts/list_changed notification on all connected sessions */
+export async function triggerPromptsChanged(port?: number): Promise<{ ok: boolean; sessions_notified: number }> {
+  const res = await fetch(controlUrl('/control/notify-prompts-changed', port), {
+    method: 'POST',
+    headers: { 'Content-Type': 'application/json' },
+  });
+  return res.json() as Promise<{ ok: boolean; sessions_notified: number }>;
+}
+
+/** Trigger resources/list_changed notification on all connected sessions */
+export async function triggerResourcesChanged(port?: number): Promise<{ ok: boolean; sessions_notified: number }> {
+  const res = await fetch(controlUrl('/control/notify-resources-changed', port), {
+    method: 'POST',
+    headers: { 'Content-Type': 'application/json' },
+  });
+  return res.json() as Promise<{ ok: boolean; sessions_notified: number }>;
+}
+
+/** Dynamically add a tool to the stub server and notify connected sessions */
+export async function addDynamicTool(
+  name: string,
+  description?: string,
+  port?: number,
+): Promise<{ ok: boolean; tool: string; sessions_updated: number }> {
+  const res = await fetch(controlUrl('/control/add-tool', port), {
+    method: 'POST',
+    headers: { 'Content-Type': 'application/json' },
+    body: JSON.stringify({ name, description }),
+  });
+  return res.json() as Promise<{ ok: boolean; tool: string; sessions_updated: number }>;
+}
+
+/** Dynamically remove a tool from the stub server and notify connected sessions */
+export async function removeDynamicTool(
+  name: string,
+  port?: number,
+): Promise<{ ok: boolean; tool: string; sessions_notified: number }> {
+  const res = await fetch(controlUrl('/control/remove-tool', port), {
+    method: 'POST',
+    headers: { 'Content-Type': 'application/json' },
+    body: JSON.stringify({ name }),
+  });
+  return res.json() as Promise<{ ok: boolean; tool: string; sessions_notified: number }>;
+}