File size: 7,310 Bytes
17f328f | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 | use std::sync::Arc;
use std::sync::Mutex;
use std::sync::atomic::AtomicUsize;
use std::sync::atomic::Ordering;
use std::time::Duration;
use anyhow::anyhow;
use pretty_assertions::assert_eq;
use super::MAX_MCP_CATALOG_ITEMS;
use super::MAX_MCP_CATALOG_PAGES;
use super::MAX_MCP_PAGINATION_CURSOR_BYTES;
use super::collect_paginated;
#[tokio::test]
async fn collects_all_pages_including_an_empty_cursor() {
let requests = Arc::new(Mutex::new(Vec::new()));
let observed = Arc::clone(&requests);
let pages = collect_paginated("tools/list", /*overall_timeout*/ None, move |params| {
let observed = Arc::clone(&observed);
async move {
let cursor = params.and_then(|params| params.cursor);
observed.lock().expect("request lock").push(cursor.clone());
match cursor.as_deref() {
None => Ok((vec!["first"], Some(String::new()))),
Some("") => Ok((vec!["second"], Some("last".to_string()))),
Some("last") => Ok((vec!["third"], None)),
Some(cursor) => Err(anyhow!("unexpected cursor: {cursor}")),
}
}
})
.await
.expect("paginated request succeeds");
assert_eq!(pages, vec!["first", "second", "third"]);
assert_eq!(
*requests.lock().expect("request lock"),
vec![None, Some(String::new()), Some("last".to_string())]
);
}
#[tokio::test]
async fn rejects_nonconsecutive_repeated_cursors() {
let error = collect_paginated(
"resources/list",
/*overall_timeout*/ None,
|params| async move {
let cursor = params.and_then(|params| params.cursor);
let next = match cursor.as_deref() {
None => "first",
Some("first") => "second",
Some("second") => "first",
Some(cursor) => return Err(anyhow!("unexpected cursor: {cursor}")),
};
Ok((Vec::<()>::new(), Some(next.to_string())))
},
)
.await
.expect_err("a repeated cursor must fail");
assert_eq!(
error.to_string(),
"resources/list returned a repeated pagination cursor"
);
}
#[tokio::test]
async fn rejects_excessive_pagination_before_fetching_another_page() {
let requests = Arc::new(AtomicUsize::new(0));
let observed = Arc::clone(&requests);
let error = collect_paginated(
"tools/list",
/*overall_timeout*/ None,
move |_params| {
let observed = Arc::clone(&observed);
async move {
let page = observed.fetch_add(1, Ordering::Relaxed);
Ok((Vec::<()>::new(), Some(page.to_string())))
}
},
)
.await
.expect_err("unbounded pagination must fail");
assert_eq!(
error.to_string(),
format!("tools/list exceeded the pagination limit of {MAX_MCP_CATALOG_PAGES} pages")
);
assert_eq!(requests.load(Ordering::Relaxed), MAX_MCP_CATALOG_PAGES);
}
#[tokio::test]
async fn rejects_a_page_exceeding_the_catalog_item_limit() {
let error = collect_paginated(
"tools/list",
/*overall_timeout*/ None,
|_params| async { Ok((vec![(); MAX_MCP_CATALOG_ITEMS + 1], None)) },
)
.await
.expect_err("an oversized catalog page must fail");
assert_eq!(
error.to_string(),
format!("tools/list exceeded the catalog limit of {MAX_MCP_CATALOG_ITEMS} items")
);
}
#[tokio::test]
async fn rejects_a_catalog_exceeding_the_item_limit_across_pages() {
let error = collect_paginated(
"tools/list",
/*overall_timeout*/ None,
|params| async move {
match params.and_then(|params| params.cursor) {
None => Ok((vec![(); MAX_MCP_CATALOG_ITEMS], Some("last".to_string()))),
Some(cursor) if cursor == "last" => Ok((vec![()], None)),
Some(cursor) => Err(anyhow!("unexpected cursor: {cursor}")),
}
},
)
.await
.expect_err("catalog items must be bounded across pages");
assert_eq!(
error.to_string(),
format!("tools/list exceeded the catalog limit of {MAX_MCP_CATALOG_ITEMS} items")
);
}
#[tokio::test]
async fn rejects_oversized_pagination_cursors_before_following_them() {
let requests = Arc::new(AtomicUsize::new(0));
let observed = Arc::clone(&requests);
let error = collect_paginated(
"resources/list",
/*overall_timeout*/ None,
move |_params| {
let observed = Arc::clone(&observed);
async move {
observed.fetch_add(1, Ordering::Relaxed);
Ok((
Vec::<()>::new(),
Some("x".repeat(MAX_MCP_PAGINATION_CURSOR_BYTES + 1)),
))
}
},
)
.await
.expect_err("an oversized pagination cursor must fail");
assert_eq!(
error.to_string(),
format!(
"resources/list returned a pagination cursor exceeding {MAX_MCP_PAGINATION_CURSOR_BYTES} bytes"
)
);
assert_eq!(requests.load(Ordering::Relaxed), 1);
}
#[tokio::test]
async fn forwards_page_failures() {
let error = collect_paginated(
"resources/templates/list",
/*overall_timeout*/ None,
|_params| async { Err::<(Vec<()>, Option<String>), _>(anyhow!("page failed")) },
)
.await
.expect_err("a page error must fail");
assert_eq!(error.to_string(), "page failed");
}
#[tokio::test(start_paused = true)]
async fn applies_a_default_timeout_when_no_timeout_is_configured() {
let error = collect_paginated(
"resources/list",
/*overall_timeout*/ None,
|_params| async {
tokio::time::sleep(Duration::from_secs(31)).await;
Ok((Vec::<()>::new(), None))
},
)
.await
.expect_err("pagination without a configured timeout must still be bounded");
assert_eq!(
error.to_string(),
"resources/list pagination timed out after 30s"
);
}
#[tokio::test(start_paused = true)]
async fn applies_one_timeout_across_individually_timely_pages() {
let requests = Arc::new(Mutex::new(Vec::new()));
let observed = Arc::clone(&requests);
let error = collect_paginated("tools/list", Some(Duration::from_secs(5)), move |params| {
let observed = Arc::clone(&observed);
async move {
let cursor = params.and_then(|params| params.cursor);
observed.lock().expect("request lock").push(cursor.clone());
tokio::time::sleep(Duration::from_secs(2)).await;
match cursor.as_deref() {
None => Ok((vec!["first"], Some("second".to_string()))),
Some("second") => Ok((vec!["second"], Some("third".to_string()))),
Some("third") => Ok((vec!["third"], None)),
Some(cursor) => Err(anyhow!("unexpected cursor: {cursor}")),
}
}
})
.await
.expect_err("the combined page duration must exceed the shared timeout");
assert_eq!(
error.to_string(),
"tools/list pagination timed out after 5s"
);
assert_eq!(
*requests.lock().expect("request lock"),
vec![None, Some("second".to_string()), Some("third".to_string())]
);
}
|