test: use uuid dataplane subject fixtures

README.md

@@ -25,16 +25,18 @@ This should spin up Redis instance and two mcp-gateways: a simple counter and a
 
 3. Get a test JWT token
 ```bash
+USER_ID=11111111-1111-1111-1111-111111111111
+
 curl --request GET \
-  --url http://127.0.0.1:8001/contextforge-rs/admin/tokens/admin@example.com \
+  --url "http://127.0.0.1:8001/contextforge-rs/admin/tokens/${USER_ID}" \
   --header 'accept: application/json' \  
   --header 'content-type: application/json'
 ```
 
 4. Use the token to add a test user to Redis
 ```bash
 curl --request POST \
-  --url http://127.0.0.1:8001/contextforge-rs/admin/userconfigs/admin@example.com \
+  --url "http://127.0.0.1:8001/contextforge-rs/admin/userconfigs/${USER_ID}" \
   --header 'authorization: Bearer {{token}}' \
   --header 'content-type: application/json' \
   --data '{
@@ -168,8 +170,10 @@ cargo run --release --features test-plugins --bin contextforge-gateway-rs -- \
 Get a token:
 
 ```bash
+USER_ID=11111111-1111-1111-1111-111111111111
+
 TOKEN=$(curl --silent --show-error --request GET \
-  --url http://127.0.0.1:8001/contextforge-rs/admin/tokens/admin@example.com \
+  --url "http://127.0.0.1:8001/contextforge-rs/admin/tokens/${USER_ID}" \
   --header 'accept: application/json' \
   --header 'content-type: application/json')
 ```
@@ -178,7 +182,7 @@ Create the gateway user config:
 
 ```bash
 curl --silent --show-error --request POST \
-  --url http://127.0.0.1:8001/contextforge-rs/admin/userconfigs/admin@example.com \
+  --url "http://127.0.0.1:8001/contextforge-rs/admin/userconfigs/${USER_ID}" \
   --header "authorization: Bearer ${TOKEN}" \
   --header 'content-type: application/json' \
   --data '{

crates/contextforge-gateway-rs-lib/src/layers/claims_id.rs

