diff --git a/Cargo.lock b/Cargo.lock index 5ec2daff..0a2824f2 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -607,14 +607,12 @@ dependencies = [ "contextforge-data-plane-apis", "contextforge-data-plane-cpex", "cpex", - "cpex-secrets-detection", "futures", "http", "hyper", "hyper-util", "jsonwebtoken", "lru_time_cache", - "openport", "opentelemetry", "opentelemetry_sdk", "redis", diff --git a/_context/wiki/index.md b/_context/wiki/index.md index abbed897..d3f8b6a3 100644 --- a/_context/wiki/index.md +++ b/_context/wiki/index.md @@ -25,7 +25,7 @@ then follow only the links that are relevant. - **Repo**: `contextforge-data-plane` — the Rust ContextForge external dataplane. - **Core invariant**: the ContextForge external dataplane is pure routing logic. No IAM, UI, or metrics storage. -- **Protocol target**: the ContextForge external dataplane supports MCP `2026-07-28` and `2025-11-25` over Streamable HTTP. Both use stateless request handling; `initialize` remains supported but does not establish external-dataplane session state. Legacy SSE paths are being removed from the external dataplane. +- **Protocol target**: new external-dataplane behavior targets MCP `2026-07-28` over Streamable HTTP with `server/discover` and per-request client metadata. The remaining `2025-11-25`/`initialize` paths are temporary compatibility coverage, not an expansion surface. Legacy SSE is outside the external dataplane. - **Status convention**: project, architecture, routing, and operations pages describe the current implementation. The page under **Upcoming** describes the tentative ContextForge 2.0 target and migration roadmap. diff --git a/_context/wiki/routing.md b/_context/wiki/routing.md index 9f300386..f62579fd 100644 --- a/_context/wiki/routing.md +++ b/_context/wiki/routing.md @@ -12,7 +12,7 @@ The control plane builds and publishes the routing tables to Redis; the dataplan ## Routing table shape -```rust +```text VirtualHost { backends: HashMap, tools: HashMap, resources: HashMap, diff --git a/_context/wiki/testing.md b/_context/wiki/testing.md index 85177610..d3cc32bb 100644 --- a/_context/wiki/testing.md +++ b/_context/wiki/testing.md @@ -13,26 +13,50 @@ CI runs these on every change; run them locally before pushing: ```bash cargo fmt --all --check -cargo clippy --locked --workspace --all-targets -- -D warnings -cargo nextest run --locked --workspace +cargo clippy --locked --workspace --all-targets --all-features -- -D warnings +cargo nextest run --locked --workspace --all-features +cargo shear --check-test-targets --deny-warnings --locked ``` Use `cargo test` when nextest is unavailable. For wiki changes, also run `mdbook build _context/wiki` and `mdbook test _context/wiki`. -Protocol-sensitive tests and fixtures must cover MCP `2026-07-28` and `2025-11-25` in all four incoming-client/selected-backend combinations. The same-version paths are supported directly; the two cross-version paths are best effort and tests must cover both successful adaptation and explicit failure for semantics that cannot be translated without state. Every case must prove request independence: no required `Mcp-Session-Id`, session affinity, or retained backend transport. Keep `2026-07-28` coverage for `server/discover` and required per-request client metadata, and retain `initialize` coverage as a stateless compatibility request. SSE remains outside the external-dataplane contract. +New protocol-sensitive tests target MCP `2026-07-28`, connect through +`server/discover`, and send the required per-request client metadata. A small +`compatibility` module retains the active `2025-11-25`/`initialize` cases until +that production compatibility surface is removed in a dedicated change; do not +add new behavior to that lane. Every case must remain request-independent, with +no required `Mcp-Session-Id`, session affinity, or retained backend transport. +SSE remains outside the external-dataplane contract. ## In-Repo Integration Tests -`crates/contextforge-data-plane-lib/tests/` exercises the gateway against in-process mock MCP backends (shared helpers live in `tests/support/`): +`crates/contextforge-data-plane-lib/tests/gateway.rs` is the single library +integration target. It exercises the public gateway API against in-process MCP +backends without recompiling a shared support tree for every feature file. -| Test file | Covers | +| Area | Covers | | --- | --- | -| `gateway_list_tools.rs` | List fanout, prefixing, and merged output. | -| `gateway_prompts.rs` | Prompt listing and prefixed `get_prompt` routing. | -| `gateway_resource_templates.rs` | Template fanout with prefixed names and URI templates, plus `read_resource` round-trips. | -| `gateway_plugins.rs` | Request-scoped parameter-header validation/forwarding, CPEX pre/post tool hooks around `call_tool` and stream events, and prompt hooks around `get_prompt`. | +| `gateway/{tools,prompts,resources,subscriptions}.rs` | Active routed operations and exact routing failures. | +| `gateway/plugins.rs` | Gateway-owned CPEX ordering, mutation, denial, progress, and prompt seams using deterministic recording plugins. Concrete plugin behavior stays in each plugin crate. | +| `gateway/harness/` | Authentication, modern and compatibility clients, in-memory configuration, concrete mock backends, and owned server fixtures. | +| `gateway/future_contracts/` | Deferred fanout, pagination, TLS, completions, subscriptions, and cancellation contracts. | -These run in `cargo nextest run` with no Docker dependencies. +`TestServer` binds `127.0.0.1:0` before spawning, uses cooperative +cancellation, and has a `Drop` fallback. `GatewayFixture` owns the gateway and +all backend servers. Tests should request the minimum topology: one virtual host +and one backend by default, with extra backends declared explicitly by the case. + +The workspace currently keeps 13 ignored tests: 11 library future contracts +and two real-process Redis/binary E2E tests. Ignored tests are not dead tests: +keep them compiling, keep their intended assertions, give each a concrete +blocker reason, and list them with: + +```bash +cargo nextest list --locked --workspace --all-features --run-ignored only +``` + +The two binary E2E tests and `tests/conformance/` remain separate infrastructure +boundaries. Active in-process tests run with no Docker or Redis dependency. Parameter-header integration tests verify that calls without a published tool schema skip local `Mcp-Param-*` validation and still reach the backend. Unit and diff --git a/crates/contextforge-data-plane-lib/Cargo.toml b/crates/contextforge-data-plane-lib/Cargo.toml index c601175c..7c2df933 100644 --- a/crates/contextforge-data-plane-lib/Cargo.toml +++ b/crates/contextforge-data-plane-lib/Cargo.toml @@ -59,8 +59,6 @@ with_tools = [] [dev-dependencies] opentelemetry_sdk.workspace = true cpex.workspace = true -openport.workspace = true -cpex-secrets-detection.workspace = true test-log = "0.2.20" axum-server = { version = "0.8.0", features = ["tls-rustls"] } diff --git a/crates/contextforge-data-plane-lib/src/lib.rs b/crates/contextforge-data-plane-lib/src/lib.rs index 48397b18..e268ad11 100644 --- a/crates/contextforge-data-plane-lib/src/lib.rs +++ b/crates/contextforge-data-plane-lib/src/lib.rs @@ -70,7 +70,7 @@ pub struct Gateway { impl Gateway { pub async fn run_gateway(self) -> Result<()> { let config = self.config.clone(); - let app = self.build_app().await?; + let app = self.into_router().await?; let mut handlers = vec![]; @@ -98,7 +98,11 @@ impl Gateway { Ok(()) } - async fn build_app(self) -> Result { + /// Builds the complete gateway application without binding a network listener. + /// + /// This supports embedding the dataplane in an existing Axum server and lets + /// callers bind listeners before starting the service. + pub async fn into_router(self) -> Result { let Gateway { config, session_manager, user_config_store_type, plugin_runtime, authorization_service } = self; let user_config_store = match user_config_store_type { UserConfigStoreType::Redis => Arc::new(get_config_store(&config).await?), @@ -212,7 +216,7 @@ mod tests { .with_session_manager(Arc::new(LocalSessionManager::default())) .with_user_config_store_type(UserConfigStoreType::Test(Arc::new(UnusedConfigStore))) .build() - .build_app() + .into_router() .await .expect("Expecting this to work"); let request = Request::builder() diff --git a/crates/contextforge-data-plane-lib/tests/gateway.rs b/crates/contextforge-data-plane-lib/tests/gateway.rs new file mode 100644 index 00000000..0abd8d15 --- /dev/null +++ b/crates/contextforge-data-plane-lib/tests/gateway.rs @@ -0,0 +1,19 @@ +#[path = "gateway/harness/mod.rs"] +mod harness; + +#[path = "gateway/compatibility.rs"] +mod compatibility; +#[path = "gateway/completions.rs"] +mod completions; +#[path = "gateway/future_contracts/mod.rs"] +mod future_contracts; +#[path = "gateway/plugins.rs"] +mod plugins; +#[path = "gateway/prompts.rs"] +mod prompts; +#[path = "gateway/resources.rs"] +mod resources; +#[path = "gateway/subscriptions.rs"] +mod subscriptions; +#[path = "gateway/tools.rs"] +mod tools; diff --git a/crates/contextforge-data-plane-lib/tests/gateway/compatibility.rs b/crates/contextforge-data-plane-lib/tests/gateway/compatibility.rs new file mode 100644 index 00000000..382220fd --- /dev/null +++ b/crates/contextforge-data-plane-lib/tests/gateway/compatibility.rs @@ -0,0 +1,81 @@ +use std::sync::Arc; + +use contextforge_data_plane_cpex::CpexRuntimeRegistry; +use contextforge_data_plane_lib::Result; +use rmcp::model::{CallToolRequestParams, ErrorCode, ProtocolVersion, ReadResourceRequestParams, ResourceContents}; + +use crate::harness::{ + TEST_USER_ID, connect_client_with_protocol, connect_modern_client, create_client, modern_client_info, + start_counter_gateway, start_gateway, start_legacy_counter_gateway, +}; + +const DECREMENT_TOOL: &str = "00000000-0000-0000-0000-000000000001-decrement"; +const MEMO_RESOURCE: &str = "00000000-0000-0000-0000-000000000001-memo://insights"; +const EXPECTED_MEMO: &str = "Business Intelligence Memo\n\nAnalysis has revealed 5 key insights ..."; + +#[tokio::test] +async fn plaintext_call_prefixed_backend_tools_modern_legacy() -> Result<()> { + let fixture = start_legacy_counter_gateway(TEST_USER_ID).await?; + assert_decrement(fixture.gateway_url, ProtocolVersion::V_2026_07_28).await +} + +#[tokio::test] +async fn plaintext_call_prefixed_backend_tools_legacy_modern() -> Result<()> { + let fixture = start_counter_gateway(TEST_USER_ID).await?; + assert_decrement(fixture.gateway_url, ProtocolVersion::V_2025_11_25).await +} + +#[tokio::test] +async fn plaintext_call_prefixed_read_resources_modern_legacy() -> Result<()> { + let fixture = start_legacy_counter_gateway(TEST_USER_ID).await?; + assert_memo_read(fixture.gateway_url, ProtocolVersion::V_2026_07_28).await +} + +#[tokio::test] +async fn plaintext_call_prefixed_read_resources_legacy_modern() -> Result<()> { + let fixture = start_counter_gateway(TEST_USER_ID).await?; + assert_memo_read(fixture.gateway_url, ProtocolVersion::V_2025_11_25).await +} + +#[tokio::test] +async fn legacy_tool_call_without_published_schema_reaches_backend() { + let gateway = start_gateway(TEST_USER_ID, false, Arc::new(CpexRuntimeRegistry::default())).await; + let service = gateway.connect_legacy(TEST_USER_ID).await; + let error = service.call_tool(CallToolRequestParams::new("missing_schema_tool")).await.unwrap_err(); + let rmcp::service::ServiceError::McpError(error) = error else { + panic!("expected backend MCP error, got {error:?}"); + }; + assert_eq!(ErrorCode::METHOD_NOT_FOUND, error.code); + let backend_calls = gateway.backend_state.calls.lock().expect("backend calls lock poisoned"); + assert_eq!("missing_schema_tool", backend_calls[0].tool_name); +} + +async fn assert_decrement(gateway_url: String, client_protocol: ProtocolVersion) -> Result<()> { + let service = connect_for_protocol(gateway_url, client_protocol).await?; + let result = service.call_tool(CallToolRequestParams::new(DECREMENT_TOOL)).await?; + let text = result.content.first().and_then(|content| content.as_text()).expect("text tool result"); + assert_eq!("-1", text.text); + Ok(()) +} + +async fn assert_memo_read(gateway_url: String, client_protocol: ProtocolVersion) -> Result<()> { + let service = connect_for_protocol(gateway_url, client_protocol).await?; + let response = service.read_resource(ReadResourceRequestParams::new(MEMO_RESOURCE)).await?; + let Some(ResourceContents::TextResourceContents { text, .. }) = response.contents.first() else { + panic!("expected one text resource, got {:?}", response.contents); + }; + assert_eq!(EXPECTED_MEMO, text); + Ok(()) +} + +async fn connect_for_protocol( + gateway_url: String, + client_protocol: ProtocolVersion, +) -> Result> { + let client = create_client(TEST_USER_ID); + if client_protocol == ProtocolVersion::V_2026_07_28 { + Ok(connect_modern_client(&gateway_url, client, modern_client_info()).await) + } else { + connect_client_with_protocol(gateway_url, client, client_protocol).await + } +} diff --git a/crates/contextforge-data-plane-lib/tests/gateway/completions.rs b/crates/contextforge-data-plane-lib/tests/gateway/completions.rs new file mode 100644 index 00000000..8840a73b --- /dev/null +++ b/crates/contextforge-data-plane-lib/tests/gateway/completions.rs @@ -0,0 +1,15 @@ +use contextforge_data_plane_lib::Result; + +use crate::harness::{TEST_USER_ID, connect_modern_client, create_client, modern_client_info, start_counter_gateway}; + +#[tokio::test] +async fn plaintext_complete_for_unrouted_reference_errors() -> Result<()> { + let fixture = start_counter_gateway(TEST_USER_ID).await?; + let service = connect_modern_client(&fixture.gateway_url, create_client(TEST_USER_ID), modern_client_info()).await; + + service + .complete_prompt_simple("unrouted_prompt", "message", "h") + .await + .expect_err("an unrouted completion reference must fail"); + Ok(()) +} diff --git a/crates/contextforge-data-plane-lib/tests/gateway/future_contracts/completions.rs b/crates/contextforge-data-plane-lib/tests/gateway/future_contracts/completions.rs new file mode 100644 index 00000000..d5eefa7d --- /dev/null +++ b/crates/contextforge-data-plane-lib/tests/gateway/future_contracts/completions.rs @@ -0,0 +1,41 @@ +use contextforge_data_plane_lib::Result; + +use crate::harness::{TEST_USER_ID, connect_modern_client, create_client, modern_client_info, start_counter_gateway}; + +#[tokio::test] +#[ignore = "blocked on federated prompt-completion capability and routing"] +async fn plaintext_completes_prompt_argument_through_prefixed_backend() -> Result<()> { + let fixture = start_counter_gateway(TEST_USER_ID).await?; + let service = connect_modern_client(&fixture.gateway_url, create_client(TEST_USER_ID), modern_client_info()).await; + + assert!( + service.peer_info().and_then(|info| info.capabilities.completions.clone()).is_some(), + "gateway must advertise completions before clients can call completion/complete" + ); + let prompts = service.list_prompts(None).await?; + let prompt_name = prompts + .prompts + .iter() + .find(|prompt| prompt.name.ends_with("-example_prompt")) + .map(|prompt| prompt.name.clone()) + .ok_or("expected a federated example_prompt")?; + let values = service.complete_prompt_simple(prompt_name, "message", "h").await?; + assert!(values.contains(&"hello".to_owned()), "expected backend prompt completions, got {values:?}"); + Ok(()) +} + +#[tokio::test] +#[ignore = "blocked on federated resource-completion capability and routing"] +async fn plaintext_completes_resource_argument_through_prefixed_backend() -> Result<()> { + let fixture = start_counter_gateway(TEST_USER_ID).await?; + let service = connect_modern_client(&fixture.gateway_url, create_client(TEST_USER_ID), modern_client_info()).await; + + let resources = service.list_resources(None).await?; + let uri = resources.resources.first().ok_or("expected at least one federated resource")?.uri.clone(); + let values = service.complete_resource_simple(uri, "path", "").await?; + assert!( + values.first().is_some_and(|value| !value.starts_with("backend-")), + "expected a stripped backend URI, got {values:?}" + ); + Ok(()) +} diff --git a/crates/contextforge-data-plane-lib/tests/gateway/future_contracts/mod.rs b/crates/contextforge-data-plane-lib/tests/gateway/future_contracts/mod.rs new file mode 100644 index 00000000..2bc95f4c --- /dev/null +++ b/crates/contextforge-data-plane-lib/tests/gateway/future_contracts/mod.rs @@ -0,0 +1,7 @@ +mod completions; +mod pagination; +mod plugins; +mod prompts; +mod resource_templates; +mod subscriptions; +mod transport; diff --git a/crates/contextforge-data-plane-lib/tests/gateway/future_contracts/pagination.rs b/crates/contextforge-data-plane-lib/tests/gateway/future_contracts/pagination.rs new file mode 100644 index 00000000..bab5244c --- /dev/null +++ b/crates/contextforge-data-plane-lib/tests/gateway/future_contracts/pagination.rs @@ -0,0 +1,129 @@ +use std::collections::HashMap; + +use contextforge_data_plane_apis::{ + User, + user_store::{BackendMCPGateway, UserConfig, VirtualHost}, +}; +use contextforge_data_plane_lib::{Config, Result, UpstreamConnectionMode, UserConfigStore}; +use rmcp::{ + model::PaginatedRequestParams, + transport::{ + StreamableHttpServerConfig, StreamableHttpService, streamable_http_server::session::local::LocalSessionManager, + }, +}; + +use crate::harness::{ + GatewayFixture, GatewayTestConfig, MemoryUserConfigStore, TEST_USER_ID, TestServer, connect_modern_client, + create_client, create_default_config, modern_client_info, paginating_mock, +}; + +const VIRTUAL_HOST_ID: &str = "33333333-3333-3333-3333-333333333333"; + +async fn start_paginating_gateway(backend_count: usize) -> Result { + let mut backend_servers = Vec::with_capacity(backend_count); + let mut backends = HashMap::with_capacity(backend_count); + + for backend_number in 1..=backend_count { + let service = StreamableHttpService::new( + || Ok(paginating_mock::PaginatingServer), + LocalSessionManager::default().into(), + StreamableHttpServerConfig::default(), + ); + let server = TestServer::start_http(axum::Router::new().route_service("/mcp", service)).await?; + let backend_id = format!("00000000-0000-0000-0000-{backend_number:012}"); + backends.insert( + backend_id, + BackendMCPGateway { + name: format!("paginating-backend-{backend_number}"), + url: server.url("/mcp").parse().expect("backend URL"), + mcp_protocol_version: rmcp::model::ProtocolVersion::V_2026_07_28, + passthrough_headers: Vec::new(), + add_headers: HashMap::new(), + remove_headers: Vec::new(), + tool_schemas: HashMap::new(), + completion: HashMap::new(), + }, + ); + backend_servers.push(server); + } + + let store = MemoryUserConfigStore::default(); + store + .set_config( + &User::new(TEST_USER_ID), + &UserConfig { + virtual_hosts: HashMap::from([( + VIRTUAL_HOST_ID.to_owned(), + VirtualHost { + backends, + tools: HashMap::new(), + resources: HashMap::new(), + resource_templates: HashMap::new(), + prompts: HashMap::new(), + }, + )]), + }, + ) + .await?; + + GatewayFixture::start(GatewayTestConfig { + config: Config { + upstream_connection_mode: Some(UpstreamConnectionMode::PlainTextOrTls), + ..create_default_config() + }, + user_store: store, + user_id: TEST_USER_ID.to_owned(), + virtual_host_id: VIRTUAL_HOST_ID.to_owned(), + backends: backend_servers, + plugin_runtime: None, + }) + .await +} + +#[tokio::test] +#[ignore = "blocked on federated list-tools pagination support"] +async fn single_backend_pagination_all_tools_reachable() -> Result<()> { + let fixture = start_paginating_gateway(1).await?; + let service = + connect_modern_client(&fixture.gateway_url(), create_client(TEST_USER_ID), modern_client_info()).await; + + let page1 = service.list_tools(None).await.expect("page 1"); + let page1_names: Vec<&str> = page1.tools.iter().map(|tool| tool.name.as_ref()).collect(); + assert!(page1.next_cursor.is_some(), "page 1 must carry a next_cursor"); + assert_eq!(page1_names, ["tool_alpha", "tool_beta"]); + + let cursor = page1.next_cursor.map(|cursor| PaginatedRequestParams::default().with_cursor(Some(cursor))); + let page2 = service.list_tools(cursor).await.expect("page 2"); + let page2_names: Vec<&str> = page2.tools.iter().map(|tool| tool.name.as_ref()).collect(); + assert!(page2.next_cursor.is_none(), "page 2 must be the final page"); + assert_eq!(page2_names, ["tool_gamma"]); + + let mut all_names = page1_names; + all_names.extend_from_slice(&page2_names); + all_names.sort_unstable(); + assert_eq!(all_names, paginating_mock::PaginatingServer::all_tool_names()); + Ok(()) +} + +#[tokio::test] +#[ignore = "blocked on federated list-tools pagination support"] +async fn multi_backend_exhausted_backend_not_requeried() -> Result<()> { + let fixture = start_paginating_gateway(2).await?; + let service = + connect_modern_client(&fixture.gateway_url(), create_client(TEST_USER_ID), modern_client_info()).await; + + let page1 = service.list_tools(None).await.expect("page 1"); + assert!(page1.next_cursor.is_some(), "page 1 must carry a next_cursor"); + assert_eq!(page1.tools.len(), 4, "page 1 should have two tools from each backend"); + + let cursor = page1.next_cursor.map(|cursor| PaginatedRequestParams::default().with_cursor(Some(cursor))); + let page2 = service.list_tools(cursor).await.expect("page 2"); + assert!(page2.next_cursor.is_none(), "page 2 must be the final page"); + assert_eq!(page2.tools.len(), 2, "page 2 should have one tool from each backend"); + + let mut all_names: Vec<_> = page1.tools.iter().chain(page2.tools.iter()).map(|tool| tool.name.clone()).collect(); + all_names.sort_unstable(); + all_names.dedup(); + assert_eq!(all_names.len(), page1.tools.len() + page2.tools.len(), "no duplicate tools across pages"); + Ok(()) +} diff --git a/crates/contextforge-data-plane-lib/tests/gateway/future_contracts/plugins.rs b/crates/contextforge-data-plane-lib/tests/gateway/future_contracts/plugins.rs new file mode 100644 index 00000000..94139769 --- /dev/null +++ b/crates/contextforge-data-plane-lib/tests/gateway/future_contracts/plugins.rs @@ -0,0 +1,44 @@ +use std::sync::{Arc, Mutex}; + +use contextforge_data_plane_cpex::CpexRuntimeRegistry; +use rmcp::{ + model::{CallToolRequestParams, ClientRequest, Request}, + service::PeerRequestOptions, +}; + +use crate::harness::{TEST_USER_ID, connect_modern_client, create_client, modern_client_info, start_gateway}; + +#[tokio::test] +#[ignore = "blocked on downstream cancellation relay for the 2026-07-28 lifecycle"] +async fn downstream_cancellation_is_relayed_to_backend() { + tokio::time::timeout(std::time::Duration::from_secs(3), assert_downstream_cancellation()) + .await + .expect("downstream cancellation relay completes within three seconds"); +} + +async fn assert_downstream_cancellation() { + let gateway = start_gateway(TEST_USER_ID, true, Arc::new(CpexRuntimeRegistry::default())).await; + let service = connect_modern_client(gateway.gateway_url(), create_client(TEST_USER_ID), modern_client_info()).await; + + let handle = service + .send_cancellable_request( + ClientRequest::CallToolRequest(Request::new(CallToolRequestParams::new("wait_for_cancellation"))), + PeerRequestOptions::no_options(), + ) + .await + .expect("wait_for_cancellation request is sent"); + wait_for_event_count(&gateway.backend_state.calls, 1).await; + + handle.cancel(Some("client gave up".to_owned())).await.expect("cancellation is sent"); + wait_for_event_count(&gateway.backend_state.cancellations, 1).await; +} + +async fn wait_for_event_count(events: &Mutex>, expected: usize) { + for _ in 0..50 { + if events.lock().expect("events lock poisoned").len() >= expected { + return; + } + tokio::time::sleep(std::time::Duration::from_millis(20)).await; + } + panic!("expected {expected} recorded events"); +} diff --git a/crates/contextforge-data-plane-lib/tests/gateway/future_contracts/prompts.rs b/crates/contextforge-data-plane-lib/tests/gateway/future_contracts/prompts.rs new file mode 100644 index 00000000..afb6edfc --- /dev/null +++ b/crates/contextforge-data-plane-lib/tests/gateway/future_contracts/prompts.rs @@ -0,0 +1,18 @@ +use contextforge_data_plane_lib::Result; + +use crate::harness::{TEST_USER_ID, connect_modern_client, create_client, modern_client_info, start_counter_gateway}; + +const EXPECTED_PROMPTS: &[&str] = + &["00000000-0000-0000-0000-000000000001-counter_analysis", "00000000-0000-0000-0000-000000000001-example_prompt"]; + +#[tokio::test] +#[ignore = "blocked on federated list-prompts support"] +async fn plaintext_lists_prefixed_backend_prompts() -> Result<()> { + let fixture = start_counter_gateway(TEST_USER_ID).await?; + let service = connect_modern_client(&fixture.gateway_url, create_client(TEST_USER_ID), modern_client_info()).await; + let response = service.list_prompts(None).await?; + let mut names: Vec<_> = response.prompts.iter().map(|prompt| prompt.name.as_str()).collect(); + names.sort_unstable(); + assert_eq!(EXPECTED_PROMPTS, names); + Ok(()) +} diff --git a/crates/contextforge-data-plane-lib/tests/gateway/future_contracts/resource_templates.rs b/crates/contextforge-data-plane-lib/tests/gateway/future_contracts/resource_templates.rs new file mode 100644 index 00000000..8c1cc1d9 --- /dev/null +++ b/crates/contextforge-data-plane-lib/tests/gateway/future_contracts/resource_templates.rs @@ -0,0 +1,94 @@ +use contextforge_data_plane_lib::Result; +use rmcp::model::{ReadResourceRequestParams, ResourceContents}; +use tracing::{info, warn}; + +use crate::harness::{TEST_USER_ID, connect_modern_client, create_client, modern_client_info, start_counter_gateway}; + +const EXPECTED_TEMPLATE_NAMES: &[&str] = + &["00000000-0000-0000-0000-000000000001-filesystem", "00000000-0000-0000-0000-000000000001-memo"]; +const EXPECTED_TEMPLATE_URIS: &[&str] = + &["00000000-0000-0000-0000-000000000001-memo://{id}", "00000000-0000-0000-0000-000000000001-str:////{path}"]; + +#[tokio::test(flavor = "multi_thread", worker_threads = 1)] +#[test_log::test] +#[ignore = "blocked on federated list-resource-templates support"] +async fn plaintext_lists_prefixed_backend_resource_templates() -> Result<()> { + let fixture = start_counter_gateway(TEST_USER_ID).await?; + assert_list_resource_templates( + fixture.gateway_url.clone(), + create_client(TEST_USER_ID), + EXPECTED_TEMPLATE_NAMES, + EXPECTED_TEMPLATE_URIS, + ) + .await +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 1)] +#[test_log::test] +#[ignore = "blocked on federated resource-template discovery and routing"] +async fn plaintext_reads_resource_from_prefixed_template() -> Result<()> { + let fixture = start_counter_gateway(TEST_USER_ID).await?; + assert_read_resource_from_template(fixture.gateway_url.clone(), create_client(TEST_USER_ID)).await +} + +async fn assert_list_resource_templates( + gateway_url: String, + client: reqwest::Client, + expected_template_names: &[&str], + expected_template_uris: &[&str], +) -> Result<()> { + info!("Sending request to {gateway_url}"); + + let running_service = connect_modern_client(&gateway_url, client, modern_client_info()).await; + + let list_templates = running_service.list_resource_templates(None).await; + let Ok(list_templates) = list_templates else { + let msg = format!("List resource templates returned error {list_templates:?}"); + warn!(msg); + return Err(msg.into()); + }; + + let mut names: Vec<&str> = + list_templates.resource_templates.iter().map(|template| template.name.as_str()).collect(); + names.sort_unstable(); + + if expected_template_names != names { + warn!("Actual {names:#?} Expected {expected_template_names:#?}"); + return Err("Expected resource template names don't match actual".into()); + } + + let mut uris: Vec<&str> = + list_templates.resource_templates.iter().map(|template| template.uri_template.as_str()).collect(); + uris.sort_unstable(); + + if expected_template_uris != uris { + warn!("Actual {uris:#?} Expected {expected_template_uris:#?}"); + return Err("Expected resource template uris don't match actual".into()); + } + + Ok(()) +} + +async fn assert_read_resource_from_template(gateway_url: String, client: reqwest::Client) -> Result<()> { + let running_service = connect_modern_client(&gateway_url, client, modern_client_info()).await; + + let list_templates = running_service.list_resource_templates(None).await?; + let Some(memo_template) = list_templates.resource_templates.iter().find(|t| t.uri_template.contains("memo://")) + else { + return Err("Expected a memo resource template".into()); + }; + + // Expand the namespaced template the way a client would, then read it back through the gateway. + let uri = memo_template.uri_template.replace("{id}", "insights"); + let result = running_service.read_resource(ReadResourceRequestParams::new(uri)).await?; + + let Some(ResourceContents::TextResourceContents { text, .. }) = result.contents.first() else { + return Err("Expected text resource contents".into()); + }; + + if !text.contains("Business Intelligence Memo") { + return Err(format!("Expected routed resource to include memo content, got: {text}").into()); + } + + Ok(()) +} diff --git a/crates/contextforge-data-plane-lib/tests/gateway/future_contracts/subscriptions.rs b/crates/contextforge-data-plane-lib/tests/gateway/future_contracts/subscriptions.rs new file mode 100644 index 00000000..84154ed8 --- /dev/null +++ b/crates/contextforge-data-plane-lib/tests/gateway/future_contracts/subscriptions.rs @@ -0,0 +1,123 @@ +use std::{ + collections::HashSet, + sync::{Arc, Mutex}, + time::Instant, +}; + +use contextforge_data_plane_lib::Result; +use futures::future::try_join_all; +use rmcp::{ + ClientHandler, + model::{ + ClientCapabilities, Implementation, InitializeRequestParams, ResourceUpdatedNotificationParam, + SubscribeRequestParams, UnsubscribeRequestParams, + }, + service::{NotificationContext, RoleClient}, +}; + +use crate::harness::{ + CLIENT_CONNECT_TIMEOUT, TEST_POLL_INTERVAL, TEST_USER_ID, connect_modern_client, create_client, + mock_counter::{KNOWN_RESOURCE_URIS, RESOURCE_UPDATE_NOTIFY_INTERVAL}, + start_counter_gateway_with_backends, +}; + +const MIN_UPDATES_PER_BACKEND: usize = 4; + +type Recorded = Arc>>; + +#[derive(Clone, Default)] +struct RecordingClient { + resource_updates: Recorded, +} + +impl ClientHandler for RecordingClient { + fn get_info(&self) -> InitializeRequestParams { + InitializeRequestParams::new( + ClientCapabilities::default(), + Implementation::new("resource-update-recording-test-client", "0.1.0"), + ) + } + + async fn on_resource_updated( + &self, + params: ResourceUpdatedNotificationParam, + _context: NotificationContext, + ) { + self.resource_updates.lock().expect("resource update lock poisoned").push(params); + } +} + +#[tokio::test] +#[ignore = "blocked on federated resource-subscription support"] +#[expect(deprecated, reason = "subscription APIs remain deferred to the control plane")] +async fn plaintext_subscribes_and_unsubscribes_through_two_prefixed_backends() -> Result<()> { + let fixture = start_counter_gateway_with_backends(TEST_USER_ID, 2).await?; + let recording_client = RecordingClient::default(); + let resource_updates = Arc::clone(&recording_client.resource_updates); + let service = connect_modern_client(&fixture.gateway_url, create_client(TEST_USER_ID), recording_client).await; + + let resources = service.list_resources(None).await?; + let mut selected_backends = HashSet::new(); + let mut selected_uris = Vec::new(); + for resource in resources.resources { + let backend_name = mock_backend_name(&resource.uri) + .ok_or_else(|| format!("expected mock backend-prefixed URI, got {}", resource.uri))?; + if selected_backends.insert(backend_name) { + selected_uris.push(resource.uri); + } + if selected_uris.len() == 2 { + break; + } + } + assert_eq!(2, selected_uris.len(), "expected resources from two backends"); + + try_join_all(selected_uris.iter().map(|uri| service.subscribe(SubscribeRequestParams::new(uri.clone())))).await?; + wait_for_resource_updates(&resource_updates, &selected_uris, MIN_UPDATES_PER_BACKEND).await?; + for uri in selected_uris { + service.unsubscribe(UnsubscribeRequestParams::new(uri)).await?; + } + assert_no_more_resource_updates(&resource_updates).await +} + +async fn assert_no_more_resource_updates( + resource_updates: &Mutex>, +) -> Result<()> { + tokio::time::sleep(RESOURCE_UPDATE_NOTIFY_INTERVAL * 5).await; + let count_after_drain = resource_updates.lock().expect("resource update lock poisoned").len(); + tokio::time::sleep(RESOURCE_UPDATE_NOTIFY_INTERVAL * 10).await; + let count_after_quiet = resource_updates.lock().expect("resource update lock poisoned").len(); + assert_eq!(count_after_drain, count_after_quiet, "updates continued after unsubscribe"); + Ok(()) +} + +async fn wait_for_resource_updates( + resource_updates: &Mutex>, + expected_uris: &[String], + expected_count_per_uri: usize, +) -> Result<()> { + let deadline = Instant::now() + CLIENT_CONNECT_TIMEOUT; + loop { + let counts = { + let updates = resource_updates.lock().expect("resource update lock poisoned"); + expected_uris + .iter() + .map(|uri| updates.iter().filter(|update| update.uri == *uri).count()) + .collect::>() + }; + if counts.iter().all(|count| *count >= expected_count_per_uri) { + return Ok(()); + } + if Instant::now() >= deadline { + return Err(format!("expected {expected_count_per_uri} updates per URI, got {counts:?}").into()); + } + tokio::time::sleep(TEST_POLL_INTERVAL).await; + } +} + +fn mock_backend_name(uri: &str) -> Option { + KNOWN_RESOURCE_URIS + .iter() + .find_map(|known| uri.strip_suffix(known)) + .and_then(|prefix| prefix.strip_suffix('-')) + .map(str::to_owned) +} diff --git a/crates/contextforge-data-plane-lib/tests/gateway/future_contracts/transport.rs b/crates/contextforge-data-plane-lib/tests/gateway/future_contracts/transport.rs new file mode 100644 index 00000000..252eae56 --- /dev/null +++ b/crates/contextforge-data-plane-lib/tests/gateway/future_contracts/transport.rs @@ -0,0 +1,52 @@ +use std::{fs::File, io::Read}; + +use contextforge_data_plane_lib::Result; +use rustls::crypto; + +use crate::harness::{ + TEST_USER_ID, connect_modern_client, create_client, create_tls_client, modern_client_info, start_counter_gateway, + start_tls_counter_gateway, +}; + +const EXPECTED_TOOLS: &[&str] = &[ + "00000000-0000-0000-0000-000000000001-decrement", + "00000000-0000-0000-0000-000000000001-echo", + "00000000-0000-0000-0000-000000000001-get_session_id", + "00000000-0000-0000-0000-000000000001-get_value", + "00000000-0000-0000-0000-000000000001-increment", + "00000000-0000-0000-0000-000000000001-long_task", + "00000000-0000-0000-0000-000000000001-say_hello", + "00000000-0000-0000-0000-000000000001-sum", +]; + +#[tokio::test(flavor = "multi_thread", worker_threads = 1)] +#[test_log::test] +#[ignore = "blocked on federated list-tools support"] +async fn plaintext_lists_prefixed_backend_tools() -> Result<()> { + let fixture = start_counter_gateway(TEST_USER_ID).await?; + assert_list_tools(fixture.gateway_url.clone(), create_client(TEST_USER_ID)).await +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 1)] +#[test_log::test] +#[ignore = "blocked on control-plane TLS publication and federated list-tools support"] +async fn tls_lists_prefixed_backend_tools() -> Result<()> { + let provider = crypto::ring::default_provider(); + _ = provider.install_default(); + let fixture = start_tls_counter_gateway(TEST_USER_ID).await?; + + let mut trust_bundle = Vec::new(); + File::open("../../assets/contextforgeCA/contextforge.intermediate.ca-chain.cert.pem")? + .read_to_end(&mut trust_bundle)?; + let certificates = reqwest::Certificate::from_pem_bundle(&trust_bundle)?; + assert_list_tools(fixture.gateway_url.clone(), create_tls_client(TEST_USER_ID, certificates)).await +} + +async fn assert_list_tools(gateway_url: String, client: reqwest::Client) -> Result<()> { + let running_service = connect_modern_client(&gateway_url, client, modern_client_info()).await; + let response = running_service.list_tools(None).await?; + let mut names: Vec<&str> = response.tools.iter().map(|tool| tool.name.as_ref()).collect(); + names.sort_unstable(); + assert_eq!(EXPECTED_TOOLS, names); + Ok(()) +} diff --git a/crates/contextforge-data-plane-lib/tests/support/auth.rs b/crates/contextforge-data-plane-lib/tests/gateway/harness/auth.rs similarity index 100% rename from crates/contextforge-data-plane-lib/tests/support/auth.rs rename to crates/contextforge-data-plane-lib/tests/gateway/harness/auth.rs diff --git a/crates/contextforge-data-plane-lib/tests/support/client.rs b/crates/contextforge-data-plane-lib/tests/gateway/harness/client.rs similarity index 56% rename from crates/contextforge-data-plane-lib/tests/support/client.rs rename to crates/contextforge-data-plane-lib/tests/gateway/harness/client.rs index 506de43c..56f36d01 100644 --- a/crates/contextforge-data-plane-lib/tests/support/client.rs +++ b/crates/contextforge-data-plane-lib/tests/gateway/harness/client.rs @@ -1,15 +1,13 @@ use std::time::{Duration, Instant}; -use contextforge_data_plane_lib::Result; use http::{HeaderMap, HeaderValue}; use rmcp::{ - ServiceExt, - model::{InitializeRequestParams, ProtocolVersion}, + model::InitializeRequestParams, transport::{StreamableHttpClientTransport, streamable_http_client::StreamableHttpClientTransportConfig}, }; use tracing::warn; -use super::auth::token; +use super::auth::token; // pragma: allowlist secret pub(crate) const CLIENT_CONNECT_TIMEOUT: Duration = Duration::from_secs(2); pub(crate) const TEST_POLL_INTERVAL: Duration = Duration::from_millis(20); @@ -37,59 +35,6 @@ fn auth_headers(user: &str) -> HeaderMap { default_headers } -pub(crate) async fn connect_client( - gateway_url: String, - client: reqwest::Client, -) -> Result> { - connect_client_with_handler( - gateway_url, - client, - InitializeRequestParams::default().with_protocol_version(ProtocolVersion::V_2026_07_28), - ) - .await -} - -pub(crate) async fn connect_client_with_protocol( - gateway_url: String, - client: reqwest::Client, - protocol_version: ProtocolVersion, -) -> Result> { - connect_client_with_handler( - gateway_url, - client, - InitializeRequestParams::default().with_protocol_version(protocol_version), - ) - .await -} - -/// Connects any `ClientHandler` to the gateway, retrying until `CLIENT_CONNECT_TIMEOUT`. -pub(crate) async fn connect_client_with_handler( - gateway_url: String, - client: reqwest::Client, - handler: H, -) -> Result> -where - H: rmcp::ClientHandler + Clone, -{ - let deadline = Instant::now() + CLIENT_CONNECT_TIMEOUT; - loop { - let config = StreamableHttpClientTransportConfig::with_uri(gateway_url.clone()); - let transport = StreamableHttpClientTransport::with_client(client.clone(), config); - - match handler.clone().serve(transport).await { - Ok(running_service) => return Ok(running_service), - Err(error) if Instant::now() < deadline => { - warn!("No Service {error:?}"); - tokio::time::sleep(TEST_POLL_INTERVAL).await; - }, - Err(error) => { - warn!("No Service {error:?}"); - return Err("Couldn't get a service".into()); - }, - } - } -} - pub(crate) fn modern_client_info() -> InitializeRequestParams { use rmcp::model::{ClientCapabilities, Implementation, ProtocolVersion}; InitializeRequestParams::new(ClientCapabilities::default(), Implementation::new("stateless-test-client", "0.1.0")) diff --git a/crates/contextforge-data-plane-lib/tests/gateway/harness/compatibility.rs b/crates/contextforge-data-plane-lib/tests/gateway/harness/compatibility.rs new file mode 100644 index 00000000..b812657e --- /dev/null +++ b/crates/contextforge-data-plane-lib/tests/gateway/harness/compatibility.rs @@ -0,0 +1,36 @@ +use std::time::Instant; + +use contextforge_data_plane_lib::Result; +use rmcp::{ + ServiceExt, + model::{InitializeRequestParams, ProtocolVersion}, + transport::{StreamableHttpClientTransport, streamable_http_client::StreamableHttpClientTransportConfig}, +}; +use tracing::warn; + +use super::{CLIENT_CONNECT_TIMEOUT, TEST_POLL_INTERVAL}; + +/// Connects through the pre-discovery lifecycle retained for compatibility coverage. +pub(crate) async fn connect_client_with_protocol( + gateway_url: String, + client: reqwest::Client, + protocol_version: ProtocolVersion, +) -> Result> { + let handler = InitializeRequestParams::default().with_protocol_version(protocol_version); + let deadline = Instant::now() + CLIENT_CONNECT_TIMEOUT; + + loop { + let transport = StreamableHttpClientTransport::with_client( + client.clone(), + StreamableHttpClientTransportConfig::with_uri(gateway_url.clone()), + ); + match handler.clone().serve(transport).await { + Ok(service) => return Ok(service), + Err(error) if Instant::now() < deadline => { + warn!("compatibility client has not connected yet: {error:?}"); + tokio::time::sleep(TEST_POLL_INTERVAL).await; + }, + Err(error) => return Err(format!("compatibility client could not connect: {error:?}").into()), + } + } +} diff --git a/crates/contextforge-data-plane-lib/tests/gateway/harness/gateway_fixture.rs b/crates/contextforge-data-plane-lib/tests/gateway/harness/gateway_fixture.rs new file mode 100644 index 00000000..07869977 --- /dev/null +++ b/crates/contextforge-data-plane-lib/tests/gateway/harness/gateway_fixture.rs @@ -0,0 +1,76 @@ +use std::{path::Path, sync::Arc}; + +use contextforge_data_plane_cpex::GatewayPluginRuntimeHandle; +use contextforge_data_plane_lib::{Config, Gateway, Result, UserConfigStoreType}; +use rmcp::transport::streamable_http_server::session::local::LocalSessionManager; + +use super::{MemoryUserConfigStore, TestServer, auth::AlwaysAllowAuthorizatioService}; // pragma: allowlist secret + +/// Owns a gateway and every backend server used by one integration test. +#[must_use = "dropping the fixture shuts down all in-process servers"] +pub(crate) struct GatewayFixture { + gateway: TestServer, + backends: Vec, + virtual_host_id: String, +} + +pub(crate) struct GatewayTestConfig { + pub(crate) config: Config, + pub(crate) user_store: MemoryUserConfigStore, + pub(crate) user_id: String, + pub(crate) virtual_host_id: String, + pub(crate) backends: Vec, + pub(crate) plugin_runtime: Option, +} + +impl GatewayFixture { + pub(crate) async fn start(test_config: GatewayTestConfig) -> Result { + let GatewayTestConfig { config, user_store, user_id, virtual_host_id, backends, plugin_runtime } = test_config; + let router = Self::router(config, user_store, user_id, plugin_runtime).await?; + let gateway = TestServer::start_http(router).await?; + + Ok(Self { gateway, backends, virtual_host_id }) + } + + pub(crate) async fn start_tls( + test_config: GatewayTestConfig, + certificate: impl AsRef, + private_key: impl AsRef, // pragma: allowlist secret + ) -> Result { + let GatewayTestConfig { config, user_store, user_id, virtual_host_id, backends, plugin_runtime } = test_config; + let router = Self::router(config, user_store, user_id, plugin_runtime).await?; + let gateway = TestServer::start_tls(router, certificate, private_key).await?; + + Ok(Self { gateway, backends, virtual_host_id }) + } + + async fn router( + config: Config, + user_store: MemoryUserConfigStore, + user_id: String, + plugin_runtime: Option, + ) -> Result { + Gateway::builder() + .with_config(config) + .with_session_manager(Arc::new(LocalSessionManager::default())) + .with_user_config_store_type(UserConfigStoreType::Test(Arc::new(user_store))) + .with_plugin_runtime(plugin_runtime) + .with_authorization_service(Arc::new(AlwaysAllowAuthorizatioService::new(user_id))) + .build() + .into_router() + .await + } + + pub(crate) fn gateway_url(&self) -> String { + self.gateway.url(&format!("/contextforge-rs/servers/{}/mcp", self.virtual_host_id)) + } + + pub(crate) async fn shutdown(self) -> Result<()> { + let Self { gateway, backends, .. } = self; + gateway.shutdown().await?; + for backend in backends { + backend.shutdown().await?; + } + Ok(()) + } +} diff --git a/crates/contextforge-data-plane-lib/tests/support/mock_counter.rs b/crates/contextforge-data-plane-lib/tests/gateway/harness/mock_counter.rs similarity index 94% rename from crates/contextforge-data-plane-lib/tests/support/mock_counter.rs rename to crates/contextforge-data-plane-lib/tests/gateway/harness/mock_counter.rs index 08dcd545..dcf887ce 100644 --- a/crates/contextforge-data-plane-lib/tests/support/mock_counter.rs +++ b/crates/contextforge-data-plane-lib/tests/gateway/harness/mock_counter.rs @@ -1,18 +1,10 @@ #![allow(clippy::pedantic)] -#![allow(dead_code)] use std::{sync::Arc, time::Duration}; use rmcp::{ - ErrorData as McpError, RoleServer, ServerHandler, - handler::server::{ - router::{prompt::PromptRouter, tool::ToolRouter}, - wrapper::Parameters, - }, - model::*, - prompt, prompt_handler, prompt_router, schemars, - service::RequestContext, - tool, tool_handler, tool_router, + ErrorData as McpError, RoleServer, ServerHandler, handler::server::wrapper::Parameters, model::*, prompt, + prompt_handler, prompt_router, schemars, service::RequestContext, tool, tool_handler, tool_router, }; use serde_json::json; use tokio::sync::Mutex; @@ -41,19 +33,12 @@ pub struct CounterAnalysisArgs { #[derive(Clone)] pub struct Counter { counter: Arc>, - tool_router: ToolRouter, - prompt_router: PromptRouter, } #[tool_router] impl Counter { - #[allow(dead_code)] pub fn new() -> Self { - Self { - counter: Arc::new(Mutex::new(0)), - tool_router: Self::tool_router(), - prompt_router: Self::prompt_router(), - } + Self { counter: Arc::new(Mutex::new(0)) } } fn _create_resource_text(&self, uri: &str, name: &str) -> Resource { @@ -179,7 +164,7 @@ impl ServerHandler for Counter { .build(), ) .with_server_info(Implementation::from_build_env()) - .with_protocol_version(ProtocolVersion::V_2024_11_05) + .with_protocol_version(ProtocolVersion::V_2026_07_28) .with_instructions("This server provides counter tools and prompts. Tools: increment, decrement, get_value, say_hello, echo, sum. Prompts: example_prompt (takes a message), counter_analysis (analyzes counter state with a goal).".to_owned()) } diff --git a/crates/contextforge-data-plane-lib/tests/support/mod.rs b/crates/contextforge-data-plane-lib/tests/gateway/harness/mod.rs similarity index 83% rename from crates/contextforge-data-plane-lib/tests/support/mod.rs rename to crates/contextforge-data-plane-lib/tests/gateway/harness/mod.rs index e04e427c..a718e9d7 100644 --- a/crates/contextforge-data-plane-lib/tests/support/mod.rs +++ b/crates/contextforge-data-plane-lib/tests/gateway/harness/mod.rs @@ -1,12 +1,13 @@ -#![allow(dead_code, unused_imports, reason = "shared CPEX test fixture is used by separate integration test targets")] - -pub mod auth; +mod auth; mod client; +mod compatibility; +mod gateway_fixture; pub(crate) mod mock_counter; pub(crate) mod paginating_mock; mod plugin; mod plugin_gateway; mod runtime; +mod server; mod test_gateways; mod tool; mod user_config_store; @@ -17,12 +18,14 @@ pub(crate) const TEST_USER_EMAIL: &str = "admin@example.com"; #[cfg(feature = "with_tools")] use std::{path::PathBuf, str::FromStr}; -pub(crate) use auth::token; +pub(crate) use auth::token; // pragma: allowlist secret pub(crate) use client::{ - CLIENT_CONNECT_TIMEOUT, TEST_POLL_INTERVAL, connect_client, connect_client_with_handler, - connect_client_with_protocol, connect_modern_client, create_client, create_tls_client, modern_client_info, + CLIENT_CONNECT_TIMEOUT, TEST_POLL_INTERVAL, connect_modern_client, create_client, create_tls_client, + modern_client_info, }; +pub(crate) use compatibility::connect_client_with_protocol; use contextforge_data_plane_lib::{Config, RedisConnectionMode}; +pub(crate) use gateway_fixture::{GatewayFixture, GatewayTestConfig}; pub(crate) use plugin::{ POST_DENY_ERROR_CODE, PRE_DENY_ERROR_CODE, PROMPT_ERROR_MESSAGE, PROMPT_POST_DENY_ERROR_CODE, PromptBehavior, PromptTestPlugin, REWRITTEN_PROMPT_RESOURCE, REWRITTEN_PROMPT_TEXT, REWRITTEN_PROMPT_TOPIC, REWRITTEN_SUM_A, @@ -33,9 +36,9 @@ pub(crate) use plugin_gateway::{ start_gateway_with_json_backend_responses, start_gateway_with_parameter_headers, }; pub(crate) use runtime::{runtime_with_post, runtime_with_pre, runtime_with_pre_and_post, runtime_with_prompt_plugin}; +pub(crate) use server::TestServer; pub(crate) use test_gateways::{ - ListToolsGatewaySettings, create_gateway_with_four_counters, create_gateway_with_four_legacy_counters, - create_ports, create_tls_gateway_with_four_tls_counters, plaintext_config, + start_counter_gateway, start_counter_gateway_with_backends, start_legacy_counter_gateway, start_tls_counter_gateway, }; pub(crate) use tool::{error_code, error_parts, sum_request, text}; pub(crate) use user_config_store::MemoryUserConfigStore; diff --git a/crates/contextforge-data-plane-lib/tests/support/paginating_mock.rs b/crates/contextforge-data-plane-lib/tests/gateway/harness/paginating_mock.rs similarity index 50% rename from crates/contextforge-data-plane-lib/tests/support/paginating_mock.rs rename to crates/contextforge-data-plane-lib/tests/gateway/harness/paginating_mock.rs index 9522b51d..cbe11b51 100755 --- a/crates/contextforge-data-plane-lib/tests/support/paginating_mock.rs +++ b/crates/contextforge-data-plane-lib/tests/gateway/harness/paginating_mock.rs @@ -1,6 +1,10 @@ -#![allow(dead_code, reason = "shared test fixture used by separate integration test targets")] - -use rmcp::{ErrorData as McpError, RoleServer, ServerHandler, model::*, service::RequestContext}; +use rmcp::{ + ErrorData as McpError, RoleServer, ServerHandler, + model::{ + Implementation, ListToolsResult, PaginatedRequestParams, ProtocolVersion, ServerCapabilities, ServerInfo, Tool, + }, + service::RequestContext, +}; /// Backend cursor the mock uses to signal "page 2 available". const PAGE2_CURSOR: &str = "page2"; @@ -33,6 +37,23 @@ impl ServerHandler for PaginatingServer { fn get_info(&self) -> ServerInfo { ServerInfo::new(ServerCapabilities::builder().enable_tools().build()) .with_server_info(Implementation::from_build_env()) - .with_protocol_version(ProtocolVersion::V_2024_11_05) + .with_protocol_version(ProtocolVersion::V_2026_07_28) + } + + fn list_tools( + &self, + request: Option, + _: RequestContext, + ) -> impl std::future::Future> + Send { + let result = match request.and_then(|params| params.cursor) { + None => { + let mut result = ListToolsResult::with_all_items(Self::page1_tools()); + result.next_cursor = Some(PAGE2_CURSOR.to_owned()); + Ok(result) + }, + Some(cursor) if cursor == PAGE2_CURSOR => Ok(ListToolsResult::with_all_items(Self::page2_tools())), + Some(cursor) => Err(McpError::invalid_params(format!("unknown pagination cursor: {cursor}"), None)), + }; + std::future::ready(result) } } diff --git a/crates/contextforge-data-plane-lib/tests/support/plugin.rs b/crates/contextforge-data-plane-lib/tests/gateway/harness/plugin.rs similarity index 98% rename from crates/contextforge-data-plane-lib/tests/support/plugin.rs rename to crates/contextforge-data-plane-lib/tests/gateway/harness/plugin.rs index 3d1ea7e4..a44603b1 100644 --- a/crates/contextforge-data-plane-lib/tests/support/plugin.rs +++ b/crates/contextforge-data-plane-lib/tests/gateway/harness/plugin.rs @@ -81,15 +81,6 @@ impl TestPlugin { } } - pub(crate) fn rewrite_from_config(config: PluginConfig) -> Self { - Self { - config, - observations: Arc::new(Mutex::new(Observations::default())), - pre_behavior: PreBehavior::Rewrite, - post_behavior: PostBehavior::Allow, - } - } - pub(crate) fn with_pre_rewrite(mut self) -> Self { self.pre_behavior = PreBehavior::Rewrite; self diff --git a/crates/contextforge-data-plane-lib/tests/support/plugin_gateway.rs b/crates/contextforge-data-plane-lib/tests/gateway/harness/plugin_gateway.rs similarity index 81% rename from crates/contextforge-data-plane-lib/tests/support/plugin_gateway.rs rename to crates/contextforge-data-plane-lib/tests/gateway/harness/plugin_gateway.rs index 4442f46f..cd60e3b3 100644 --- a/crates/contextforge-data-plane-lib/tests/support/plugin_gateway.rs +++ b/crates/contextforge-data-plane-lib/tests/gateway/harness/plugin_gateway.rs @@ -1,19 +1,22 @@ use std::{ collections::HashMap, - sync::{Arc, Mutex as StdMutex, OnceLock}, + sync::{Arc, Mutex as StdMutex}, time::{Duration, Instant}, }; +use super::{ + GatewayFixture, GatewayTestConfig, MemoryUserConfigStore, TestServer, connect_client_with_protocol, create_client, + create_default_config, modern_client_info, test_gateways::construct_services, token, +}; use contextforge_data_plane_apis::{ User, - user_store::{BackendMCPGateway, UserConfig, VirtualHost}, + user_store::{BackendMCPGateway, ServiceRoute, UserConfig, VirtualHost}, }; use contextforge_data_plane_cpex::CpexRuntimeRegistry; -use contextforge_data_plane_lib::{Config, Gateway, UpstreamConnectionMode, UserConfigStore, UserConfigStoreType}; -use futures::FutureExt; +use contextforge_data_plane_lib::{Config, UpstreamConnectionMode, UserConfigStore}; use http::{HeaderMap, HeaderValue, request::Parts}; use rmcp::{ - ErrorData, RoleClient, RoleServer, ServerHandler, ServiceExt, + ErrorData, RoleClient, RoleServer, ServerHandler, model::{ CallToolRequestParams, CallToolResponse, CallToolResult, ContentBlock, ErrorCode, GetPromptRequestParams, GetPromptResponse, GetPromptResult, Implementation, InitializeRequestParams, InitializeResult, NumberOrString, @@ -27,18 +30,11 @@ use rmcp::{ }, }; use serde_json::{Map, Value, json}; -use tokio::sync::Mutex as TokioMutex; - -use crate::support::{self, create_default_config, test_gateways::construct_services}; - -use super::{MemoryUserConfigStore, token}; pub(crate) const BACKEND_PROMPT_RESOURCE: &str = "token=secret"; pub(crate) const BACKEND_PROMPT_IMAGE: &str = "aW1hZ2UtYnl0ZXM="; -static GATEWAY_PORT_LOCK: OnceLock>> = OnceLock::new(); const CLIENT_CONNECT_TIMEOUT: Duration = Duration::from_secs(2); -const GATEWAY_PORT_READY_TIMEOUT: Duration = Duration::from_secs(10); const TEST_POLL_INTERVAL: Duration = Duration::from_millis(20); #[derive(Clone)] @@ -246,7 +242,7 @@ pub(crate) struct RunningGateway { pub(crate) backend_state: BackendState, pub(crate) backend_name: String, gateway_url: String, - handle: Option>>>, + fixture: GatewayFixture, } impl RunningGateway { @@ -258,7 +254,20 @@ impl RunningGateway { &self, user: &str, ) -> rmcp::service::RunningService { - self.connect_with_handler(user, InitializeRequestParams::default()).await + self.connect_with_handler(user, modern_client_info()).await + } + + pub(crate) async fn connect_legacy( + &self, + user: &str, + ) -> rmcp::service::RunningService { + connect_client_with_protocol( + self.gateway_url.clone(), + create_client(user), + rmcp::model::ProtocolVersion::V_2025_11_25, + ) + .await + .expect("legacy compatibility client connects") } pub(crate) async fn connect_with_handler( @@ -281,7 +290,15 @@ impl RunningGateway { client, StreamableHttpClientTransportConfig::with_uri(self.gateway_url.clone()), ); - match handler.clone().serve(transport).await { + match rmcp::service::serve_client_with_lifecycle( + handler.clone(), + transport, + rmcp::ClientLifecycleMode::Discover { + preferred_versions: vec![rmcp::model::ProtocolVersion::V_2026_07_28], + }, + ) + .await + { Ok(service) => return service, Err(error) if Instant::now() < deadline => { let _ = error; @@ -291,13 +308,9 @@ impl RunningGateway { } } } -} -impl Drop for RunningGateway { - fn drop(&mut self) { - if let Some(handle) = self.handle.take() { - handle.abort(); - } + pub(crate) async fn shutdown(self) -> contextforge_data_plane_lib::Result<()> { + self.fixture.shutdown().await } } @@ -363,12 +376,7 @@ async fn start_gateway_with_state( json_backend_responses: bool, backend_state: BackendState, ) -> RunningGateway { - let port_lock = Arc::clone(GATEWAY_PORT_LOCK.get_or_init(|| Arc::new(TokioMutex::new(())))); - let port_guard = port_lock.lock().await; - let gateway_port = openport::pick_random_unused_port().expect("gateway port"); - let backend_listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.expect("backend binds"); - let backend_port = backend_listener.local_addr().expect("backend address").port(); - let backend_name = format!("backend-{backend_port}"); + let backend_name = "00000000-0000-0000-0000-000000000001".to_owned(); let virtual_host_id = "vh-cpex-test"; let parameter_headers = backend_state.parameter_headers; @@ -381,7 +389,13 @@ async fn start_gateway_with_state( StreamableHttpServerConfig::default().with_json_response(json_backend_responses), ); let backend_router = axum::Router::new().route_service("/mcp", backend_service); + let backend = TestServer::start_http(backend_router).await.expect("backend starts"); + let mut tools = construct_services(&backend_name, TOOL_NAMES); + tools.insert( + format!("{backend_name}-sum"), + ServiceRoute { backend_name: backend_name.clone(), upstream_name: "sum".to_owned() }, + ); let user_store = MemoryUserConfigStore::default(); user_store .set_config( @@ -393,7 +407,7 @@ async fn start_gateway_with_state( backends: HashMap::from([( backend_name.clone(), BackendMCPGateway { - url: format!("http://127.0.0.1:{backend_port}/mcp").parse().expect("backend URL"), + url: backend.url("/mcp").parse().expect("backend URL"), name: String::new(), mcp_protocol_version: rmcp::model::ProtocolVersion::V_2026_07_28, passthrough_headers: Vec::new(), @@ -403,7 +417,7 @@ async fn start_gateway_with_state( completion: HashMap::new(), }, )]), - tools: construct_services(&backend_name, TOOL_NAMES), + tools, resources: construct_services(&backend_name, RESOURCE_URIS), resource_templates: HashMap::new(), prompts: construct_services(&backend_name, PROMPT_NAMES), @@ -414,48 +428,21 @@ async fn start_gateway_with_state( .await .expect("user config is stored"); - let gateway = Gateway::builder() - .with_config(Config { - address: Some(format!("127.0.0.1:{gateway_port}").parse().expect("This should work")), + let fixture = GatewayFixture::start(GatewayTestConfig { + config: Config { upstream_connection_mode: Some(UpstreamConnectionMode::PlainTextOrTls), runtime_plugins_enabled: Some(runtime_plugins_enabled), ..create_default_config() - }) - .with_session_manager(Arc::new(LocalSessionManager::default())) - .with_user_config_store_type(UserConfigStoreType::Test(Arc::new(user_store))) - .with_plugin_runtime(runtime_plugins_enabled.then(|| plugin_runtime.handle())) - .with_authorization_service(Arc::new(support::auth::AlwaysAllowAuthorizatioService::new(user.to_owned()))) - .build(); - - let gateway = async move { gateway.run_gateway().await }.boxed(); - let backend = async move { - axum::serve(backend_listener, backend_router).await.expect("backend serves"); - Ok(()) - } - .boxed(); - - let handle = tokio::spawn(futures::future::join_all(vec![gateway, backend])); - wait_for_gateway_port(gateway_port).await; - drop(port_guard); - - RunningGateway { - backend_state, - backend_name, - gateway_url: format!("http://127.0.0.1:{gateway_port}/contextforge-rs/servers/{virtual_host_id}/mcp"), - handle: Some(handle), - } -} + }, + user_store, + user_id: user.to_owned(), + virtual_host_id: virtual_host_id.to_owned(), + backends: vec![backend], + plugin_runtime: runtime_plugins_enabled.then(|| plugin_runtime.handle()), + }) + .await + .expect("gateway starts"); + let gateway_url = fixture.gateway_url(); -async fn wait_for_gateway_port(port: u16) { - let deadline = Instant::now() + GATEWAY_PORT_READY_TIMEOUT; - loop { - match tokio::net::TcpStream::connect(("127.0.0.1", port)).await { - Ok(_) => return, - Err(error) if Instant::now() < deadline => { - let _ = error; - tokio::time::sleep(TEST_POLL_INTERVAL).await; - }, - Err(error) => panic!("gateway TCP listener starts: {error:?}"), - } - } + RunningGateway { backend_state, backend_name, gateway_url, fixture } } diff --git a/crates/contextforge-data-plane-lib/tests/support/runtime.rs b/crates/contextforge-data-plane-lib/tests/gateway/harness/runtime.rs similarity index 100% rename from crates/contextforge-data-plane-lib/tests/support/runtime.rs rename to crates/contextforge-data-plane-lib/tests/gateway/harness/runtime.rs diff --git a/crates/contextforge-data-plane-lib/tests/gateway/harness/server.rs b/crates/contextforge-data-plane-lib/tests/gateway/harness/server.rs new file mode 100644 index 00000000..feb5056a --- /dev/null +++ b/crates/contextforge-data-plane-lib/tests/gateway/harness/server.rs @@ -0,0 +1,97 @@ +use std::{net::SocketAddr, path::Path, time::Duration}; + +use axum::Router; +use contextforge_data_plane_lib::Result; +use tokio::task::JoinHandle; +use tokio_util::sync::CancellationToken; + +const SHUTDOWN_TIMEOUT: Duration = Duration::from_secs(2); + +/// An in-process HTTP server whose listener is bound before its task starts. +#[must_use = "dropping the server shuts it down"] +pub(crate) struct TestServer { + address: SocketAddr, + scheme: &'static str, + shutdown: CancellationToken, + handle: Option>>, +} + +impl TestServer { + pub(crate) async fn start_http(router: Router) -> Result { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await?; + let address = listener.local_addr()?; + let shutdown = CancellationToken::new(); + let server_shutdown = shutdown.clone(); + let handle = tokio::spawn(async move { + axum::serve(listener, router).with_graceful_shutdown(server_shutdown.cancelled_owned()).await?; + Ok(()) + }); + + Ok(Self { address, scheme: "http", shutdown, handle: Some(handle) }) + } + + pub(crate) async fn start_tls( + router: Router, + certificate: impl AsRef, + private_key: impl AsRef, // pragma: allowlist secret + ) -> Result { + let tls_config = axum_server::tls_rustls::RustlsConfig::from_pem_file(certificate, private_key).await?; + let listener = std::net::TcpListener::bind("127.0.0.1:0")?; + let address = listener.local_addr()?; + listener.set_nonblocking(true)?; + + let shutdown = CancellationToken::new(); + let server_shutdown = shutdown.clone(); + let server_handle = axum_server::Handle::new(); + let graceful_handle = server_handle.clone(); + let handle = tokio::spawn(async move { + let server = axum_server::from_tcp_rustls(listener, tls_config)? + .handle(server_handle) + .serve(router.into_make_service()); + tokio::pin!(server); + + tokio::select! { + result = &mut server => result?, + () = server_shutdown.cancelled() => { + graceful_handle.graceful_shutdown(None); + server.await?; + } + } + Ok(()) + }); + + Ok(Self { address, scheme: "https", shutdown, handle: Some(handle) }) + } + + pub(crate) fn url(&self, path: &str) -> String { + format!("{}://{}{}", self.scheme, self.address, path) + } + + pub(crate) async fn shutdown(mut self) -> Result<()> { + self.stop().await + } + + async fn stop(&mut self) -> Result<()> { + self.shutdown.cancel(); + let Some(mut handle) = self.handle.take() else { + return Ok(()); + }; + + let Ok(result) = tokio::time::timeout(SHUTDOWN_TIMEOUT, &mut handle).await else { + handle.abort(); + let _ = handle.await; + return Ok(()); + }; + result??; + Ok(()) + } +} + +impl Drop for TestServer { + fn drop(&mut self) { + self.shutdown.cancel(); + if let Some(handle) = self.handle.take() { + handle.abort(); + } + } +} diff --git a/crates/contextforge-data-plane-lib/tests/gateway/harness/test_gateways.rs b/crates/contextforge-data-plane-lib/tests/gateway/harness/test_gateways.rs new file mode 100644 index 00000000..ae030def --- /dev/null +++ b/crates/contextforge-data-plane-lib/tests/gateway/harness/test_gateways.rs @@ -0,0 +1,204 @@ +use std::collections::HashMap; + +use contextforge_data_plane_apis::{ + User, + user_store::{BackendMCPGateway, ServiceRoute, UserConfig, VirtualHost}, +}; +use contextforge_data_plane_lib::{Config, Result, UpstreamConnectionMode, UserConfigStore}; +use rmcp::transport::{ + StreamableHttpServerConfig, StreamableHttpService, streamable_http_server::session::local::LocalSessionManager, +}; + +use super::{ + GatewayFixture, GatewayTestConfig, MemoryUserConfigStore, TestServer, create_default_config, mock_counter, +}; + +const SERVER_CERTIFICATE: &str = "../../assets/contextforgeCA/contextforge-server.cert.pem"; +const SERVER_PRIVATE_KEY: &str = "../../assets/contextforgeCA/contextforge-server.key.pem"; // pragma: allowlist secret +const UPSTREAM_TRUST_BUNDLE: &str = "../../assets/contextforgeCA/contextforge.intermediate.ca-chain.cert.pem"; +const VIRTUAL_HOST_ID: &str = "11111111-1111-1111-1111-111111111111"; + +const MOCK_COUNTER_TOOL_NAMES: &[&str] = + &["decrement", "echo", "get_session_id", "get_value", "increment", "long_task", "say_hello", "sum"]; +const MOCK_COUNTER_PROMPT_NAMES: &[&str] = &["counter_analysis", "example_prompt"]; +const MOCK_COUNTER_RESOURCE_URIS: &[&str] = &["memo://insights", "str:////Users/to/some/path/"]; + +#[must_use = "dropping the fixture shuts down the gateway and its backends"] +pub(crate) struct CounterGatewayFixture { + fixture: GatewayFixture, + pub(crate) gateway_url: String, +} + +impl CounterGatewayFixture { + pub(crate) async fn shutdown(self) -> Result<()> { + self.fixture.shutdown().await + } +} + +#[derive(Clone, Copy)] +enum TestTransport { + Plaintext, + Tls, +} + +pub(crate) async fn start_counter_gateway(user: &str) -> Result { + start_counter_gateway_with_protocol(user, rmcp::model::ProtocolVersion::V_2026_07_28).await +} + +pub(crate) async fn start_legacy_counter_gateway(user: &str) -> Result { + start_counter_gateway_with_protocol(user, rmcp::model::ProtocolVersion::V_2025_11_25).await +} + +pub(crate) async fn start_counter_gateway_with_backends( + user: &str, + backend_count: usize, +) -> Result { + start_counter_gateway_inner( + user, + rmcp::model::ProtocolVersion::V_2026_07_28, + backend_count, + TestTransport::Plaintext, + ) + .await +} + +pub(crate) async fn start_tls_counter_gateway(user: &str) -> Result { + start_counter_gateway_inner(user, rmcp::model::ProtocolVersion::V_2026_07_28, 1, TestTransport::Tls).await +} + +async fn start_counter_gateway_with_protocol( + user: &str, + protocol_version: rmcp::model::ProtocolVersion, +) -> Result { + start_counter_gateway_inner(user, protocol_version, 1, TestTransport::Plaintext).await +} + +async fn start_counter_gateway_inner( + user: &str, + protocol_version: rmcp::model::ProtocolVersion, + backend_count: usize, + transport: TestTransport, +) -> Result { + assert!(backend_count > 0, "a gateway fixture needs at least one backend"); + + let mut backend_servers = Vec::with_capacity(backend_count); + let mut backends = HashMap::with_capacity(backend_count); + + for backend_number in 1..=backend_count { + let service = StreamableHttpService::new( + || Ok(mock_counter::Counter::new()), + LocalSessionManager::default().into(), + StreamableHttpServerConfig::default(), + ); + let router = axum::Router::new().route_service("/mcp", service); + let server = match transport { + TestTransport::Plaintext => TestServer::start_http(router).await?, + TestTransport::Tls => TestServer::start_tls(router, SERVER_CERTIFICATE, SERVER_PRIVATE_KEY).await?, + }; + + let backend_id = backend_id(backend_number); + let url = server.url("/mcp").parse().expect("backend URL is valid"); + backends.insert(backend_id.clone(), backend_config(&backend_id, url, protocol_version.clone())); + backend_servers.push(server); + } + + let user_store = MemoryUserConfigStore::default(); + let tools = routes(&backends, MOCK_COUNTER_TOOL_NAMES); + let resources = routes(&backends, MOCK_COUNTER_RESOURCE_URIS); + let prompts = routes(&backends, MOCK_COUNTER_PROMPT_NAMES); + user_store + .set_config( + &User::new(user), + &UserConfig { + virtual_hosts: HashMap::from([( + VIRTUAL_HOST_ID.to_owned(), + VirtualHost { backends, tools, resources, resource_templates: HashMap::new(), prompts }, + )]), + }, + ) + .await?; + + let config = Config { + upstream_connection_mode: Some(UpstreamConnectionMode::PlainTextOrTls), + upstream_trust_bundle: matches!(transport, TestTransport::Tls).then(|| UPSTREAM_TRUST_BUNDLE.into()), + server_certificate: matches!(transport, TestTransport::Tls).then(|| SERVER_CERTIFICATE.into()), + server_private_key: matches!(transport, TestTransport::Tls).then(|| SERVER_PRIVATE_KEY.into()), // pragma: allowlist secret + ..create_default_config() + }; + + let fixture = match transport { + TestTransport::Plaintext => { + GatewayFixture::start(GatewayTestConfig { + config, + user_store, + user_id: user.to_owned(), + virtual_host_id: VIRTUAL_HOST_ID.to_owned(), + backends: backend_servers, + plugin_runtime: None, + }) + .await? + }, + TestTransport::Tls => { + GatewayFixture::start_tls( + GatewayTestConfig { + config, + user_store, + user_id: user.to_owned(), + virtual_host_id: VIRTUAL_HOST_ID.to_owned(), + backends: backend_servers, + plugin_runtime: None, + }, + SERVER_CERTIFICATE, + SERVER_PRIVATE_KEY, + ) + .await? + }, + }; + let gateway_url = fixture.gateway_url(); + + Ok(CounterGatewayFixture { fixture, gateway_url }) +} + +fn backend_config( + backend_id: &str, + url: url::Url, + protocol_version: rmcp::model::ProtocolVersion, +) -> BackendMCPGateway { + BackendMCPGateway { + name: backend_id.to_owned(), + url, + mcp_protocol_version: protocol_version, + passthrough_headers: Vec::new(), + add_headers: HashMap::new(), + remove_headers: Vec::new(), + tool_schemas: MOCK_COUNTER_TOOL_NAMES.iter().map(|name| ((*name).to_owned(), serde_json::Map::new())).collect(), + completion: HashMap::new(), + } +} + +pub(crate) fn construct_services(backend_name: &str, service_names: &[&str]) -> HashMap { + service_names + .iter() + .map(|&name| { + (name.to_owned(), ServiceRoute { backend_name: backend_name.to_owned(), upstream_name: name.to_owned() }) + }) + .collect() +} + +fn routes(backends: &HashMap, names: &[&str]) -> HashMap { + backends + .keys() + .flat_map(|backend_name| { + names.iter().map(move |&name| { + ( + format!("{backend_name}-{name}"), + ServiceRoute { backend_name: backend_name.clone(), upstream_name: name.to_owned() }, + ) + }) + }) + .collect() +} + +fn backend_id(backend_number: usize) -> String { + format!("00000000-0000-0000-0000-{backend_number:012}") +} diff --git a/crates/contextforge-data-plane-lib/tests/support/tool.rs b/crates/contextforge-data-plane-lib/tests/gateway/harness/tool.rs similarity index 100% rename from crates/contextforge-data-plane-lib/tests/support/tool.rs rename to crates/contextforge-data-plane-lib/tests/gateway/harness/tool.rs diff --git a/crates/contextforge-data-plane-lib/tests/support/user_config_store.rs b/crates/contextforge-data-plane-lib/tests/gateway/harness/user_config_store.rs similarity index 100% rename from crates/contextforge-data-plane-lib/tests/support/user_config_store.rs rename to crates/contextforge-data-plane-lib/tests/gateway/harness/user_config_store.rs diff --git a/crates/contextforge-data-plane-lib/tests/gateway_plugins.rs b/crates/contextforge-data-plane-lib/tests/gateway/plugins.rs similarity index 78% rename from crates/contextforge-data-plane-lib/tests/gateway_plugins.rs rename to crates/contextforge-data-plane-lib/tests/gateway/plugins.rs index cde1b098..0fd159cc 100644 --- a/crates/contextforge-data-plane-lib/tests/gateway_plugins.rs +++ b/crates/contextforge-data-plane-lib/tests/gateway/plugins.rs @@ -1,10 +1,7 @@ -mod support; - use std::sync::{Arc, Mutex as StdMutex}; use contextforge_data_plane_cpex::CpexRuntimeRegistry; use cpex::cpex_core::cmf::Role; -use cpex::cpex_core::config::CpexConfig; use cpex::cpex_core::hooks::types::cmf_hook_names; use rmcp::{ ClientHandler, @@ -17,7 +14,7 @@ use rmcp::{ }; use serde_json::{Map, Value, json}; -use support::{ +use crate::harness::{ BACKEND_PROMPT_IMAGE, BACKEND_PROMPT_RESOURCE, POST_DENY_ERROR_CODE, PRE_DENY_ERROR_CODE, PROMPT_ERROR_MESSAGE, PROMPT_POST_DENY_ERROR_CODE, PromptBehavior, PromptTestPlugin, REWRITTEN_PROMPT_RESOURCE, REWRITTEN_PROMPT_TEXT, REWRITTEN_PROMPT_TOPIC, REWRITTEN_SUM_A, REWRITTEN_SUM_B, RunningGateway, TEST_USER_ID, TestPlugin, error_code, @@ -151,43 +148,6 @@ fn raw_tool_call(tool_name: &str, request_id: i64, progress_token: &str) -> Valu }) } -fn fake_aws_access_key(suffix: &str) -> String { - ["AKIA", suffix].concat() -} - -fn sum_request_with_secret(secret_field: &str, secret: String) -> CallToolRequestParams { - let mut request = sum_request("sum", 1, 2); - request - .arguments - .as_mut() - .expect("sum request has arguments") - .insert(secret_field.to_owned(), Value::String(secret)); - request -} - -fn reflect_text_request(text: String) -> CallToolRequestParams { - CallToolRequestParams::new("reflect_text") - .with_arguments(Map::from_iter([("text".to_owned(), Value::String(text))])) -} - -async fn runtime_with_secrets_detection(hooks: Vec<&'static str>, plugin_config: Value) -> Arc { - let mut runtime = CpexRuntimeRegistry::default(); - runtime - .register_factory(cpex_secrets_detection::KIND, Box::new(cpex_secrets_detection::SecretsDetectionFactory)) - .expect("secrets detection factory registers"); - let config: CpexConfig = serde_json::from_value(json!({ - "plugins": [{ - "name": "secrets-detection", - "kind": cpex_secrets_detection::KIND, - "hooks": hooks, - "config": plugin_config, - }] - })) - .expect("secrets detection CPEX config parses"); - runtime.apply_config(Some(config)).await.expect("secrets detection runtime applies"); - Arc::new(runtime) -} - fn sse_data_values(body: &str) -> Vec { let values = body .lines() @@ -401,10 +361,10 @@ async fn disabled_runtime_does_not_invoke_registered_plugin() { async fn stateless_tool_call_forwards_validated_parameter_headers_unchanged() { let gateway = start_gateway_with_parameter_headers(TEST_USER_ID, false, Arc::new(CpexRuntimeRegistry::default())).await; - let service = support::connect_modern_client( + let service = crate::harness::connect_modern_client( gateway.gateway_url(), client_with_parameter_headers("1", "2"), - support::modern_client_info(), + crate::harness::modern_client_info(), ) .await; let result = service.call_tool(sum_request("sum", 1, 2)).await.expect("stateless tool call succeeds"); @@ -419,10 +379,10 @@ async fn stateless_tool_call_forwards_validated_parameter_headers_unchanged() { async fn stateless_tool_call_with_mismatched_parameter_header_is_rejected() { let gateway = start_gateway_with_parameter_headers(TEST_USER_ID, false, Arc::new(CpexRuntimeRegistry::default())).await; - let service = support::connect_modern_client( + let service = crate::harness::connect_modern_client( gateway.gateway_url(), client_with_parameter_headers("9", "2"), - support::modern_client_info(), + crate::harness::modern_client_info(), ) .await; let error = service.call_tool(sum_request("sum", 1, 2)).await.expect_err("mismatched header is rejected"); @@ -438,10 +398,10 @@ async fn stateless_tool_call_with_mismatched_parameter_header_is_rejected() { async fn stateless_tool_call_without_required_parameter_headers_is_rejected() { let gateway = start_gateway_with_parameter_headers(TEST_USER_ID, false, Arc::new(CpexRuntimeRegistry::default())).await; - let service = support::connect_modern_client( + let service = crate::harness::connect_modern_client( gateway.gateway_url(), - support::create_client(TEST_USER_ID), - support::modern_client_info(), + crate::harness::create_client(TEST_USER_ID), + crate::harness::modern_client_info(), ) .await; @@ -457,10 +417,10 @@ async fn stateless_tool_call_without_required_parameter_headers_is_rejected() { #[tokio::test(flavor = "multi_thread", worker_threads = 1)] async fn stateless_tool_call_without_published_schema_reaches_backend() { let gateway = start_gateway(TEST_USER_ID, false, Arc::new(CpexRuntimeRegistry::default())).await; - let service = support::connect_modern_client( + let service = crate::harness::connect_modern_client( gateway.gateway_url(), client_with_parameter_headers("1", "2"), - support::modern_client_info(), + crate::harness::modern_client_info(), ) .await; let error = service.call_tool(CallToolRequestParams::new("missing_schema_tool")).await.unwrap_err(); @@ -477,27 +437,13 @@ async fn stateless_tool_call_without_published_schema_reaches_backend() { assert_eq!("2", headers["Mcp-Param-B"]); } -#[tokio::test(flavor = "multi_thread", worker_threads = 1)] -async fn legacy_tool_call_without_published_schema_reaches_backend() { - let gateway = start_gateway(TEST_USER_ID, false, Arc::new(CpexRuntimeRegistry::default())).await; - let service = gateway.connect(TEST_USER_ID).await; - let error = service.call_tool(CallToolRequestParams::new("missing_schema_tool")).await.unwrap_err(); - let rmcp::service::ServiceError::McpError(error) = error else { - panic!("expected backend MCP error, got {error:?}"); - }; - - assert_eq!(ErrorCode::METHOD_NOT_FOUND, error.code); - let backend_calls = gateway.backend_state.calls.lock().expect("backend calls lock poisoned"); - assert_eq!("missing_schema_tool", backend_calls[0].tool_name); -} - #[tokio::test(flavor = "multi_thread", worker_threads = 1)] async fn stateless_tool_error_round_trips() { let gateway = start_gateway(TEST_USER_ID, false, Arc::new(CpexRuntimeRegistry::default())).await; - let service = support::connect_modern_client( + let service = crate::harness::connect_modern_client( gateway.gateway_url(), - support::create_client(TEST_USER_ID), - support::modern_client_info(), + crate::harness::create_client(TEST_USER_ID), + crate::harness::modern_client_info(), ) .await; let error = service.call_tool(CallToolRequestParams::new("missing_tool")).await.unwrap_err(); @@ -509,24 +455,23 @@ async fn stateless_tool_error_round_trips() { #[tokio::test(flavor = "multi_thread", worker_threads = 1)] async fn stateless_alias_and_namespaced_tool_names_route() { - let gateway_port = support::create_ports(1)[0]; - let support::ListToolsGatewaySettings { handle, gateway_url, expected_tool_names, .. } = - support::create_gateway_with_four_counters(TEST_USER_ID, support::plaintext_config(gateway_port)) - .await - .expect("gateway starts"); - let service = support::connect_modern_client( - &gateway_url, - support::create_client(TEST_USER_ID), - support::modern_client_info(), + let gateway = start_gateway(TEST_USER_ID, false, Arc::new(CpexRuntimeRegistry::default())).await; + let service = crate::harness::connect_modern_client( + gateway.gateway_url(), + crate::harness::create_client(TEST_USER_ID), + crate::harness::modern_client_info(), ) .await; - let alias = expected_tool_names.iter().find(|name| name.ends_with("sum")).expect("sum alias is advertised"); - let alias_result = service.call_tool(sum_request(alias, 1, 2)).await.expect("alias routes"); - let namespaced_result = service.call_tool(sum_request(alias, 3, 4)).await.expect("namespace routes"); + let alias_result = service.call_tool(sum_request("sum", 1, 2)).await.expect("control-plane alias routes"); + let namespaced_result = service + .call_tool(sum_request(format!("{}-sum", gateway.backend_name), 3, 4)) + .await + .expect("namespaced tool routes"); assert_eq!("3", text(&alias_result)); assert_eq!("7", text(&namespaced_result)); - handle.abort(); + drop(service); + gateway.shutdown().await.expect("gateway shuts down"); } #[tokio::test(flavor = "multi_thread", worker_threads = 2)] @@ -534,8 +479,12 @@ async fn stateless_concurrent_progress_calls_remain_request_scoped() { let gateway = start_gateway(TEST_USER_ID, false, Arc::new(CpexRuntimeRegistry::default())).await; let client = RecordingClient::default(); let progress = Arc::clone(&client.progress); - let service = - support::connect_modern_client(gateway.gateway_url(), support::create_client(TEST_USER_ID), client).await; + let service = crate::harness::connect_modern_client( + gateway.gateway_url(), + crate::harness::create_client(TEST_USER_ID), + client, + ) + .await; let first = send_progress_call(&service, "progress_sum").await; let first_progress_token = first.progress_token.clone(); let second = send_progress_call(&service, "progress_sum").await; @@ -560,139 +509,6 @@ async fn stateless_concurrent_progress_calls_remain_request_scoped() { assert_eq!(4, progress.iter().filter(|event| event.progress_token == second_progress_token).count()); } -#[tokio::test(flavor = "multi_thread", worker_threads = 1)] -async fn secrets_detection_pre_hook_redacts_tool_arguments_before_backend_call() { - let runtime = runtime_with_secrets_detection( - vec![cmf_hook_names::TOOL_PRE_INVOKE], - json!({ - "redact": true, - "redaction_text": "[redacted]", - "block_on_detection": false, - }), - ) - .await; - let gateway = start_gateway("admin@example.com", true, runtime).await; - let service = gateway.connect("admin@example.com").await; - - let result = service - .call_tool(sum_request_with_secret("credential", fake_aws_access_key("1111111111111111"))) - .await - .expect("secret argument is redacted and call succeeds"); - - assert_eq!("3", text(&result)); - let backend_calls = gateway.backend_state.calls.lock().expect("backend calls lock poisoned"); - assert_eq!(1, backend_calls.len()); - assert_eq!( - Some(&Value::from("[redacted]")), - backend_calls[0].args.as_ref().and_then(|args| args.get("credential")) - ); -} - -#[tokio::test(flavor = "multi_thread", worker_threads = 1)] -async fn secrets_detection_clean_tool_payload_passes_through_unchanged() { - let runtime = runtime_with_secrets_detection( - vec![cmf_hook_names::TOOL_PRE_INVOKE, cmf_hook_names::TOOL_POST_INVOKE], - json!({ - "redact": true, - "redaction_text": "[redacted]", - "block_on_detection": true, - }), - ) - .await; - let gateway = start_gateway("admin@example.com", true, runtime).await; - let service = gateway.connect("admin@example.com").await; - - let result = - service.call_tool(sum_request("sum", 1, 2)).await.expect("clean argument payload passes through unchanged"); - - let result_text = text(&result); - assert_eq!("3", result_text.as_str()); - assert!(!result_text.contains("[redacted]")); - let backend_calls = gateway.backend_state.calls.lock().expect("backend calls lock poisoned"); - assert_eq!(1, backend_calls.len()); - assert_eq!("sum", backend_calls[0].tool_name); - let args = backend_calls[0].args.as_ref().expect("backend call has args"); - assert_eq!(2, args.len()); - assert_eq!(Some(&Value::from(1)), args.get("a")); - assert_eq!(Some(&Value::from(2)), args.get("b")); -} - -#[tokio::test(flavor = "multi_thread", worker_threads = 1)] -async fn secrets_detection_pre_hook_blocks_tool_arguments_before_backend_call() { - let runtime = runtime_with_secrets_detection( - vec![cmf_hook_names::TOOL_PRE_INVOKE], - json!({ - "redact": false, - "block_on_detection": true, - }), - ) - .await; - let gateway = start_gateway("admin@example.com", true, runtime).await; - let service = gateway.connect("admin@example.com").await; - - let error = service - .call_tool(sum_request_with_secret("credential", fake_aws_access_key("2222222222222222"))) - .await - .expect_err("secret argument blocks the call"); - - assert_eq!(ErrorCode::INVALID_REQUEST, error_code(error)); - assert!(gateway.backend_state.calls.lock().expect("backend calls lock poisoned").is_empty()); -} - -#[tokio::test(flavor = "multi_thread", worker_threads = 1)] -async fn secrets_detection_post_hook_redacts_tool_result_before_client_response() { - let runtime = runtime_with_secrets_detection( - vec![cmf_hook_names::TOOL_POST_INVOKE], - json!({ - "redact": true, - "redaction_text": "[redacted]", - "block_on_detection": false, - }), - ) - .await; - let gateway = start_gateway("admin@example.com", true, runtime).await; - let service = gateway.connect("admin@example.com").await; - - let result = service - .call_tool(reflect_text_request(fake_aws_access_key("3333333333333333"))) - .await - .expect("secret result is redacted and call succeeds"); - - assert_eq!("[redacted]", text(&result)); - assert_eq!(1, gateway.backend_state.calls.lock().expect("backend calls lock poisoned").len()); -} - -#[tokio::test(flavor = "multi_thread", worker_threads = 1)] -async fn secrets_detection_pre_hook_respects_field_allowlist() { - let runtime = runtime_with_secrets_detection( - vec![cmf_hook_names::TOOL_PRE_INVOKE], - json!({ - "redact": true, - "redaction_text": "[redacted]", - "block_on_detection": false, - "field_allowlist": ["credential"], - }), - ) - .await; - let gateway = start_gateway("admin@example.com", true, runtime).await; - let service = gateway.connect("admin@example.com").await; - let ignored_secret = fake_aws_access_key("4444444444444444"); - let mut request = sum_request_with_secret("credential", fake_aws_access_key("5555555555555555")); - request - .arguments - .as_mut() - .expect("sum request has arguments") - .insert("ignored".to_owned(), Value::String(ignored_secret.clone())); - - let result = service.call_tool(request).await.expect("allowed field is redacted"); - - assert_eq!("3", text(&result)); - let backend_calls = gateway.backend_state.calls.lock().expect("backend calls lock poisoned"); - let args = backend_calls[0].args.as_ref().expect("backend call has args"); - assert_eq!(Some(&Value::from("[redacted]")), args.get("credential")); - assert_eq!(Some(&Value::from(ignored_secret)), args.get("ignored")); -} - #[tokio::test(flavor = "multi_thread", worker_threads = 1)] async fn pre_hook_rewrites_payload_without_changing_forwarded_parameter_headers() { let plugin = Arc::new(TestPlugin::new("pre", vec![cmf_hook_names::TOOL_PRE_INVOKE]).with_pre_rewrite()); @@ -700,10 +516,10 @@ async fn pre_hook_rewrites_payload_without_changing_forwarded_parameter_headers( let runtime = runtime_with_pre(plugin).await; let gateway = start_gateway_with_parameter_headers(TEST_USER_ID, true, runtime).await; - let service = support::connect_modern_client( + let service = crate::harness::connect_modern_client( gateway.gateway_url(), client_with_parameter_headers("1", "2"), - support::modern_client_info(), + crate::harness::modern_client_info(), ) .await; let result = service.call_tool(sum_request("sum", 1, 2)).await.unwrap(); @@ -794,26 +610,6 @@ async fn post_hook_deny_drops_progress_notifications_without_failing_call() { assert!(progress.lock().expect("progress lock poisoned").is_empty()); } -#[tokio::test(flavor = "multi_thread", worker_threads = 1)] -#[ignore = "2026-07-28 protocol transition"] -async fn downstream_cancellation_is_relayed_to_backend() { - let gateway = start_gateway(TEST_USER_ID, true, Arc::new(CpexRuntimeRegistry::default())).await; - let service = gateway.connect(TEST_USER_ID).await; - - let request = CallToolRequestParams::new("wait_for_cancellation"); - let handle = service - .send_cancellable_request( - ClientRequest::CallToolRequest(Request::new(request)), - PeerRequestOptions::no_options(), - ) - .await - .expect("wait_for_cancellation request is sent"); - wait_for_event_count(&gateway.backend_state.calls, 1).await; - - handle.cancel(Some("client gave up".to_owned())).await.expect("cancellation is sent"); - wait_for_event_count(&gateway.backend_state.cancellations, 1).await; -} - #[tokio::test(flavor = "multi_thread", worker_threads = 1)] async fn post_hook_can_return_raw_cmf_result_content() { let plugin = Arc::new(TestPlugin::new("post", vec![cmf_hook_names::TOOL_POST_INVOKE]).with_raw_post_rewrite()); diff --git a/crates/contextforge-data-plane-lib/tests/gateway/prompts.rs b/crates/contextforge-data-plane-lib/tests/gateway/prompts.rs new file mode 100644 index 00000000..023a416a --- /dev/null +++ b/crates/contextforge-data-plane-lib/tests/gateway/prompts.rs @@ -0,0 +1,25 @@ +use contextforge_data_plane_lib::Result; +use rmcp::model::GetPromptRequestParams; +use serde_json::json; + +use crate::harness::{TEST_USER_ID, connect_modern_client, create_client, modern_client_info, start_counter_gateway}; + +const EXAMPLE_PROMPT: &str = "00000000-0000-0000-0000-000000000001-example_prompt"; + +#[tokio::test] +async fn plaintext_gets_prompt_from_prefixed_backend_name() -> Result<()> { + let fixture = start_counter_gateway(TEST_USER_ID).await?; + let service = connect_modern_client(&fixture.gateway_url, create_client(TEST_USER_ID), modern_client_info()).await; + let mut arguments = serde_json::Map::new(); + arguments.insert("message".to_owned(), json!("hello from gateway")); + + let result = service.get_prompt(GetPromptRequestParams::new(EXAMPLE_PROMPT).with_arguments(arguments)).await?; + let text = result + .messages + .first() + .and_then(|message| message.content.as_text()) + .map(|content| &content.text) + .ok_or("expected a text prompt message")?; + assert!(text.contains("hello from gateway"), "unexpected prompt text: {text}"); + Ok(()) +} diff --git a/crates/contextforge-data-plane-lib/tests/gateway/resources.rs b/crates/contextforge-data-plane-lib/tests/gateway/resources.rs new file mode 100644 index 00000000..2fafbfe8 --- /dev/null +++ b/crates/contextforge-data-plane-lib/tests/gateway/resources.rs @@ -0,0 +1,35 @@ +use contextforge_data_plane_lib::Result; +use rmcp::model::{ErrorCode, ReadResourceRequestParams, ResourceContents}; + +use crate::harness::{ + TEST_USER_ID, connect_modern_client, create_client, error_parts, modern_client_info, start_counter_gateway, +}; + +const MEMO_RESOURCE: &str = "00000000-0000-0000-0000-000000000001-memo://insights"; +const EXPECTED_MEMO: &str = "Business Intelligence Memo\n\nAnalysis has revealed 5 key insights ..."; + +#[tokio::test] +async fn plaintext_call_prefixed_read_resources_modern_modern() -> Result<()> { + let fixture = start_counter_gateway(TEST_USER_ID).await?; + let service = connect_modern_client(&fixture.gateway_url, create_client(TEST_USER_ID), modern_client_info()).await; + let response = service.read_resource(ReadResourceRequestParams::new(MEMO_RESOURCE)).await?; + let Some(ResourceContents::TextResourceContents { text, .. }) = response.contents.first() else { + panic!("expected one text resource, got {:?}", response.contents); + }; + assert_eq!(EXPECTED_MEMO, text); + Ok(()) +} + +#[tokio::test] +async fn plaintext_read_invalid_backend_resource() -> Result<()> { + let fixture = start_counter_gateway(TEST_USER_ID).await?; + let service = connect_modern_client(&fixture.gateway_url, create_client(TEST_USER_ID), modern_client_info()).await; + let error = service + .read_resource(ReadResourceRequestParams::new("http://dummy.dummy")) + .await + .expect_err("an unknown resource must fail routing"); + let (code, message) = error_parts(error); + assert_eq!(ErrorCode::INVALID_PARAMS, code); + assert_eq!("Routing problem... resource not found", message); + Ok(()) +} diff --git a/crates/contextforge-data-plane-lib/tests/gateway/subscriptions.rs b/crates/contextforge-data-plane-lib/tests/gateway/subscriptions.rs new file mode 100644 index 00000000..acafda34 --- /dev/null +++ b/crates/contextforge-data-plane-lib/tests/gateway/subscriptions.rs @@ -0,0 +1,22 @@ +use contextforge_data_plane_lib::Result; +use rmcp::model::{ErrorCode, SubscribeRequestParams}; + +use crate::harness::{ + TEST_USER_ID, connect_modern_client, create_client, error_parts, modern_client_info, start_counter_gateway, +}; + +#[tokio::test] +#[expect(deprecated, reason = "legacy RMCP API used to assert the delegated operation response")] +async fn plaintext_subscribe_to_unrouted_resource_errors() -> Result<()> { + let fixture = start_counter_gateway(TEST_USER_ID).await?; + let service = connect_modern_client(&fixture.gateway_url, create_client(TEST_USER_ID), modern_client_info()).await; + + let error = service + .subscribe(SubscribeRequestParams::new("unrouted://resource")) + .await + .expect_err("subscriptions are delegated to the control plane"); + let (code, message) = error_parts(error); + assert_eq!(ErrorCode::METHOD_NOT_FOUND, code); + assert_eq!("resources/subscribe", message); + Ok(()) +} diff --git a/crates/contextforge-data-plane-lib/tests/gateway/tools.rs b/crates/contextforge-data-plane-lib/tests/gateway/tools.rs new file mode 100644 index 00000000..6ac4f77c --- /dev/null +++ b/crates/contextforge-data-plane-lib/tests/gateway/tools.rs @@ -0,0 +1,33 @@ +use contextforge_data_plane_lib::Result; +use rmcp::model::{CallToolRequestParams, ErrorCode}; + +use crate::harness::{ + TEST_USER_ID, connect_modern_client, create_client, error_parts, modern_client_info, start_counter_gateway, +}; + +const DECREMENT_TOOL: &str = "00000000-0000-0000-0000-000000000001-decrement"; + +#[tokio::test] +async fn plaintext_call_prefixed_backend_tools_modern_modern() -> Result<()> { + let fixture = start_counter_gateway(TEST_USER_ID).await?; + let service = connect_modern_client(&fixture.gateway_url, create_client(TEST_USER_ID), modern_client_info()).await; + let result = service.call_tool(CallToolRequestParams::new(DECREMENT_TOOL)).await?; + let text = result.content.first().and_then(|content| content.as_text()).expect("text tool result"); + assert_eq!("-1", text.text); + drop(service); + fixture.shutdown().await +} + +#[tokio::test] +async fn plaintext_call_invalid_backend_tools() -> Result<()> { + let fixture = start_counter_gateway(TEST_USER_ID).await?; + let service = connect_modern_client(&fixture.gateway_url, create_client(TEST_USER_ID), modern_client_info()).await; + let error = service + .call_tool(CallToolRequestParams::new("dummy_tool")) + .await + .expect_err("an unknown tool must fail routing"); + let (code, message) = error_parts(error); + assert_eq!(ErrorCode::INVALID_PARAMS, code); + assert_eq!("Routing problem... tool not found", message); + Ok(()) +} diff --git a/crates/contextforge-data-plane-lib/tests/gateway_call_tools.rs b/crates/contextforge-data-plane-lib/tests/gateway_call_tools.rs deleted file mode 100644 index 3f8ba965..00000000 --- a/crates/contextforge-data-plane-lib/tests/gateway_call_tools.rs +++ /dev/null @@ -1,199 +0,0 @@ -mod support; - -use contextforge_data_plane_lib::{Config, Result, UpstreamConnectionMode}; -use rmcp::model::{CallToolRequestParams, ProtocolVersion}; -use tracing::{info, warn}; - -use support::{ListToolsGatewaySettings, TEST_USER_ID, create_client, create_gateway_with_four_counters, create_ports}; - -use crate::support::{ - connect_client_with_protocol, connect_modern_client, create_default_config, - create_gateway_with_four_legacy_counters, -}; - -#[tokio::test(flavor = "multi_thread", worker_threads = 1)] -#[test_log::test] -async fn plaintext_call_prefixed_backend_tools_modern_modern() -> Result<()> { - let gateway_port = create_ports(1)[0]; - - let config = Config { - address: Some(format!("127.0.0.1:{gateway_port}").parse().expect("This should work")), - upstream_connection_mode: Some(UpstreamConnectionMode::PlainTextOrTls), - ..create_default_config() - }; - - let user = TEST_USER_ID; - - let Ok(ListToolsGatewaySettings { handle, gateway_url, expected_tool_names, .. }) = - create_gateway_with_four_counters(user, config).await - else { - panic!("Invalid configuration "); - }; - - let client = create_client(user); - - let mut call_params = CallToolRequestParams::default(); - call_params.name = expected_tool_names[0].clone().into(); - let maybe_passed = - assert_tools_call(gateway_url, client, call_params, "-1".to_owned(), ProtocolVersion::V_2026_07_28).await; - - handle.abort(); - if maybe_passed.is_ok() { - info!("Test passed"); - } else { - info!("Test NOT passed {maybe_passed:?}"); - panic!() - } - - Ok(()) -} - -#[tokio::test(flavor = "multi_thread", worker_threads = 1)] -#[test_log::test] -async fn plaintext_call_prefixed_backend_tools_modern_legacy() -> Result<()> { - let gateway_port = create_ports(1)[0]; - - let config = Config { - address: Some(format!("127.0.0.1:{gateway_port}").parse().expect("This should work")), - upstream_connection_mode: Some(UpstreamConnectionMode::PlainTextOrTls), - ..create_default_config() - }; - - let user = TEST_USER_ID; - - let Ok(ListToolsGatewaySettings { handle, gateway_url, expected_tool_names, .. }) = - create_gateway_with_four_legacy_counters(user, config).await - else { - panic!("Invalid configuration "); - }; - - let client = create_client(user); - - let mut call_params = CallToolRequestParams::default(); - call_params.name = expected_tool_names[0].clone().into(); - let maybe_passed = - assert_tools_call(gateway_url, client, call_params, "-1".to_owned(), ProtocolVersion::V_2026_07_28).await; - - handle.abort(); - if maybe_passed.is_ok() { - info!("Test passed"); - } else { - info!("Test NOT passed {maybe_passed:?}"); - panic!() - } - - Ok(()) -} - -#[tokio::test(flavor = "multi_thread", worker_threads = 1)] -#[test_log::test] -async fn plaintext_call_prefixed_backend_tools_legacy_modern() -> Result<()> { - let gateway_port = create_ports(1)[0]; - - let config = Config { - address: Some(format!("127.0.0.1:{gateway_port}").parse().expect("This should work")), - upstream_connection_mode: Some(UpstreamConnectionMode::PlainTextOrTls), - ..create_default_config() - }; - - let user = TEST_USER_ID; - - let Ok(ListToolsGatewaySettings { handle, gateway_url, expected_tool_names, .. }) = - create_gateway_with_four_counters(user, config).await - else { - panic!("Invalid configuration "); - }; - - let client = create_client(user); - - let mut call_params = CallToolRequestParams::default(); - call_params.name = expected_tool_names[0].clone().into(); - let maybe_passed = - assert_tools_call(gateway_url, client, call_params, "-1".to_owned(), ProtocolVersion::V_2025_11_25).await; - - handle.abort(); - if maybe_passed.is_ok() { - info!("Test passed"); - } else { - info!("Test NOT passed {maybe_passed:?}"); - panic!() - } - - Ok(()) -} - -async fn assert_tools_call( - gateway_url: String, - client: reqwest::Client, - call_tool_params: CallToolRequestParams, - expected_result: String, - protocol_version: ProtocolVersion, -) -> Result<()> { - info!("Seding request to {gateway_url}"); - - let running_service = if protocol_version == ProtocolVersion::V_2026_07_28 { - connect_modern_client(&gateway_url, client, support::modern_client_info()).await - } else { - connect_client_with_protocol(gateway_url, client, protocol_version).await? - }; - - let call_tool = running_service.call_tool(call_tool_params).await; - let Ok(call_tool) = call_tool else { - let msg = format!("Call tool returned error {call_tool:?}"); - warn!(msg); - return Err(msg.into()); - }; - - if call_tool.content.is_empty() { - let msg = format!("Call tool returned empty response {call_tool:?}"); - warn!(msg); - return Err(msg.into()); - } - - if let Some(text) = call_tool.content[0].as_text() { - if text.text != expected_result { - let msg = format!("Call tool returned unexpected response {} {} {call_tool:?}", text.text, expected_result); - warn!(msg); - return Err(msg.into()); - } - } else { - let msg = format!("Call tool returned non text response {call_tool:?}"); - warn!(msg); - return Err(msg.into()); - } - - Ok(()) -} - -#[tokio::test(flavor = "multi_thread", worker_threads = 1)] -#[test_log::test] -async fn plaintext_call_invalid_backend_tools() -> Result<()> { - let gateway_port = create_ports(1)[0]; - - let config = Config { - address: Some(format!("127.0.0.1:{gateway_port}").parse().expect("This should work")), - upstream_connection_mode: Some(UpstreamConnectionMode::PlainTextOrTls), - ..create_default_config() - }; - - let user = TEST_USER_ID; - - let Ok(ListToolsGatewaySettings { handle, gateway_url, .. }) = - create_gateway_with_four_counters(user, config).await - else { - panic!("Invalid configuration "); - }; - - let client = create_client(user); - - let mut call_params = CallToolRequestParams::default(); - call_params.name = "dummy_tool".into(); - let maybe_passed = - assert_tools_call(gateway_url, client, call_params, "-1".to_owned(), ProtocolVersion::V_2026_07_28).await; - handle.abort(); - if maybe_passed.is_ok() { - info!("Test NOT passed {maybe_passed:?}"); - panic!() - } - Ok(()) -} diff --git a/crates/contextforge-data-plane-lib/tests/gateway_completions.rs b/crates/contextforge-data-plane-lib/tests/gateway_completions.rs deleted file mode 100644 index 6fefff01..00000000 --- a/crates/contextforge-data-plane-lib/tests/gateway_completions.rs +++ /dev/null @@ -1,121 +0,0 @@ -mod support; - -use contextforge_data_plane_lib::Result; -use tracing::info; - -use support::{ - ListToolsGatewaySettings, TEST_USER_ID, connect_client, create_client, create_gateway_with_four_counters, - create_ports, plaintext_config, -}; - -use crate::support::connect_modern_client; - -#[tokio::test(flavor = "multi_thread", worker_threads = 1)] -#[test_log::test] -#[ignore = "2026-07-28 protocol transition"] -async fn plaintext_completes_prompt_argument_through_prefixed_backend() -> Result<()> { - let gateway_port = create_ports(1)[0]; - let user = TEST_USER_ID; - let Ok(ListToolsGatewaySettings { handle, gateway_url, .. }) = - create_gateway_with_four_counters(user, plaintext_config(gateway_port)).await - else { - panic!("Invalid configuration "); - }; - - let client = create_client(user); - let maybe_passed = assert_prompt_completion(gateway_url, client).await; - - handle.abort(); - maybe_passed -} - -#[tokio::test(flavor = "multi_thread", worker_threads = 1)] -#[test_log::test] -#[ignore = "2026-07-28 protocol transition"] -async fn plaintext_completes_resource_argument_through_prefixed_backend() -> Result<()> { - let gateway_port = create_ports(1)[0]; - let user = TEST_USER_ID; - let Ok(ListToolsGatewaySettings { handle, gateway_url, .. }) = - create_gateway_with_four_counters(user, plaintext_config(gateway_port)).await - else { - panic!("Invalid configuration "); - }; - - let client = create_client(user); - let maybe_passed = assert_resource_completion(gateway_url, client).await; - - handle.abort(); - maybe_passed -} - -#[tokio::test(flavor = "multi_thread", worker_threads = 1)] -#[test_log::test] -async fn plaintext_complete_for_unrouted_reference_errors() -> Result<()> { - let gateway_port = create_ports(1)[0]; - let user = TEST_USER_ID; - let Ok(ListToolsGatewaySettings { handle, gateway_url, .. }) = - create_gateway_with_four_counters(user, plaintext_config(gateway_port)).await - else { - panic!("Invalid configuration "); - }; - - let client = create_client(user); - let maybe_passed = assert_unrouted_completion_errors(gateway_url, client).await; - - handle.abort(); - maybe_passed -} - -async fn assert_prompt_completion(gateway_url: String, client: reqwest::Client) -> Result<()> { - info!("Sending request to {gateway_url}"); - let running_service = connect_client(gateway_url, client).await?; - - // Spec-compliant clients only issue completion/complete when the server advertises the - // capability, so the gateway must declare it for the proxying below to be reachable. - if running_service.peer_info().and_then(|info| info.capabilities.completions.clone()).is_none() { - return Err("gateway must advertise the completions capability".into()); - } - - let prompts = running_service.list_prompts(None).await?; - let prompt_name = prompts - .prompts - .iter() - .find(|p| p.name.ends_with("-example_prompt")) - .map(|p| p.name.clone()) - .ok_or("expected a federated example_prompt")?; - - // The backend only knows the un-prefixed prompt name, so a non-empty result proves the gateway - // routed to a single backend and stripped the namespace prefix before forwarding. - let values = running_service.complete_prompt_simple(prompt_name, "message", "h").await?; - if !values.contains(&"hello".to_owned()) { - return Err(format!("expected backend prompt completions, got: {values:?}").into()); - } - - Ok(()) -} - -async fn assert_resource_completion(gateway_url: String, client: reqwest::Client) -> Result<()> { - let running_service = connect_client(gateway_url, client).await?; - - let resources = running_service.list_resources(None).await?; - let uri = resources.resources.first().ok_or("expected at least one federated resource")?.uri.clone(); - - // The mock echoes back the URI it received; it must be the stripped backend-local URI. - let values = running_service.complete_resource_simple(uri, "path", "").await?; - match values.first() { - Some(value) if !value.starts_with("backend-") => Ok(()), - other => Err(format!("expected stripped backend URI in completion, got: {other:?}").into()), - } -} - -async fn assert_unrouted_completion_errors(gateway_url: String, client: reqwest::Client) -> Result<()> { - let running_service = connect_modern_client(&gateway_url, client, support::modern_client_info()).await; - - // No backend namespace prefix => no route, so the gateway must reject it. - let result = running_service.complete_prompt_simple("unrouted_prompt", "message", "h").await; - if result.is_ok() { - return Err("expected a routing error for an unrouted completion reference".into()); - } - - Ok(()) -} diff --git a/crates/contextforge-data-plane-lib/tests/gateway_list_tools.rs b/crates/contextforge-data-plane-lib/tests/gateway_list_tools.rs deleted file mode 100644 index 97c9adcd..00000000 --- a/crates/contextforge-data-plane-lib/tests/gateway_list_tools.rs +++ /dev/null @@ -1,127 +0,0 @@ -mod support; - -use std::{fs::File, io::Read}; - -use contextforge_data_plane_lib::{Config, Result, UpstreamConnectionMode}; -use futures::{FutureExt, future::BoxFuture}; -use rustls::crypto; -use tracing::{info, warn}; - -use support::{ - ListToolsGatewaySettings, TEST_USER_ID, connect_client, create_client, create_gateway_with_four_counters, - create_ports, create_tls_client, create_tls_gateway_with_four_tls_counters, -}; - -use crate::support::create_default_config; - -#[tokio::test(flavor = "multi_thread", worker_threads = 1)] -#[test_log::test] -#[ignore = "Fan out list tools is not supported at the moment. This should be enabled in 2.x"] -async fn plaintext_lists_prefixed_backend_tools() -> Result<()> { - let gateway_port = create_ports(1)[0]; - - let config = Config { - address: Some(format!("127.0.0.1:{gateway_port}").parse().expect("This should work")), - upstream_connection_mode: Some(UpstreamConnectionMode::PlainTextOrTls), - ..create_default_config() - }; - - let user = TEST_USER_ID; - - let Ok(ListToolsGatewaySettings { handle, gateway_url, expected_tool_names, .. }) = - create_gateway_with_four_counters(user, config).await - else { - panic!("Invalid configuration "); - }; - - let client = create_client(user); - let maybe_passed = assert_list_tools(gateway_url, client, expected_tool_names).await; - - handle.abort(); - if maybe_passed.is_ok() { - info!("Test passed"); - } else { - info!("Test NOT passed {maybe_passed:?}"); - panic!() - } - - Ok(()) -} - -#[tokio::test(flavor = "multi_thread", worker_threads = 1)] -#[test_log::test] -#[ignore = "Fan out list tools is not supported at the moment. This should be enabled in 2.x"] -async fn tls_lists_prefixed_backend_tools() -> Result<()> { - let provider = crypto::ring::default_provider(); - _ = provider.install_default(); - let gateway_port = create_ports(1)[0]; - let server_socket_addr: std::net::SocketAddr = - format!("127.0.0.1:{gateway_port}").parse().expect("This should work"); - - let config = Config { - upstream_connection_mode: Some(UpstreamConnectionMode::PlainTextOrTls), - tls_address: Some(server_socket_addr), - server_private_key: Some("../../assets/contextforgeCA/contextforge-server.key.pem".into()), - server_certificate: Some("../../assets/contextforgeCA/contextforge-server.cert.pem".into()), - upstream_trust_bundle: Some("../../assets/contextforgeCA/contextforge.intermediate.ca-chain.cert.pem".into()), - ..create_default_config() - }; - - let user = TEST_USER_ID; - - let Ok(ListToolsGatewaySettings { handle, gateway_url, expected_tool_names, .. }) = - create_tls_gateway_with_four_tls_counters(user, config).await - else { - panic!("Invalid configuration "); - }; - - let test_future: BoxFuture> = async { - let mut buf = Vec::new(); - File::open("../../assets/contextforgeCA/contextforge.intermediate.ca-chain.cert.pem")?.read_to_end(&mut buf)?; - let certificates = reqwest::Certificate::from_pem_bundle(&buf)?; - - let client = create_tls_client(user, certificates); - assert_list_tools(gateway_url, client, expected_tool_names).await - } - .boxed(); - - let maybe_passed = test_future.await; - - handle.abort(); - if maybe_passed.is_ok() { - info!("Test passed"); - } else { - info!("Test NOT passed {maybe_passed:?}"); - panic!() - } - - Ok(()) -} - -async fn assert_list_tools( - gateway_url: String, - client: reqwest::Client, - expected_tool_names: Vec, -) -> Result<()> { - info!("Seding request to {gateway_url}"); - - let running_service = connect_client(gateway_url, client).await?; - - let list_tools = running_service.list_tools(None).await; - let Ok(list_tools) = list_tools else { - let msg = format!("List tools returned error {list_tools:?}"); - warn!(msg); - return Err(msg.into()); - }; - - let mut names: Vec = list_tools.tools.iter().map(|t| t.name.to_string()).collect(); - names.sort(); - - info!("Tool names {names:#?}"); - if expected_tool_names != names { - warn!("Actual {names:#?} Expected {expected_tool_names:#?}"); - return Err("Expected tool names don't match actual".into()); - } - - Ok(()) -} diff --git a/crates/contextforge-data-plane-lib/tests/gateway_pagination.rs b/crates/contextforge-data-plane-lib/tests/gateway_pagination.rs deleted file mode 100644 index fa06f477..00000000 --- a/crates/contextforge-data-plane-lib/tests/gateway_pagination.rs +++ /dev/null @@ -1,197 +0,0 @@ -mod support; - -use std::{collections::HashMap, sync::Arc}; - -use contextforge_data_plane_apis::{ - User, - user_store::{BackendMCPGateway, UserConfig, VirtualHost}, -}; -use contextforge_data_plane_lib::{Config, Gateway, Result, UserConfigStore, UserConfigStoreType}; -use rmcp::{ - model::PaginatedRequestParams, - transport::{ - StreamableHttpServerConfig, StreamableHttpService, streamable_http_server::session::local::LocalSessionManager, - }, -}; -use tracing::warn; - -use support::{ - MemoryUserConfigStore, TEST_USER_ID, connect_client, create_client, create_ports, paginating_mock, plaintext_config, -}; - -/// Build a single-backend `BackendMCPGateway` pointed at `port`. -fn paginating_backend(port: u16) -> BackendMCPGateway { - BackendMCPGateway { - name: format!("backend-{port}"), - url: format!("http://127.0.0.1:{port}/mcp").parse().expect("valid url"), - mcp_protocol_version: rmcp::model::ProtocolVersion::V_2026_07_28, - passthrough_headers: Vec::new(), - add_headers: HashMap::new(), - remove_headers: Vec::new(), - tool_schemas: HashMap::new(), - completion: HashMap::new(), - } -} - -fn backend_id(port: u16) -> String { - format!("00000000-0000-0000-0000-{port:012}") -} - -/// Bind the TCP port for a backend; returns the ready listener. -/// Call this *before* `tokio::spawn` so the port is reserved before the test proceeds. -async fn bind_backend_port(port: u16) -> tokio::net::TcpListener { - tokio::net::TcpListener::bind(format!("127.0.0.1:{port}")).await.expect("bind backend") -} - -/// Start an axum MCP server on an already-bound listener serving a `PaginatingServer`. -async fn serve_paginating_backend(listener: tokio::net::TcpListener) { - let service = StreamableHttpService::new( - || Ok(paginating_mock::PaginatingServer), - LocalSessionManager::default().into(), - StreamableHttpServerConfig::default(), - ); - let router = axum::Router::new().route_service("/mcp", service); - axum::serve(listener, router).await.expect("backend server"); -} - -/// Boot the gateway with the given config and user config; return the gateway URL. -async fn start_gateway(config: Config, virtual_host_id: &str, user_config: UserConfig) -> String { - let store = MemoryUserConfigStore::default(); - store.set_config(&User::new(TEST_USER_ID), &user_config).await.expect("set config"); - - let address = config.address.expect("This should be set"); - let gateway_url = format!("http://{address}/contextforge-rs/servers/{virtual_host_id}/mcp"); - - let gateway = Gateway::builder() - .with_config(config) - .with_session_manager(Arc::new(LocalSessionManager::default())) - .with_user_config_store_type(UserConfigStoreType::Test(Arc::new(store))) - .with_authorization_service(Arc::new(support::auth::AlwaysAllowAuthorizatioService::new( - TEST_USER_ID.to_owned(), - ))) - .build(); - - tokio::spawn(async move { - let res = gateway.run_gateway().await; - warn!("Gateway exited {res:?}"); - }); - - gateway_url -} - -/// A paginating backend returns tools across two pages; the gateway must expose -/// all of them to the client without any items being silently dropped. -#[tokio::test(flavor = "multi_thread", worker_threads = 1)] -#[test_log::test] -#[ignore = "Fan out list tools is not supported at the moment. This should be enabled in 2.x"] -async fn single_backend_pagination_all_tools_reachable() -> Result<()> { - let ports = create_ports(2); - let (backend_port, gateway_port) = (ports[0], ports[1]); - let config = plaintext_config(gateway_port); - - let virtual_host_id = "22222222-2222-2222-2222-222222222222"; - let backends = HashMap::from([(backend_id(backend_port), paginating_backend(backend_port))]); - let user_config = UserConfig { - virtual_hosts: HashMap::from([( - virtual_host_id.to_owned(), - VirtualHost { - backends, - tools: HashMap::new(), - resources: HashMap::new(), - resource_templates: HashMap::new(), - prompts: HashMap::new(), - }, - )]), - }; - - let backend_listener = bind_backend_port(backend_port).await; - tokio::spawn(serve_paginating_backend(backend_listener)); - let gateway_url = start_gateway(config, virtual_host_id, user_config).await; - - let svc = connect_client(gateway_url, create_client(TEST_USER_ID)).await?; - - // Page 1 - let page1 = svc.list_tools(None).await.expect("page 1"); - let page1_names: Vec<&str> = page1.tools.iter().map(|t| t.name.as_ref()).collect(); - assert!(page1.next_cursor.is_some(), "page 1 must carry a next_cursor"); - assert_eq!(page1_names, ["tool_alpha", "tool_beta"]); - - // Page 2 - let cursor = page1.next_cursor.map(|c| PaginatedRequestParams::default().with_cursor(Some(c))); - let page2 = svc.list_tools(cursor).await.expect("page 2"); - let page2_names: Vec<&str> = page2.tools.iter().map(|t| t.name.as_ref()).collect(); - assert!(page2.next_cursor.is_none(), "page 2 must be the final page"); - assert_eq!(page2_names, ["tool_gamma"]); - - // All tools reachable with no duplication - let mut all_names = page1_names.clone(); - all_names.extend_from_slice(&page2_names); - all_names.sort_unstable(); - assert_eq!(all_names, paginating_mock::PaginatingServer::all_tool_names()); - - Ok(()) -} - -/// When one backend exhausts its pages, it must be excluded from the resume -/// request. Without the filter, the exhausted backend would be re-queried and -/// its tools would appear in every subsequent page as duplicates. -#[tokio::test(flavor = "multi_thread", worker_threads = 1)] -#[test_log::test] -#[ignore = "Fan out list tools is not supported at the moment. This should be enabled in 2.x"] -async fn multi_backend_exhausted_backend_not_requeried() -> Result<()> { - // Backend A: PaginatingServer (2 pages: 2 tools + 1 tool) - // Backend B: another PaginatingServer (same 2 pages, different backend ID) - // - // With 2 backends, tool names get the backend-ID prefix. - // Page 1: 2 tools from A page 1 + 2 tools from B page 1 = 4 total - // Page 2: 1 tool from A page 2 + 1 tool from B page 2 = 2 total - // If either backend were re-queried, its page-1 tools would reappear. - let ports = create_ports(3); - let (port_a, port_b, gateway_port) = (ports[0], ports[1], ports[2]); - let config = plaintext_config(gateway_port); - - let virtual_host_id = "33333333-3333-3333-3333-333333333333"; - let backends = HashMap::from([ - (backend_id(port_a), paginating_backend(port_a)), - (backend_id(port_b), paginating_backend(port_b)), - ]); - let user_config = UserConfig { - virtual_hosts: HashMap::from([( - virtual_host_id.to_owned(), - VirtualHost { - backends, - tools: HashMap::new(), - resources: HashMap::new(), - resource_templates: HashMap::new(), - prompts: HashMap::new(), - }, - )]), - }; - - let listener_a = bind_backend_port(port_a).await; - let listener_b = bind_backend_port(port_b).await; - tokio::spawn(serve_paginating_backend(listener_a)); - tokio::spawn(serve_paginating_backend(listener_b)); - let gateway_url = start_gateway(config, virtual_host_id, user_config).await; - - let svc = connect_client(gateway_url, create_client(TEST_USER_ID)).await?; - - // Page 1: both backends contribute their first page (2 tools each) - let page1 = svc.list_tools(None).await.expect("page 1"); - assert!(page1.next_cursor.is_some(), "page 1 must carry a next_cursor"); - assert_eq!(page1.tools.len(), 4, "page 1 should have 2 tools from each backend"); - - // Page 2: both backends contribute their second page (1 tool each) - let cursor = page1.next_cursor.map(|c| PaginatedRequestParams::default().with_cursor(Some(c))); - let page2 = svc.list_tools(cursor).await.expect("page 2"); - assert!(page2.next_cursor.is_none(), "page 2 must be the final page"); - assert_eq!(page2.tools.len(), 2, "page 2 should have 1 tool from each backend"); - - // Union has 6 unique tools, no duplicates - let mut all_names: Vec<_> = page1.tools.iter().chain(page2.tools.iter()).map(|t| t.name.clone()).collect(); - all_names.sort_unstable(); - all_names.dedup(); - assert_eq!(all_names.len(), page1.tools.len() + page2.tools.len(), "no duplicate tools across pages"); - - Ok(()) -} diff --git a/crates/contextforge-data-plane-lib/tests/gateway_prompts.rs b/crates/contextforge-data-plane-lib/tests/gateway_prompts.rs deleted file mode 100644 index 213df7e8..00000000 --- a/crates/contextforge-data-plane-lib/tests/gateway_prompts.rs +++ /dev/null @@ -1,117 +0,0 @@ -mod support; - -use contextforge_data_plane_lib::{Config, Result, UpstreamConnectionMode}; -use rmcp::model::GetPromptRequestParams; -use serde_json::json; -use tracing::{info, warn}; - -use support::{ - ListToolsGatewaySettings, TEST_USER_ID, connect_client, create_client, create_gateway_with_four_counters, - create_ports, -}; - -use crate::support::{connect_modern_client, create_default_config}; - -#[tokio::test(flavor = "multi_thread", worker_threads = 1)] -#[test_log::test] -#[ignore = "Fan out list tools is not supported at the moment. This should be enabled in 2.x"] -async fn plaintext_lists_prefixed_backend_prompts() -> Result<()> { - let gateway_port = create_ports(1)[0]; - - let config = Config { - address: Some(format!("127.0.0.1:{gateway_port}").parse().expect("This should work")), - upstream_connection_mode: Some(UpstreamConnectionMode::PlainTextOrTls), - ..create_default_config() - }; - - let user = TEST_USER_ID; - let Ok(ListToolsGatewaySettings { handle, gateway_url, expected_prompt_names, .. }) = - create_gateway_with_four_counters(user, config).await - else { - panic!("Invalid configuration "); - }; - - let client = create_client(user); - let maybe_passed = assert_list_prompts(gateway_url, client, expected_prompt_names).await; - - handle.abort(); - maybe_passed -} - -#[tokio::test(flavor = "multi_thread", worker_threads = 1)] -#[test_log::test] -async fn plaintext_gets_prompt_from_prefixed_backend_name() -> Result<()> { - let gateway_port = create_ports(1)[0]; - - let config = Config { - address: Some(format!("127.0.0.1:{gateway_port}").parse().expect("This should work")), - upstream_connection_mode: Some(UpstreamConnectionMode::PlainTextOrTls), - ..create_default_config() - }; - - let user = TEST_USER_ID; - let Ok(ListToolsGatewaySettings { handle, gateway_url, expected_prompt_names, .. }) = - create_gateway_with_four_counters(user, config).await - else { - panic!("Invalid configuration "); - }; - - let prompt_name = expected_prompt_names - .iter() - .find(|name| name.ends_with("-example_prompt")) - .expect("example prompt is registered") - .clone(); - - let client = create_client(user); - let maybe_passed = assert_get_prompt(gateway_url, client, prompt_name).await; - - handle.abort(); - maybe_passed -} - -async fn assert_list_prompts( - gateway_url: String, - client: reqwest::Client, - expected_prompt_names: Vec, -) -> Result<()> { - info!("Sending request to {gateway_url}"); - - let running_service = connect_client(gateway_url, client).await?; - - let list_prompts = running_service.list_prompts(None).await; - let Ok(list_prompts) = list_prompts else { - let msg = format!("List prompts returned error {list_prompts:?}"); - warn!(msg); - return Err(msg.into()); - }; - - let mut names: Vec = list_prompts.prompts.iter().map(|p| p.name.clone()).collect(); - names.sort(); - - if expected_prompt_names != names { - warn!("Actual {names:#?} Expected {expected_prompt_names:#?}"); - return Err("Expected prompt names don't match actual".into()); - } - - Ok(()) -} - -async fn assert_get_prompt(gateway_url: String, client: reqwest::Client, prompt_name: String) -> Result<()> { - let running_service = connect_modern_client(&gateway_url, client, support::modern_client_info()).await; - let mut arguments = serde_json::Map::new(); - arguments.insert("message".to_owned(), json!("hello from gateway")); - - let result = running_service.get_prompt(GetPromptRequestParams::new(prompt_name).with_arguments(arguments)).await?; - let Some(message) = result.messages.first() else { - return Err("Expected prompt message".into()); - }; - let Some(text) = message.content.as_text().map(|content| &content.text) else { - return Err("Expected text prompt message".into()); - }; - - if !text.contains("hello from gateway") { - return Err(format!("Expected routed prompt to include argument, got: {text}").into()); - } - - Ok(()) -} diff --git a/crates/contextforge-data-plane-lib/tests/gateway_resource_read.rs b/crates/contextforge-data-plane-lib/tests/gateway_resource_read.rs deleted file mode 100644 index b02b8515..00000000 --- a/crates/contextforge-data-plane-lib/tests/gateway_resource_read.rs +++ /dev/null @@ -1,223 +0,0 @@ -mod support; - -use contextforge_data_plane_lib::{Config, Result, UpstreamConnectionMode}; -use rmcp::model::{ProtocolVersion, ReadResourceRequestParams}; -use tracing::{info, warn}; - -use support::{ListToolsGatewaySettings, TEST_USER_ID, create_client, create_gateway_with_four_counters, create_ports}; - -use crate::support::{ - connect_client_with_protocol, connect_modern_client, create_default_config, - create_gateway_with_four_legacy_counters, -}; - -#[tokio::test(flavor = "multi_thread", worker_threads = 1)] -#[test_log::test] -async fn plaintext_call_prefixed_read_resources_modern_modern() -> Result<()> { - let gateway_port = create_ports(1)[0]; - - let config = Config { - address: Some(format!("127.0.0.1:{gateway_port}").parse().expect("This should work")), - upstream_connection_mode: Some(UpstreamConnectionMode::PlainTextOrTls), - ..create_default_config() - }; - - let user = TEST_USER_ID; - - let Ok(ListToolsGatewaySettings { handle, gateway_url, expected_resource_uris, .. }) = - create_gateway_with_four_counters(user, config).await - else { - panic!("Invalid configuration "); - }; - - let client = create_client(user); - - let call_params = ReadResourceRequestParams::new(expected_resource_uris.first().expect("should work")); - - let maybe_passed = assert_resource_read( - gateway_url, - client, - call_params, - ProtocolVersion::V_2026_07_28, - "Business Intelligence Memo\n\nAnalysis has revealed 5 key insights ...".to_owned(), - ) - .await; - - handle.abort(); - if maybe_passed.is_ok() { - info!("Test passed"); - } else { - info!("Test NOT passed {maybe_passed:?}"); - panic!() - } - - Ok(()) -} - -#[tokio::test(flavor = "multi_thread", worker_threads = 1)] -#[test_log::test] -async fn plaintext_call_prefixed_read_resources_modern_legacy() -> Result<()> { - let gateway_port = create_ports(1)[0]; - - let config = Config { - address: Some(format!("127.0.0.1:{gateway_port}").parse().expect("This should work")), - upstream_connection_mode: Some(UpstreamConnectionMode::PlainTextOrTls), - ..create_default_config() - }; - - let user = TEST_USER_ID; - - let Ok(ListToolsGatewaySettings { handle, gateway_url, expected_resource_uris, .. }) = - create_gateway_with_four_legacy_counters(user, config).await - else { - panic!("Invalid configuration "); - }; - - let client = create_client(user); - - let call_params = ReadResourceRequestParams::new(expected_resource_uris.first().expect("should work")); - - let maybe_passed = assert_resource_read( - gateway_url, - client, - call_params, - ProtocolVersion::V_2026_07_28, - "Business Intelligence Memo\n\nAnalysis has revealed 5 key insights ...".to_owned(), - ) - .await; - - handle.abort(); - if maybe_passed.is_ok() { - info!("Test passed"); - } else { - info!("Test NOT passed {maybe_passed:?}"); - panic!() - } - - Ok(()) -} - -#[tokio::test(flavor = "multi_thread", worker_threads = 1)] -#[test_log::test] -async fn plaintext_call_prefixed_read_resources_legacy_modern() -> Result<()> { - let gateway_port = create_ports(1)[0]; - - let config = Config { - address: Some(format!("127.0.0.1:{gateway_port}").parse().expect("This should work")), - upstream_connection_mode: Some(UpstreamConnectionMode::PlainTextOrTls), - ..create_default_config() - }; - - let user = TEST_USER_ID; - - let Ok(ListToolsGatewaySettings { handle, gateway_url, expected_resource_uris, .. }) = - create_gateway_with_four_counters(user, config).await - else { - panic!("Invalid configuration "); - }; - - let client = create_client(user); - - let call_params = ReadResourceRequestParams::new(expected_resource_uris.first().expect("should work")); - - let maybe_passed = assert_resource_read( - gateway_url, - client, - call_params, - ProtocolVersion::V_2025_11_25, - "Business Intelligence Memo\n\nAnalysis has revealed 5 key insights ...".to_owned(), - ) - .await; - - handle.abort(); - if maybe_passed.is_ok() { - info!("Test passed"); - } else { - info!("Test NOT passed {maybe_passed:?}"); - panic!() - } - - Ok(()) -} - -async fn assert_resource_read( - gateway_url: String, - client: reqwest::Client, - params: ReadResourceRequestParams, - protocol_version: ProtocolVersion, - expected_result: String, -) -> Result<()> { - info!("Seding request to {gateway_url}"); - - let running_service = if protocol_version == ProtocolVersion::V_2026_07_28 { - connect_modern_client(&gateway_url, client, support::modern_client_info()).await - } else { - connect_client_with_protocol(gateway_url, client, protocol_version).await? - }; - - let response = running_service.read_resource(params).await; - let Ok(response) = response else { - let msg = format!("Request returned error {response:?}"); - warn!(msg); - return Err(msg.into()); - }; - - if response.contents.is_empty() { - let msg = format!("Request returned empty response {response:?}"); - warn!(msg); - return Err(msg.into()); - } - - if let Some(response) = response.contents.first() { - if let rmcp::model::ResourceContents::TextResourceContents { text, .. } = response { - if text != &expected_result { - let msg = format!("Request returned invalid response {text} {expected_result} {response:?}"); - warn!(msg); - return Err(msg.into()); - } - } else { - let msg = format!("Request returned non text response {response:?}"); - warn!(msg); - return Err(msg.into()); - } - } else { - let msg = format!("Request returned empty response {response:?}"); - warn!(msg); - return Err(msg.into()); - } - - Ok(()) -} - -#[tokio::test(flavor = "multi_thread", worker_threads = 1)] -#[test_log::test] -async fn plaintext_call_invalid_backend_tools() -> Result<()> { - let gateway_port = create_ports(1)[0]; - - let config = Config { - address: Some(format!("127.0.0.1:{gateway_port}").parse().expect("This should work")), - upstream_connection_mode: Some(UpstreamConnectionMode::PlainTextOrTls), - ..create_default_config() - }; - - let user = TEST_USER_ID; - - let Ok(ListToolsGatewaySettings { handle, gateway_url, .. }) = - create_gateway_with_four_counters(user, config).await - else { - panic!("Invalid configuration "); - }; - - let client = create_client(user); - - let call_params = ReadResourceRequestParams::new("http://dummy.dummy"); - let maybe_passed = - assert_resource_read(gateway_url, client, call_params, ProtocolVersion::V_2026_07_28, "-1".to_owned()).await; - - handle.abort(); - if maybe_passed.is_ok() { - info!("Test NOT passed {maybe_passed:?}"); - panic!() - } - Ok(()) -} diff --git a/crates/contextforge-data-plane-lib/tests/gateway_resource_templates.rs b/crates/contextforge-data-plane-lib/tests/gateway_resource_templates.rs deleted file mode 100644 index d99ceec4..00000000 --- a/crates/contextforge-data-plane-lib/tests/gateway_resource_templates.rs +++ /dev/null @@ -1,135 +0,0 @@ -mod support; - -use contextforge_data_plane_lib::{Config, Result, UpstreamConnectionMode}; -use rmcp::model::{ReadResourceRequestParams, ResourceContents}; -use tracing::{info, warn}; - -use support::{ - ListToolsGatewaySettings, TEST_USER_ID, connect_client, create_client, create_gateway_with_four_counters, - create_ports, -}; - -use crate::support::create_default_config; - -#[tokio::test(flavor = "multi_thread", worker_threads = 1)] -#[test_log::test] -#[ignore = "Fan out list resources is not supported at the moment. This should be enabled in 2.x"] -async fn plaintext_lists_prefixed_backend_resource_templates() -> Result<()> { - let gateway_port = create_ports(1)[0]; - - let config = Config { - address: Some(format!("127.0.0.1:{gateway_port}").parse().expect("This should work")), - upstream_connection_mode: Some(UpstreamConnectionMode::PlainTextOrTls), - ..create_default_config() - }; - - let user = TEST_USER_ID; - let Ok(ListToolsGatewaySettings { - handle, - gateway_url, - expected_resource_template_names, - expected_resource_template_uris, - .. - }) = create_gateway_with_four_counters(user, config).await - else { - panic!("Invalid configuration "); - }; - - let client = create_client(user); - let maybe_passed = assert_list_resource_templates( - gateway_url, - client, - expected_resource_template_names, - expected_resource_template_uris, - ) - .await; - - handle.abort(); - maybe_passed -} - -#[tokio::test(flavor = "multi_thread", worker_threads = 1)] -#[test_log::test] -#[ignore = "Fan out list resources is not supported at the moment. This should be enabled in 2.x"] -async fn plaintext_reads_resource_from_prefixed_template() -> Result<()> { - let gateway_port = create_ports(1)[0]; - - let config = Config { - address: Some(format!("127.0.0.1:{gateway_port}").parse().expect("This should work")), - upstream_connection_mode: Some(UpstreamConnectionMode::PlainTextOrTls), - ..create_default_config() - }; - - let user = TEST_USER_ID; - let Ok(ListToolsGatewaySettings { handle, gateway_url, .. }) = - create_gateway_with_four_counters(user, config).await - else { - panic!("Invalid configuration "); - }; - - let client = create_client(user); - let maybe_passed = assert_read_resource_from_template(gateway_url, client).await; - - handle.abort(); - maybe_passed -} - -async fn assert_list_resource_templates( - gateway_url: String, - client: reqwest::Client, - expected_template_names: Vec, - expected_template_uris: Vec, -) -> Result<()> { - info!("Sending request to {gateway_url}"); - - let running_service = connect_client(gateway_url, client).await?; - - let list_templates = running_service.list_resource_templates(None).await; - let Ok(list_templates) = list_templates else { - let msg = format!("List resource templates returned error {list_templates:?}"); - warn!(msg); - return Err(msg.into()); - }; - - let mut names: Vec = list_templates.resource_templates.iter().map(|t| t.name.clone()).collect(); - names.sort(); - - if expected_template_names != names { - warn!("Actual {names:#?} Expected {expected_template_names:#?}"); - return Err("Expected resource template names don't match actual".into()); - } - - let mut uris: Vec = list_templates.resource_templates.iter().map(|t| t.uri_template.clone()).collect(); - uris.sort(); - - if expected_template_uris != uris { - warn!("Actual {uris:#?} Expected {expected_template_uris:#?}"); - return Err("Expected resource template uris don't match actual".into()); - } - - Ok(()) -} - -async fn assert_read_resource_from_template(gateway_url: String, client: reqwest::Client) -> Result<()> { - let running_service = connect_client(gateway_url, client).await?; - - let list_templates = running_service.list_resource_templates(None).await?; - let Some(memo_template) = list_templates.resource_templates.iter().find(|t| t.uri_template.contains("memo://")) - else { - return Err("Expected a memo resource template".into()); - }; - - // Expand the namespaced template the way a client would, then read it back through the gateway. - let uri = memo_template.uri_template.replace("{id}", "insights"); - let result = running_service.read_resource(ReadResourceRequestParams::new(uri)).await?; - - let Some(ResourceContents::TextResourceContents { text, .. }) = result.contents.first() else { - return Err("Expected text resource contents".into()); - }; - - if !text.contains("Business Intelligence Memo") { - return Err(format!("Expected routed resource to include memo content, got: {text}").into()); - } - - Ok(()) -} diff --git a/crates/contextforge-data-plane-lib/tests/gateway_subscriptions.rs b/crates/contextforge-data-plane-lib/tests/gateway_subscriptions.rs deleted file mode 100644 index a73d8e6f..00000000 --- a/crates/contextforge-data-plane-lib/tests/gateway_subscriptions.rs +++ /dev/null @@ -1,203 +0,0 @@ -mod support; - -use std::{ - collections::HashSet, - sync::{Arc, Mutex as StdMutex}, - time::Instant, -}; - -use contextforge_data_plane_lib::Result; -use futures::future::try_join_all; -use rmcp::{ - ClientHandler, - model::{ - ClientCapabilities, Implementation, InitializeRequestParams, ResourceUpdatedNotificationParam, - SubscribeRequestParams, UnsubscribeRequestParams, - }, - service::{NotificationContext, RoleClient}, -}; - -use support::{ - CLIENT_CONNECT_TIMEOUT, ListToolsGatewaySettings, TEST_POLL_INTERVAL, TEST_USER_ID, connect_client_with_handler, - create_client, create_gateway_with_four_counters, create_ports, - mock_counter::{KNOWN_RESOURCE_URIS, RESOURCE_UPDATE_NOTIFY_INTERVAL}, - plaintext_config, -}; - -use crate::support::connect_modern_client; - -/// The mocks notify continuously, so this is just the threshold proving delivery works. -const MIN_UPDATES_PER_BACKEND: usize = 4; - -type Recorded = Arc>>; - -#[derive(Clone, Default)] -struct RecordingClient { - resource_updates: Recorded, -} - -impl ClientHandler for RecordingClient { - fn get_info(&self) -> InitializeRequestParams { - InitializeRequestParams::new( - ClientCapabilities::default(), - Implementation::new("resource-update-recording-test-client", "0.1.0"), - ) - } - - async fn on_resource_updated( - &self, - params: ResourceUpdatedNotificationParam, - _context: NotificationContext, - ) { - self.resource_updates.lock().expect("resource update lock poisoned").push(params); - } -} - -#[tokio::test(flavor = "multi_thread", worker_threads = 1)] -#[test_log::test] -#[ignore = "Fan out subscripbions are not supported at the moment. This should be enabled in 2.x"] -async fn plaintext_subscribes_and_unsubscribes_through_two_prefixed_backends() -> Result<()> { - let gateway_port = create_ports(1)[0]; - let user = TEST_USER_ID; - let Ok(ListToolsGatewaySettings { handle, gateway_url, .. }) = - create_gateway_with_four_counters(user, plaintext_config(gateway_port)).await - else { - panic!("Invalid configuration "); - }; - - let client = create_client(user); - let maybe_passed = assert_two_backend_subscribe_roundtrips(gateway_url, client).await; - - handle.abort(); - maybe_passed -} - -#[tokio::test(flavor = "multi_thread", worker_threads = 1)] -#[test_log::test] -async fn plaintext_subscribe_to_unrouted_resource_errors() -> Result<()> { - let gateway_port = create_ports(1)[0]; - let user = TEST_USER_ID; - let Ok(ListToolsGatewaySettings { handle, gateway_url, .. }) = - create_gateway_with_four_counters(user, plaintext_config(gateway_port)).await - else { - panic!("Invalid configuration "); - }; - - let client = create_client(user); - let maybe_passed = assert_unrouted_subscribe_errors(gateway_url, client).await; - - handle.abort(); - maybe_passed -} - -#[expect(deprecated, reason = "legacy RMCP coverage; modern subscriptions/listen tests are deferred")] -async fn assert_two_backend_subscribe_roundtrips(gateway_url: String, client: reqwest::Client) -> Result<()> { - let recording_client = RecordingClient::default(); - let resource_updates = Arc::clone(&recording_client.resource_updates); - let running_service = connect_client_with_handler(gateway_url, client, recording_client).await?; - - let resources = running_service.list_resources(None).await?; - let mut selected_backends = HashSet::new(); - let mut selected_uris = Vec::new(); - - for resource in resources.resources { - let backend_name = mock_backend_name(&resource.uri) - .ok_or_else(|| format!("expected mock backend-prefixed URI, got {}", resource.uri))?; - if selected_backends.insert(backend_name) { - selected_uris.push(resource.uri.clone()); - } - if selected_uris.len() == 2 { - break; - } - } - - if selected_uris.len() != 2 { - return Err(format!("expected resources from at least two backends, got {}", selected_uris.len()).into()); - } - - try_join_all(selected_uris.iter().map(|uri| running_service.subscribe(SubscribeRequestParams::new(uri.clone())))) - .await?; - wait_for_resource_updates(&resource_updates, &selected_uris, MIN_UPDATES_PER_BACKEND).await?; - for uri in selected_uris { - running_service.unsubscribe(UnsubscribeRequestParams::new(uri)).await?; - } - - assert_no_more_resource_updates(&resource_updates).await -} - -/// The mock backends keep notifying after unsubscribe, so any update recorded after the quiet -/// window starts would mean the gateway kept forwarding for an unsubscribed URI. -async fn assert_no_more_resource_updates( - resource_updates: &StdMutex>, -) -> Result<()> { - // Let updates the gateway forwarded before the unsubscribe finish arriving. - tokio::time::sleep(RESOURCE_UPDATE_NOTIFY_INTERVAL * 5).await; - let count_after_drain = resource_updates.lock().expect("resource update lock poisoned").len(); - - tokio::time::sleep(RESOURCE_UPDATE_NOTIFY_INTERVAL * 10).await; - let count_after_quiet = resource_updates.lock().expect("resource update lock poisoned").len(); - - if count_after_quiet != count_after_drain { - return Err(format!( - "expected no resource updates after unsubscribe, got {} new", - count_after_quiet - count_after_drain - ) - .into()); - } - Ok(()) -} - -#[expect(deprecated, reason = "legacy RMCP coverage; modern subscriptions/listen tests are deferred")] -async fn assert_unrouted_subscribe_errors(gateway_url: String, client: reqwest::Client) -> Result<()> { - let running_service = connect_modern_client(&gateway_url, client, support::modern_client_info()).await; - - // No backend namespace prefix => no route, so the gateway must reject it. - let result = running_service.subscribe(SubscribeRequestParams::new("unrouted://resource")).await; - if result.is_ok() { - return Err("expected a routing error for an unrouted resource URI".into()); - } - - Ok(()) -} - -async fn wait_for_resource_updates( - resource_updates: &StdMutex>, - expected_uris: &[String], - expected_count_per_uri: usize, -) -> Result<()> { - let deadline = Instant::now() + CLIENT_CONNECT_TIMEOUT; - - loop { - let counts = { - let updates = resource_updates.lock().expect("resource update lock poisoned"); - expected_uris - .iter() - .map(|uri| { - let count = updates.iter().filter(|update| update.uri == *uri).count(); - (uri.clone(), count) - }) - .collect::>() - }; - - if counts.iter().all(|(_, count)| *count >= expected_count_per_uri) { - return Ok(()); - } - if Instant::now() >= deadline { - return Err( - format!("expected {expected_count_per_uri} resource updates per URI, got counts {counts:?}").into() - ); - } - - tokio::time::sleep(TEST_POLL_INTERVAL).await; - } -} - -/// Extracts the backend prefix from a namespaced mock resource URI; the suffixes are the -/// backend-local URIs the mock owns, so this stays in lockstep with the mock's list. -fn mock_backend_name(uri: &str) -> Option { - KNOWN_RESOURCE_URIS - .iter() - .find_map(|known| uri.strip_suffix(known)) - .and_then(|prefix| prefix.strip_suffix('-')) - .map(str::to_owned) -} diff --git a/crates/contextforge-data-plane-lib/tests/support/test_gateways.rs b/crates/contextforge-data-plane-lib/tests/support/test_gateways.rs deleted file mode 100644 index b074bf83..00000000 --- a/crates/contextforge-data-plane-lib/tests/support/test_gateways.rs +++ /dev/null @@ -1,374 +0,0 @@ -use std::{collections::HashMap, sync::Arc}; - -use async_trait::async_trait; -use contextforge_data_plane_apis::{ - User, - user_store::{BackendMCPGateway, ServiceRoute, UserConfig, VirtualHost}, -}; -use contextforge_data_plane_lib::{ - AuthorizationClaims, AuthorizationService, Config, Gateway, Result, UpstreamConnectionMode, UserConfigStore, - UserConfigStoreType, -}; -use futures::{FutureExt, future::BoxFuture}; -use http::HeaderValue; -use rmcp::transport::{ - StreamableHttpServerConfig, StreamableHttpService, streamable_http_server::session::local::LocalSessionManager, -}; -use rustls::ProtocolVersion; -use tracing::warn; - -use crate::support::{self, create_default_config}; - -use super::{MemoryUserConfigStore, mock_counter}; - -const MOCK_COUNTER_TOOL_NAMES: &[&str] = - &["decrement", "echo", "get_session_id", "get_value", "increment", "long_task", "say_hello", "sum"]; -const MOCK_COUNTER_PROMPT_NAMES: &[&str] = &["counter_analysis", "example_prompt"]; -const MOCK_COUNTER_RESOURCE_TEMPLATE_NAMES: &[&str] = &["filesystem", "memo"]; -const MOCK_COUNTER_RESOURCE_TEMPLATE_URIS: &[&str] = &["memo://{id}", "str:////{path}"]; -const MOCK_COUNTER_RESOURCE_URIS: &[&str] = &["memo://insights", "str:////Users/to/some/path/"]; - -pub(crate) struct ListToolsGatewaySettings { - pub(crate) handle: tokio::task::JoinHandle>>, - pub(crate) gateway_url: String, - pub(crate) expected_tool_names: Vec, - pub(crate) expected_prompt_names: Vec, - pub(crate) expected_resource_template_names: Vec, - pub(crate) expected_resource_template_uris: Vec, - pub(crate) expected_resource_uris: Vec, -} - -/// Gateway config for plaintext-upstream tests, shared by the integration test binaries. -pub(crate) fn plaintext_config(gateway_port: u16) -> Config { - Config { - address: Some(format!("127.0.0.1:{gateway_port}").parse().expect("This should work")), - upstream_connection_mode: Some(UpstreamConnectionMode::PlainTextOrTls), - ..create_default_config() - } -} - -pub(crate) fn create_ports(ports: usize) -> Vec { - let mut selected = Vec::with_capacity(ports); - while selected.len() < ports { - let port = openport::pick_random_unused_port().expect("Expecting to find port"); - if !selected.contains(&port) { - selected.push(port); - } - } - selected -} - -pub fn construct_services(backend_name: &str, service_names: &[&str]) -> HashMap { - service_names - .iter() - .map(|&name| { - (name.to_owned(), ServiceRoute { backend_name: backend_name.to_owned(), upstream_name: name.to_owned() }) - }) - .collect() -} - -fn create_services_from_ports(ports: &[u16], service_names: &[&str]) -> HashMap { - let mut services = HashMap::new(); - - for &port in ports { - let backend_id = backend_id(port); - - for &service_name in service_names { - let key = format!("{backend_id}-{service_name}"); - let value = ServiceRoute { backend_name: backend_id.clone(), upstream_name: service_name.to_owned() }; - - services.insert(key, value); - } - } - services -} - -async fn create_gateway_with_four_counters_and_custom_config( - user: &str, - config: Config, - create_backends: impl Fn(&[u16]) -> HashMap, -) -> Result { - let mocked_user_config_store = MemoryUserConfigStore::default(); - - let config_address = config.address.expect("This must be set"); - let gateway_port = config_address.port(); - - let service = StreamableHttpService::new( - || Ok(mock_counter::Counter::new()), - LocalSessionManager::default().into(), - StreamableHttpServerConfig::default(), - ); - - let router = axum::Router::new().route_service("/mcp", service); - - let (gateway_one_ports, servers_one) = create_axum_servers(2, gateway_port, &router).await?; - let (gateway_two_ports, servers_two) = create_axum_servers(2, gateway_port, &router).await?; - - assert_ne!(gateway_one_ports, gateway_two_ports); - - let gateway_one_backends = create_backends(&gateway_one_ports); - let gateway_two_backends = create_backends(&gateway_two_ports); - - let mut virtual_host_one_tool_names = create_tool_names(&gateway_one_ports); - virtual_host_one_tool_names.sort(); - let mut virtual_host_one_prompt_names = create_prompt_names(&gateway_one_ports); - virtual_host_one_prompt_names.sort(); - let mut virtual_host_one_resource_template_names = create_resource_template_names(&gateway_one_ports); - virtual_host_one_resource_template_names.sort(); - let mut virtual_host_one_resource_template_uris = create_resource_template_uris(&gateway_one_ports); - virtual_host_one_resource_template_uris.sort(); - let mut virtual_host_one_resource_uris = create_resource_uris(&gateway_one_ports); - virtual_host_one_resource_uris.sort(); - - let gateway_one_tools = create_services_from_ports(&gateway_one_ports, MOCK_COUNTER_TOOL_NAMES); - let gateway_one_resources = create_services_from_ports(&gateway_one_ports, MOCK_COUNTER_RESOURCE_URIS); - let gateway_one_prompts = create_services_from_ports(&gateway_one_ports, MOCK_COUNTER_PROMPT_NAMES); - - let user_key = User::new(user); - - let virtual_host_one_id = uuid::Uuid::new_v4().to_string(); - let virtual_host_two_id = uuid::Uuid::new_v4().to_string(); - - let virtual_hosts = HashMap::from([ - ( - virtual_host_one_id.clone(), - VirtualHost { - backends: gateway_one_backends, - tools: gateway_one_tools, - resources: gateway_one_resources, - resource_templates: HashMap::new(), - prompts: gateway_one_prompts, - }, - ), - ( - virtual_host_two_id, - VirtualHost { - backends: gateway_two_backends, - tools: HashMap::new(), - resources: HashMap::new(), - resource_templates: HashMap::new(), - prompts: HashMap::new(), - }, - ), - ]); - - let user_config = UserConfig { virtual_hosts }; - - mocked_user_config_store.set_config(&user_key, &user_config).await.expect("This should work"); - - let gateway = Gateway::builder() - .with_config(config.clone()) - .with_session_manager(Arc::new(LocalSessionManager::default())) - .with_user_config_store_type(UserConfigStoreType::Test(Arc::new(mocked_user_config_store))) - .with_authorization_service(Arc::new(support::auth::AlwaysAllowAuthorizatioService::new(user.to_owned()))) - .build(); - - let gateway = async move { - let res = gateway.run_gateway().await; - warn!("Gateway exited with result {res:?}"); - Ok(()) - } - .boxed(); - - let address = config_address; - let gateway_url = format!("http://{address}/contextforge-rs/servers/{virtual_host_one_id}/mcp"); - - let handle = - tokio::spawn(futures::future::join_all(vec![gateway].into_iter().chain(servers_one).chain(servers_two))); - - Ok(ListToolsGatewaySettings { - handle, - gateway_url, - expected_tool_names: virtual_host_one_tool_names, - expected_prompt_names: virtual_host_one_prompt_names, - expected_resource_template_names: virtual_host_one_resource_template_names, - expected_resource_template_uris: virtual_host_one_resource_template_uris, - expected_resource_uris: virtual_host_one_resource_uris, - }) -} - -pub(crate) async fn create_gateway_with_four_counters(user: &str, config: Config) -> Result { - create_gateway_with_four_counters_and_custom_config(user, config, create_plain_backends).await -} - -pub(crate) async fn create_gateway_with_four_legacy_counters( - user: &str, - config: Config, -) -> Result { - create_gateway_with_four_counters_and_custom_config(user, config, create_plain_legacy_backends).await -} - -pub(crate) async fn create_tls_gateway_with_four_tls_counters( - user: &str, - config: Config, -) -> Result { - create_gateway_with_four_counters_and_custom_config(user, config, create_tls_backends).await -} - -fn create_backends( - ports: &[u16], - with_tls: bool, - protocol_version: &rmcp::model::ProtocolVersion, -) -> HashMap { - ports - .iter() - .map(|port| { - let url = if with_tls { - format!("https://127.0.0.1:{port}/mcp").parse().expect("This should work") - } else { - format!("http://127.0.0.1:{port}/mcp").parse().expect("This should work") - }; - - let backend_id = backend_id(*port); - ( - backend_id.clone(), - BackendMCPGateway { - name: format!("backend-{port}"), - url, - mcp_protocol_version: protocol_version.clone(), - passthrough_headers: Vec::new(), - add_headers: HashMap::default(), - remove_headers: Vec::new(), - tool_schemas: MOCK_COUNTER_TOOL_NAMES - .iter() - .map(|name| ((*name).to_owned(), serde_json::Map::new())) - .collect(), - - completion: HashMap::new(), - }, - ) - }) - .collect() -} - -fn create_plain_backends(ports: &[u16]) -> HashMap { - create_backends(ports, false, &rmcp::model::ProtocolVersion::V_2026_07_28) -} - -fn create_plain_legacy_backends(ports: &[u16]) -> HashMap { - create_backends(ports, false, &rmcp::model::ProtocolVersion::V_2025_11_25) -} - -fn create_tls_backends(ports: &[u16]) -> HashMap { - create_backends(ports, true, &rmcp::model::ProtocolVersion::V_2026_07_28) -} - -fn backend_id(port: u16) -> String { - format!("00000000-0000-0000-0000-{port:012}") -} - -fn create_tool_names(ports: &[u16]) -> Vec { - ports - .iter() - .flat_map(|port| { - let backend_id = backend_id(*port); - MOCK_COUNTER_TOOL_NAMES.iter().map(move |name| format!("{backend_id}-{name}")) - }) - .collect() -} - -fn create_prompt_names(ports: &[u16]) -> Vec { - ports - .iter() - .flat_map(|port| { - let backend_id = backend_id(*port); - MOCK_COUNTER_PROMPT_NAMES.iter().map(move |name| format!("{backend_id}-{name}")) - }) - .collect() -} - -fn create_resource_template_names(ports: &[u16]) -> Vec { - ports - .iter() - .flat_map(|port| { - let backend_id = backend_id(*port); - MOCK_COUNTER_RESOURCE_TEMPLATE_NAMES.iter().map(move |name| format!("{backend_id}-{name}")) - }) - .collect() -} - -fn create_resource_template_uris(ports: &[u16]) -> Vec { - ports - .iter() - .flat_map(|port| { - let backend_id = backend_id(*port); - MOCK_COUNTER_RESOURCE_TEMPLATE_URIS.iter().map(move |uri| format!("{backend_id}-{uri}")) - }) - .collect() -} - -fn create_resource_uris(ports: &[u16]) -> Vec { - ports - .iter() - .flat_map(|port| { - let backend_id = backend_id(*port); - MOCK_COUNTER_RESOURCE_URIS.iter().map(move |uri| format!("{backend_id}-{uri}")) - }) - .collect() -} - -async fn create_axum_servers( - server_count: usize, - gateway_port: u16, - router: &axum::Router, -) -> Result<(Vec, Vec>>)> { - let mut ports = Vec::with_capacity(server_count); - let mut servers = Vec::with_capacity(server_count); - - while ports.len() < server_count { - let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await?; - let port = listener.local_addr()?.port(); - if port == gateway_port { - continue; - } - - let router = router.clone(); - ports.push(port); - servers.push( - async move { - axum::serve(listener, router).await?; - Ok(()) - } - .boxed(), - ); - } - - Ok((ports, servers)) -} - -async fn create_axum_tls_servers( - server_count: usize, - gateway_port: u16, - router: axum::Router, -) -> Result<(Vec, Vec>>)> { - let config = axum_server::tls_rustls::RustlsConfig::from_pem_file( - "../../assets/contextforgeCA/contextforge-server.cert.pem", - "../../assets/contextforgeCA/contextforge-server.key.pem", - ) - .await - .expect("Expect this to work"); - let mut ports = Vec::with_capacity(server_count); - let mut servers = Vec::with_capacity(server_count); - - while ports.len() < server_count { - let listener = std::net::TcpListener::bind("127.0.0.1:0")?; - let port = listener.local_addr()?.port(); - if port == gateway_port { - continue; - } - - listener.set_nonblocking(true)?; - let server = axum_server::from_tcp_rustls(listener, config.clone())?; - let router = router.clone(); - ports.push(port); - servers.push( - async move { - server.serve(router.into_make_service()).await?; - Ok(()) - } - .boxed(), - ); - } - - Ok((ports, servers)) -}