@@ -102,7 +102,8 @@ mod test {
 
     fn active_test_claims() -> ContextForgeClaims {
         let now = now_epoch_seconds();
-        let user_id = "admin@example.com".to_owned();
+        let user_id = "11111111-1111-1111-1111-111111111111".to_owned();
+        let user_email = "admin@example.com".to_owned();
 
         ContextForgeClaims {
             iss: CONTEXT_FORGE_GATEWAY_ISSUER.to_owned(),
@@ -114,7 +115,7 @@ mod test {
             token_use: Some("api".to_owned()),
             teams: Some(vec!["team_awesome".to_owned()]),
             user: common::User::builder()
-                .email(user_id)
+                .email(user_email)
                 .auth_provider("api_token".to_owned())
                 .full_name(Some("API Token User".to_owned()))
                 .is_admin(true)
@@ -228,11 +229,11 @@ mod test {
             Response::builder().status(StatusCode::OK).body(Body::empty()).expect("Expecting this to work")
         }
 
-        let user_id = "admin@example.com".to_owned();
+        let user_email = "admin@example.com".to_owned();
         let mut claims = active_test_claims();
         claims.token_use = None;
         claims.user = common::User::builder()
-            .email(user_id)
+            .email(user_email)
             .auth_provider("local".to_owned())
             .full_name(None)
             .is_admin(true)

crates/contextforge-gateway-rs-lib/tests/gateway_completions.rs

@@ -4,15 +4,15 @@ use contextforge_gateway_rs_lib::Result;
 use tracing::info;
 
 use support::{
-    ListToolsGatewaySettings, connect_client, create_client, create_gateway_with_four_counters, create_ports,
-    plaintext_config,
+    ListToolsGatewaySettings, TEST_USER_ID, connect_client, create_client, create_gateway_with_four_counters,
+    create_ports, plaintext_config,
 };
 
 #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
 #[test_log::test]
 async fn plaintext_completes_prompt_argument_through_prefixed_backend() -> Result<()> {
     let gateway_port = create_ports(1)[0];
-    let user = "admin@example.com";
+    let user = TEST_USER_ID;
     let Ok(ListToolsGatewaySettings { handle, gateway_url, .. }) =
         create_gateway_with_four_counters(user, plaintext_config(gateway_port)).await
     else {
@@ -30,7 +30,7 @@ async fn plaintext_completes_prompt_argument_through_prefixed_backend() -> Resul
 #[test_log::test]
 async fn plaintext_completes_resource_argument_through_prefixed_backend() -> Result<()> {
     let gateway_port = create_ports(1)[0];
-    let user = "admin@example.com";
+    let user = TEST_USER_ID;
     let Ok(ListToolsGatewaySettings { handle, gateway_url, .. }) =
         create_gateway_with_four_counters(user, plaintext_config(gateway_port)).await
     else {
@@ -48,7 +48,7 @@ async fn plaintext_completes_resource_argument_through_prefixed_backend() -> Res
 #[test_log::test]
 async fn plaintext_complete_for_unrouted_reference_errors() -> Result<()> {
     let gateway_port = create_ports(1)[0];
-    let user = "admin@example.com";
+    let user = TEST_USER_ID;
     let Ok(ListToolsGatewaySettings { handle, gateway_url, .. }) =
         create_gateway_with_four_counters(user, plaintext_config(gateway_port)).await
     else {

crates/contextforge-gateway-rs-lib/tests/gateway_list_tools.rs

@@ -8,8 +8,8 @@ use rustls::crypto;
 use tracing::{info, warn};
 
 use support::{
-    ListToolsGatewaySettings, connect_client, create_client, create_gateway_with_four_counters, create_ports,
-    create_tls_client, create_tls_gateway_with_four_tls_counters,
+    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,
 };
 
 #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
@@ -24,7 +24,7 @@ async fn plaintext_lists_prefixed_backend_tools() -> Result<()> {
         ..Default::default()
     };
 
-    let user = "admin@example.com";
+    let user = TEST_USER_ID;
 
     let Ok(ListToolsGatewaySettings { handle, gateway_url, expected_tool_names, .. }) =
         create_gateway_with_four_counters(user, config).await
@@ -65,7 +65,7 @@ async fn tls_lists_prefixed_backend_tools() -> Result<()> {
         ..Default::default()
     };
 
-    let user = "admin@example.com";
+    let user = TEST_USER_ID;
 
     let Ok(ListToolsGatewaySettings { handle, gateway_url, expected_tool_names, .. }) =
         create_tls_gateway_with_four_tls_counters(user, config).await

crates/contextforge-gateway-rs-lib/tests/gateway_plugins.rs

@@ -16,8 +16,8 @@ use rmcp::{
 use serde_json::Value;
 
 use support::{
-    POST_DENY_ERROR_CODE, PRE_DENY_ERROR_CODE, REWRITTEN_SUM_A, REWRITTEN_SUM_B, RunningGateway, TestPlugin,
-    error_code, runtime_with_post, runtime_with_pre, runtime_with_pre_and_post, start_gateway,
+    POST_DENY_ERROR_CODE, PRE_DENY_ERROR_CODE, REWRITTEN_SUM_A, REWRITTEN_SUM_B, RunningGateway, TEST_USER_ID,
+    TestPlugin, error_code, runtime_with_post, runtime_with_pre, runtime_with_pre_and_post, start_gateway,
     start_gateway_with_json_backend_responses, sum_request, text, token,
 };
 
@@ -227,10 +227,10 @@ async fn read_concurrent_raw_progress_streams(
 
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
 async fn concurrent_progress_calls_forward_each_token_without_plugins() {
-    let gateway = start_gateway("admin@example.com", false, Arc::new(CpexRuntimeRegistry::default())).await;
+    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 = gateway.connect_with_handler("admin@example.com", client).await;
+    let service = gateway.connect_with_handler(TEST_USER_ID, client).await;
     let tool_name = "progress_sum";
 
     let first = send_progress_call(&service, tool_name).await;
@@ -266,22 +266,22 @@ async fn concurrent_progress_calls_forward_each_token_without_plugins() {
 
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
 async fn raw_streamable_http_concurrent_progress_calls_complete_without_plugins() {
-    let gateway = start_gateway("admin@example.com", false, Arc::new(CpexRuntimeRegistry::default())).await;
+    let gateway = start_gateway(TEST_USER_ID, false, Arc::new(CpexRuntimeRegistry::default())).await;
     let client = reqwest::Client::new();
-    let session_id = start_raw_mcp_session(&client, &gateway, "admin@example.com").await;
+    let session_id = start_raw_mcp_session(&client, &gateway, TEST_USER_ID).await;
 
     let tool_name = "progress_sum";
     let first = raw_mcp_request(
         &client,
         &gateway,
-        "admin@example.com",
+        TEST_USER_ID,
         Some(&session_id),
         &raw_tool_call(tool_name, 2, "downstream-first"),
     );
     let second = raw_mcp_request(
         &client,
         &gateway,
-        "admin@example.com",
+        TEST_USER_ID,
         Some(&session_id),
         &raw_tool_call(tool_name, 3, "downstream-second"),
     );
@@ -293,10 +293,10 @@ async fn raw_streamable_http_concurrent_progress_calls_complete_without_plugins(
 
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
 async fn backend_generated_progress_tokens_are_dropped() {
-    let gateway = start_gateway("admin@example.com", false, Arc::new(CpexRuntimeRegistry::default())).await;
+    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 = gateway.connect_with_handler("admin@example.com", client).await;
+    let service = gateway.connect_with_handler(TEST_USER_ID, client).await;
 
     let tool_name = "progress_counter_tokens";
     let handle = send_progress_call(&service, tool_name).await;
@@ -324,8 +324,8 @@ async fn disabled_runtime_does_not_invoke_registered_plugin() {
     let post_observations = post_plugin.observations();
     let runtime = runtime_with_pre_and_post(pre_plugin, post_plugin).await;
 
-    let gateway = start_gateway("admin@example.com", false, runtime).await;
-    let service = gateway.connect("admin@example.com").await;
+    let gateway = start_gateway(TEST_USER_ID, false, runtime).await;
+    let service = gateway.connect(TEST_USER_ID).await;
     let result = service.call_tool(sum_request("sum", 1, 2)).await.unwrap();
 
     assert_eq!("3", text(&result));
@@ -339,8 +339,8 @@ async fn pre_hook_modifies_backend_arguments_without_rerouting_tool() {
     let observations = plugin.observations();
     let runtime = runtime_with_pre(plugin).await;
 
-    let gateway = start_gateway("admin@example.com", true, runtime).await;
-    let service = gateway.connect("admin@example.com").await;
+    let gateway = start_gateway(TEST_USER_ID, true, runtime).await;
+    let service = gateway.connect(TEST_USER_ID).await;
     let result = service.call_tool(sum_request("sum", 1, 2)).await.unwrap();
 
     assert_eq!((REWRITTEN_SUM_A + REWRITTEN_SUM_B).to_string(), text(&result));
@@ -361,8 +361,8 @@ async fn post_hook_receives_backend_result_and_modifies_client_result() {
     let observations = plugin.observations();
     let runtime = runtime_with_post(plugin).await;
 
-    let gateway = start_gateway("admin@example.com", true, runtime).await;
-    let service = gateway.connect("admin@example.com").await;
+    let gateway = start_gateway(TEST_USER_ID, true, runtime).await;
+    let service = gateway.connect(TEST_USER_ID).await;
     let result = service.call_tool(sum_request("sum", 1, 2)).await.unwrap();
 
     assert_eq!("post:3", text(&result));
@@ -379,8 +379,8 @@ async fn post_hook_can_modify_stream_progress_notifications() {
     let observations = plugin.observations();
     let runtime = runtime_with_post(plugin).await;
 
-    let gateway = start_gateway("admin@example.com", true, runtime).await;
-    let (result, progress) = call_progress_sum(&gateway, "admin@example.com").await;
+    let gateway = start_gateway(TEST_USER_ID, true, runtime).await;
+    let (result, progress) = call_progress_sum(&gateway, TEST_USER_ID).await;
 
     assert_eq!("completed 4 packages", text(&result));
     let progress = progress.lock().expect("progress lock poisoned");
@@ -398,8 +398,8 @@ async fn json_response_mode_forwards_backend_progress_notifications() {
     let plugin = Arc::new(TestPlugin::new("post", vec![cmf_hook_names::TOOL_POST_INVOKE]).with_post_rewrite());
     let runtime = runtime_with_post(plugin).await;
 
-    let gateway = start_gateway_with_json_backend_responses("admin@example.com", true, runtime).await;
-    let (result, progress) = call_progress_sum(&gateway, "admin@example.com").await;
+    let gateway = start_gateway_with_json_backend_responses(TEST_USER_ID, true, runtime).await;
+    let (result, progress) = call_progress_sum(&gateway, TEST_USER_ID).await;
 
     assert_eq!("post:completed 4 packages", text(&result));
     assert_eq!(4, progress.lock().expect("progress lock poisoned").len());
@@ -412,8 +412,8 @@ async fn post_hook_deny_drops_progress_notifications_without_failing_call() {
     let observations = plugin.observations();
     let runtime = runtime_with_post(plugin).await;
 
-    let gateway = start_gateway("admin@example.com", true, runtime).await;
-    let (result, progress) = send_progress_sum(&gateway, "admin@example.com").await;
+    let gateway = start_gateway(TEST_USER_ID, true, runtime).await;
+    let (result, progress) = send_progress_sum(&gateway, TEST_USER_ID).await;
 
     assert_eq!("completed 4 packages", text(&result));
     // four denied progress notifications plus the final tool result
@@ -430,8 +430,8 @@ async fn post_hook_deny_drops_progress_notifications_without_failing_call() {
 
 #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
 async fn downstream_cancellation_is_relayed_to_backend() {
-    let gateway = start_gateway("admin@example.com", true, Arc::new(CpexRuntimeRegistry::default())).await;
-    let service = gateway.connect("admin@example.com").await;
+    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
@@ -452,8 +452,8 @@ 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());
     let runtime = runtime_with_post(plugin).await;
 
-    let gateway = start_gateway("admin@example.com", true, runtime).await;
-    let service = gateway.connect("admin@example.com").await;
+    let gateway = start_gateway(TEST_USER_ID, true, runtime).await;
+    let service = gateway.connect(TEST_USER_ID).await;
     let result = service.call_tool(sum_request("sum", 1, 2)).await.unwrap();
 
     assert_eq!("raw-post", text(&result));
@@ -468,8 +468,8 @@ async fn pre_and_post_hooks_share_gateway_call_context() {
     let post_observations = post_plugin.observations();
     let runtime = runtime_with_pre_and_post(pre_plugin, post_plugin).await;
 
-    let gateway = start_gateway("admin@example.com", true, runtime).await;
-    let service = gateway.connect("admin@example.com").await;
+    let gateway = start_gateway(TEST_USER_ID, true, runtime).await;
+    let service = gateway.connect(TEST_USER_ID).await;
     let result = service.call_tool(sum_request("sum", 1, 2)).await.unwrap();
 
     assert_eq!("3", text(&result));
@@ -480,16 +480,16 @@ async fn pre_and_post_hooks_share_gateway_call_context() {
 async fn pre_and_post_denials_return_plugin_error_codes() {
     let pre_plugin = Arc::new(TestPlugin::new("pre-deny", vec![cmf_hook_names::TOOL_PRE_INVOKE]).with_pre_deny());
     let runtime = runtime_with_pre(pre_plugin).await;
-    let gateway = start_gateway("admin@example.com", true, runtime).await;
-    let service = gateway.connect("admin@example.com").await;
+    let gateway = start_gateway(TEST_USER_ID, true, runtime).await;
+    let service = gateway.connect(TEST_USER_ID).await;
     let error = service.call_tool(sum_request("sum", 1, 2)).await.unwrap_err();
     assert_eq!(ErrorCode(PRE_DENY_ERROR_CODE), error_code(error));
     assert!(gateway.backend_state.calls.lock().expect("backend calls lock poisoned").is_empty());
 
     let post_plugin = Arc::new(TestPlugin::new("post-deny", vec![cmf_hook_names::TOOL_POST_INVOKE]).with_post_deny());
     let runtime = runtime_with_post(post_plugin).await;
-    let gateway = start_gateway("admin@example.com", true, runtime).await;
-    let service = gateway.connect("admin@example.com").await;
+    let gateway = start_gateway(TEST_USER_ID, true, runtime).await;
+    let service = gateway.connect(TEST_USER_ID).await;
     let error = service.call_tool(sum_request("sum", 1, 2)).await.unwrap_err();
     assert_eq!(ErrorCode(POST_DENY_ERROR_CODE), error_code(error));
     assert_eq!(1, gateway.backend_state.calls.lock().expect("backend calls lock poisoned").len());
@@ -500,8 +500,8 @@ async fn pre_hook_invalid_arguments_return_invalid_params() {
     let plugin =
         Arc::new(TestPlugin::new("invalid-args", vec![cmf_hook_names::TOOL_PRE_INVOKE]).with_invalid_pre_args());
     let runtime = runtime_with_pre(plugin).await;
-    let gateway = start_gateway("admin@example.com", true, runtime).await;
-    let service = gateway.connect("admin@example.com").await;
+    let gateway = start_gateway(TEST_USER_ID, true, runtime).await;
+    let service = gateway.connect(TEST_USER_ID).await;
     let error = service.call_tool(sum_request("sum", 1, 2)).await.unwrap_err();
 
     assert_eq!(ErrorCode::INVALID_PARAMS, error_code(error));

crates/contextforge-gateway-rs-lib/tests/gateway_prompts.rs

@@ -6,7 +6,8 @@ use serde_json::json;
 use tracing::{info, warn};
 
 use support::{
-    ListToolsGatewaySettings, connect_client, create_client, create_gateway_with_four_counters, create_ports,
+    ListToolsGatewaySettings, TEST_USER_ID, connect_client, create_client, create_gateway_with_four_counters,
+    create_ports,
 };
 
 #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
@@ -21,7 +22,7 @@ async fn plaintext_lists_prefixed_backend_prompts() -> Result<()> {
         ..Default::default()
     };
 
-    let user = "admin@example.com";
+    let user = TEST_USER_ID;
     let Ok(ListToolsGatewaySettings { handle, gateway_url, expected_prompt_names, .. }) =
         create_gateway_with_four_counters(user, config).await
     else {
@@ -47,7 +48,7 @@ async fn plaintext_gets_prompt_from_prefixed_backend_name() -> Result<()> {
         ..Default::default()
     };
 
-    let user = "admin@example.com";
+    let user = TEST_USER_ID;
     let Ok(ListToolsGatewaySettings { handle, gateway_url, expected_prompt_names, .. }) =
         create_gateway_with_four_counters(user, config).await
     else {

crates/contextforge-gateway-rs-lib/tests/gateway_resource_templates.rs

@@ -5,7 +5,8 @@ use rmcp::model::{ReadResourceRequestParams, ResourceContents};
 use tracing::{info, warn};
 
 use support::{
-    ListToolsGatewaySettings, connect_client, create_client, create_gateway_with_four_counters, create_ports,
+    ListToolsGatewaySettings, TEST_USER_ID, connect_client, create_client, create_gateway_with_four_counters,
+    create_ports,
 };
 
 #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
@@ -20,7 +21,7 @@ async fn plaintext_lists_prefixed_backend_resource_templates() -> Result<()> {
         ..Default::default()
     };
 
-    let user = "admin@example.com";
+    let user = TEST_USER_ID;
     let Ok(ListToolsGatewaySettings {
         handle,
         gateway_url,
@@ -57,7 +58,7 @@ async fn plaintext_reads_resource_from_prefixed_template() -> Result<()> {
         ..Default::default()
     };
 
-    let user = "admin@example.com";
+    let user = TEST_USER_ID;
     let Ok(ListToolsGatewaySettings { handle, gateway_url, .. }) =
         create_gateway_with_four_counters(user, config).await
     else {