use std::io::{Read as _, Write as _};
use std::time::Duration;
use axum::body::Body;
use axum::http::{Request, StatusCode};
use futures_util::StreamExt as _;
use r402_http::{PAYMENT_REQUIRED, PAYMENT_RESPONSE, PAYMENT_SIGNATURE};
use r402_protocol::payment::{Base64Bytes, PaymentPayload, PaymentRequired};
use serde_json::json;
use tower::ServiceExt;
use wiremock::matchers::{method, path};
use wiremock::{Mock, MockServer, ResponseTemplate};
use crate::config::Config;
use crate::http::app;
fn paid_toml(facilitator: &str, upstream: &str) -> String {
format!(
r#"
[server]
base_url = "https://o402.example.com"
[payment]
enabled = true
[payment.facilitator]
url = "{facilitator}"
timeout_secs = 5
supported_cache_ttl_secs = 0
[payment.pay_to]
"eip155:*" = "0x0000000000000000000000000000000000000001"
[[payment.accepts]]
scheme = "exact"
network = "eip155:8453"
asset = "usdc"
[[payment.accepts]]
scheme = "upto"
network = "eip155:8453"
asset = "usdc"
[pricing.default]
scheme = "exact"
price = "0.001"
request_floor = "0.00001"
input_per_million = "0.15"
output_per_million = "0.60"
max_input_tokens = 128000
default_max_output_tokens = 16384
[[upstreams]]
name = "stub"
base_url = "{upstream}"
api_key = "sk-upstream"
timeout_secs = 5
connect_timeout_secs = 1
[[models]]
id = "gpt-4o-mini"
upstream = "stub"
"#
)
}
fn router(facilitator: &str, upstream: &str) -> axum::Router {
app(Config::from_toml_str(&paid_toml(facilitator, upstream)).expect("config")).expect("router")
}
fn chat_body() -> Body {
Body::from(r#"{"model":"gpt-4o-mini","messages":[{"role":"user","content":"hi"}]}"#)
}
fn stream_chat_body() -> Body {
Body::from(
r#"{"model":"gpt-4o-mini","messages":[{"role":"user","content":"hi"}],"stream":true}"#,
)
}
async fn bytes_of(response: axum::http::Response<Body>) -> Vec<u8> {
axum::body::to_bytes(response.into_body(), 1 << 20)
.await
.expect("body")
.to_vec()
}
async fn mount_supported(server: &MockServer) {
Mock::given(method("GET"))
.and(path("/supported"))
.respond_with(ResponseTemplate::new(200).set_body_json(json!({
"kinds": [{
"x402Version": 2,
"scheme": "exact",
"network": "eip155:8453"
}]
})))
.mount(server)
.await;
}
async fn mount_verify(server: &MockServer) {
Mock::given(method("POST"))
.and(path("/verify"))
.respond_with(ResponseTemplate::new(200).set_body_json(json!({
"isValid": true,
"payer": "0x0000000000000000000000000000000000000001"
})))
.mount(server)
.await;
}
async fn mount_settle(server: &MockServer, delay: Option<Duration>) {
let mut template = ResponseTemplate::new(200).set_body_json(json!({
"success": true,
"transaction": "0xsettle",
"network": "eip155:8453",
"amount": "1000"
}));
if let Some(delay) = delay {
template = template.set_delay(delay);
}
Mock::given(method("POST"))
.and(path("/settle"))
.respond_with(template)
.mount(server)
.await;
}
fn signature_from_402(body: &[u8]) -> String {
let required: PaymentRequired = serde_json::from_slice(body).expect("payment required");
let accept = required.accepts.first().cloned().expect("accept");
let payload = PaymentPayload::new(accept, json!({}));
let encoded = Base64Bytes::encode(serde_json::to_vec(&payload).expect("payload"));
String::from_utf8(encoded.as_ref().to_vec()).expect("base64 ascii")
}
fn count_path(requests: &[wiremock::Request], want: &str) -> usize {
requests
.iter()
.filter(|request| request.url.path() == want)
.count()
}
async fn unpaid_402(app: axum::Router) -> (Vec<u8>, axum::http::HeaderMap) {
unpaid_402_on(app, "/v1/chat/completions").await
}
async fn unpaid_exact_402(app: axum::Router) -> (Vec<u8>, axum::http::HeaderMap) {
unpaid_402_on(app, "/v1/images/generations").await
}
async fn unpaid_402_on(app: axum::Router, path: &str) -> (Vec<u8>, axum::http::HeaderMap) {
let response = app
.oneshot(
Request::post(path)
.header("content-type", "application/json")
.body(chat_body())
.expect("request"),
)
.await
.expect("response");
assert_eq!(
response.status(),
StatusCode::PAYMENT_REQUIRED,
"unpaid status"
);
let headers = response.headers().clone();
let body = bytes_of(response).await;
(body, headers)
}
#[tokio::test]
async fn missing_signature_is_402_body_and_header() {
let facilitator = MockServer::start().await;
let upstream = MockServer::start().await;
mount_supported(&facilitator).await;
let (body, headers) = unpaid_exact_402(router(&facilitator.uri(), &upstream.uri())).await;
assert!(
headers.get(PAYMENT_REQUIRED).is_some(),
"Payment-Required header"
);
assert_eq!(
headers
.get("cache-control")
.and_then(|value| value.to_str().ok()),
Some("no-store"),
"no-store"
);
let required: PaymentRequired = serde_json::from_slice(&body).expect("json");
assert!(!required.accepts.is_empty(), "accepts on 402 body");
let header = headers.get(PAYMENT_REQUIRED).expect("header").as_bytes();
let decoded = Base64Bytes::from(header).decode().expect("b64");
assert_eq!(decoded, body, "header is base64 of body");
assert_eq!(
count_path(
&facilitator.received_requests().await.expect("recorded"),
"/settle"
),
0,
"no settle"
);
}
#[tokio::test]
async fn exact_sequential_200_has_payment_response() {
let facilitator = MockServer::start().await;
let upstream = MockServer::start().await;
mount_supported(&facilitator).await;
mount_verify(&facilitator).await;
mount_settle(&facilitator, None).await;
Mock::given(method("POST"))
.and(path("/v1/images/generations"))
.respond_with(ResponseTemplate::new(200).set_body_json(json!({"id":"ok"})))
.mount(&upstream)
.await;
let app = router(&facilitator.uri(), &upstream.uri());
let (body, _) = unpaid_exact_402(app.clone()).await;
let signature = signature_from_402(&body);
let response = app
.oneshot(
Request::post("/v1/images/generations")
.header("content-type", "application/json")
.header(PAYMENT_SIGNATURE, signature)
.body(chat_body())
.expect("request"),
)
.await
.expect("response");
assert_eq!(response.status(), StatusCode::OK, "paid status");
assert!(
response.headers().get(PAYMENT_RESPONSE).is_some(),
"Payment-Response"
);
assert_eq!(
count_path(
&facilitator.received_requests().await.expect("recorded"),
"/settle"
),
1,
"one settle"
);
}
#[tokio::test]
async fn upstream_400_does_not_settle() {
let facilitator = MockServer::start().await;
let upstream = MockServer::start().await;
mount_supported(&facilitator).await;
mount_verify(&facilitator).await;
mount_settle(&facilitator, None).await;
Mock::given(method("POST"))
.and(path("/v1/images/generations"))
.respond_with(ResponseTemplate::new(400).set_body_json(json!({
"error": {"message": "bad request", "type": "invalid_request_error"}
})))
.mount(&upstream)
.await;
let app = router(&facilitator.uri(), &upstream.uri());
let (body, _) = unpaid_exact_402(app.clone()).await;
let signature = signature_from_402(&body);
let response = app
.oneshot(
Request::post("/v1/images/generations")
.header("content-type", "application/json")
.header(PAYMENT_SIGNATURE, signature)
.body(chat_body())
.expect("request"),
)
.await
.expect("response");
assert_eq!(response.status(), StatusCode::BAD_REQUEST, "passthrough");
assert!(
response.headers().get(PAYMENT_RESPONSE).is_none(),
"no receipt on 4xx"
);
assert_eq!(
count_path(
&facilitator.received_requests().await.expect("recorded"),
"/settle"
),
0,
"no settle"
);
}
#[tokio::test]
async fn duplicate_inflight_signature_rejected() {
let facilitator = MockServer::start().await;
let upstream = MockServer::start().await;
mount_supported(&facilitator).await;
mount_verify(&facilitator).await;
mount_settle(&facilitator, Some(Duration::from_millis(800))).await;
Mock::given(method("POST"))
.and(path("/v1/images/generations"))
.respond_with(ResponseTemplate::new(200).set_body_json(json!({"id":"ok"})))
.mount(&upstream)
.await;
let app = router(&facilitator.uri(), &upstream.uri());
let (body, _) = unpaid_exact_402(app.clone()).await;
let signature = signature_from_402(&body);
let first = {
let app = app.clone();
let signature = signature.clone();
tokio::spawn(async move {
app.oneshot(
Request::post("/v1/images/generations")
.header("content-type", "application/json")
.header(PAYMENT_SIGNATURE, signature)
.body(chat_body())
.expect("request"),
)
.await
.expect("response")
})
};
let started = tokio::time::Instant::now();
loop {
if count_path(
&facilitator.received_requests().await.expect("recorded"),
"/settle",
) > 0
{
break;
}
assert!(
started.elapsed() < Duration::from_secs(5),
"first request never reached settle"
);
tokio::time::sleep(Duration::from_millis(20)).await;
}
let second = app
.oneshot(
Request::post("/v1/images/generations")
.header("content-type", "application/json")
.header(PAYMENT_SIGNATURE, signature)
.body(chat_body())
.expect("request"),
)
.await
.expect("response");
assert_eq!(
second.status(),
StatusCode::PAYMENT_REQUIRED,
"duplicate in flight"
);
let second_body = bytes_of(second).await;
let text = String::from_utf8_lossy(&second_body);
assert!(
text.contains("already in flight"),
"in-flight error: {text}"
);
let first = first.await.expect("join");
assert_eq!(first.status(), StatusCode::OK, "first completes");
assert_eq!(
count_path(
&facilitator.received_requests().await.expect("recorded"),
"/settle"
),
1,
"single settle"
);
}
#[tokio::test]
async fn ready_200_when_supported_ok() {
let facilitator = MockServer::start().await;
mount_supported(&facilitator).await;
let response = router(&facilitator.uri(), "http://127.0.0.1:9")
.oneshot(Request::get("/ready").body(Body::empty()).expect("request"))
.await
.expect("response");
assert_eq!(response.status(), StatusCode::OK, "ready");
}
#[tokio::test]
async fn ready_503_when_facilitator_down() {
let response = router("http://127.0.0.1:1", "http://127.0.0.1:9")
.oneshot(Request::get("/ready").body(Body::empty()).expect("request"))
.await
.expect("response");
assert_eq!(
response.status(),
StatusCode::SERVICE_UNAVAILABLE,
"ready down"
);
}
async fn paid_on(
app: axum::Router,
path: &str,
signature: &str,
body: Body,
) -> axum::http::Response<Body> {
app.oneshot(
Request::post(path)
.header("content-type", "application/json")
.header(PAYMENT_SIGNATURE, signature)
.body(body)
.expect("request"),
)
.await
.expect("response")
}
async fn paid_chat(app: axum::Router, signature: &str) -> axum::http::Response<Body> {
paid_on(app, "/v1/chat/completions", signature, chat_body()).await
}
async fn paid_exact(app: axum::Router, signature: &str) -> axum::http::Response<Body> {
paid_on(app, "/v1/images/generations", signature, chat_body()).await
}
async fn mount_upstream_ok(server: &MockServer) {
Mock::given(method("POST"))
.and(path("/v1/images/generations"))
.respond_with(ResponseTemplate::new(200).set_body_json(json!({"id":"ok"})))
.mount(server)
.await;
}
#[tokio::test]
async fn malformed_signature_is_402() {
let facilitator = MockServer::start().await;
let upstream = MockServer::start().await;
mount_supported(&facilitator).await;
let response = router(&facilitator.uri(), &upstream.uri())
.oneshot(
Request::post("/v1/images/generations")
.header("content-type", "application/json")
.header(PAYMENT_SIGNATURE, "not-valid-base64!!!")
.body(chat_body())
.expect("request"),
)
.await
.expect("response");
assert_eq!(response.status(), StatusCode::PAYMENT_REQUIRED, "status");
assert!(
response.headers().get(PAYMENT_REQUIRED).is_some(),
"challenge header"
);
let raw = bytes_of(response).await;
let body = String::from_utf8_lossy(&raw);
assert!(
body.contains("malformed") || body.contains("Invalid"),
"malformed: {body}"
);
}
#[tokio::test]
async fn invalid_verify_is_402_challenge() {
let facilitator = MockServer::start().await;
let upstream = MockServer::start().await;
mount_supported(&facilitator).await;
Mock::given(method("POST"))
.and(path("/verify"))
.respond_with(ResponseTemplate::new(200).set_body_json(json!({
"isValid": false,
"invalidReason": "invalid_payload"
})))
.mount(&facilitator)
.await;
mount_settle(&facilitator, None).await;
mount_upstream_ok(&upstream).await;
let app = router(&facilitator.uri(), &upstream.uri());
let (body, _) = unpaid_exact_402(app.clone()).await;
let signature = signature_from_402(&body);
let response = paid_exact(app, &signature).await;
assert_eq!(response.status(), StatusCode::PAYMENT_REQUIRED, "invalid");
assert!(
response.headers().get(PAYMENT_REQUIRED).is_some(),
"challenge header"
);
assert_eq!(
response
.headers()
.get("cache-control")
.and_then(|value| value.to_str().ok()),
Some("no-store"),
"no-store"
);
let raw = bytes_of(response).await;
let text = String::from_utf8_lossy(&raw);
assert!(
text.contains("Verification failed"),
"verify challenge: {text}"
);
assert_eq!(
count_path(
&facilitator.received_requests().await.expect("recorded"),
"/settle"
),
0,
"no settle"
);
}
#[tokio::test]
async fn verify_http_500_is_transport_502() {
let facilitator = MockServer::start().await;
let upstream = MockServer::start().await;
mount_supported(&facilitator).await;
Mock::given(method("POST"))
.and(path("/verify"))
.respond_with(ResponseTemplate::new(500).set_body_string("boom"))
.up_to_n_times(1)
.mount(&facilitator)
.await;
let app = router(&facilitator.uri(), &upstream.uri());
let (body, _) = unpaid_exact_402(app.clone()).await;
let signature = signature_from_402(&body);
let response = paid_exact(app.clone(), &signature).await;
assert_eq!(response.status(), StatusCode::BAD_GATEWAY, "transport");
let raw = bytes_of(response).await;
let text = String::from_utf8_lossy(&raw);
assert!(
text.contains("facilitator transport"),
"transport body: {text}"
);
mount_verify(&facilitator).await;
mount_settle(&facilitator, None).await;
mount_upstream_ok(&upstream).await;
let retry = paid_exact(app, &signature).await;
assert_eq!(
retry.status(),
StatusCode::OK,
"hash released after verify fail"
);
}
#[tokio::test]
async fn settle_failure_after_2xx_is_402() {
let facilitator = MockServer::start().await;
let upstream = MockServer::start().await;
mount_supported(&facilitator).await;
mount_verify(&facilitator).await;
Mock::given(method("POST"))
.and(path("/settle"))
.respond_with(ResponseTemplate::new(200).set_body_json(json!({
"success": false,
"errorReason": "invalid_transaction_state",
"transaction": "",
"network": "eip155:8453"
})))
.mount(&facilitator)
.await;
mount_upstream_ok(&upstream).await;
let app = router(&facilitator.uri(), &upstream.uri());
let (body, _) = unpaid_exact_402(app.clone()).await;
let signature = signature_from_402(&body);
let response = paid_exact(app, &signature).await;
assert_eq!(
response.status(),
StatusCode::PAYMENT_REQUIRED,
"settle fail"
);
assert_eq!(
count_path(
&facilitator.received_requests().await.expect("recorded"),
"/settle"
),
1,
"settle attempted"
);
}
#[tokio::test]
async fn timeout_during_verify_releases_hash() {
let facilitator = MockServer::start().await;
let upstream = MockServer::start().await;
mount_supported(&facilitator).await;
Mock::given(method("POST"))
.and(path("/verify"))
.respond_with(
ResponseTemplate::new(200)
.set_delay(Duration::from_secs(3))
.set_body_json(json!({
"isValid": true,
"payer": "0x0000000000000000000000000000000000000001"
})),
)
.up_to_n_times(1)
.mount(&facilitator)
.await;
mount_verify(&facilitator).await;
mount_settle(&facilitator, None).await;
mount_upstream_ok(&upstream).await;
let toml = paid_toml(&facilitator.uri(), &upstream.uri()).replace(
"base_url = \"https://o402.example.com\"",
"base_url = \"https://o402.example.com\"\nrequest_timeout_secs = 1",
);
let app = app(Config::from_toml_str(&toml).expect("config")).expect("router");
let (body, _) = unpaid_exact_402(app.clone()).await;
let signature = signature_from_402(&body);
let timed_out = paid_exact(app.clone(), &signature).await;
assert_eq!(
timed_out.status(),
StatusCode::GATEWAY_TIMEOUT,
"timeout status"
);
let retry = paid_exact(app, &signature).await;
assert_eq!(retry.status(), StatusCode::OK, "retry after timeout");
}
#[tokio::test]
async fn permit2_accept_is_advertised() {
let facilitator = MockServer::start().await;
let upstream = MockServer::start().await;
mount_supported(&facilitator).await;
let toml = paid_toml(&facilitator.uri(), &upstream.uri()).replace(
"scheme = \"exact\"\nnetwork = \"eip155:8453\"\nasset = \"usdc\"",
"scheme = \"exact\"\nnetwork = \"eip155:8453\"\nasset = \"usdc\"\ntransfer_method = \"permit2\"",
);
let (_, headers) =
unpaid_exact_402(app(Config::from_toml_str(&toml).expect("config")).expect("router")).await;
let header = headers.get(PAYMENT_REQUIRED).expect("header").as_bytes();
let decoded = Base64Bytes::from(header).decode().expect("b64");
let text = String::from_utf8_lossy(&decoded);
assert!(text.contains("permit2"), "permit2 extra: {text}");
}
#[tokio::test]
async fn svm_exact_402_lists_solana_accept() {
let facilitator = MockServer::start().await;
let upstream = MockServer::start().await;
Mock::given(method("GET"))
.and(path("/supported"))
.respond_with(ResponseTemplate::new(200).set_body_json(json!({
"kinds": [{
"x402Version": 2,
"scheme": "exact",
"network": "solana:5eykt4UsFv8P8NJdTREpY1vzqKqZKvdp"
}]
})))
.mount(&facilitator)
.await;
let toml = format!(
r#"
[server]
base_url = "https://o402.example.com"
[payment]
enabled = true
[payment.facilitator]
url = "{}"
timeout_secs = 5
supported_cache_ttl_secs = 0
[payment.pay_to]
"solana:*" = "11111111111111111111111111111111"
[[payment.accepts]]
scheme = "exact"
network = "solana:5eykt4UsFv8P8NJdTREpY1vzqKqZKvdp"
asset = "usdc"
[pricing.default]
scheme = "exact"
price = "0.001"
request_floor = "0.00001"
input_per_million = "0.15"
output_per_million = "0.60"
max_input_tokens = 128000
default_max_output_tokens = 16384
[[upstreams]]
name = "stub"
base_url = "{}"
api_key = "sk-upstream"
[[models]]
id = "gpt-4o-mini"
upstream = "stub"
"#,
facilitator.uri(),
upstream.uri()
);
let (body, _) =
unpaid_exact_402(app(Config::from_toml_str(&toml).expect("config")).expect("router")).await;
let required: PaymentRequired = serde_json::from_slice(&body).expect("json");
let network = required
.accepts
.first()
.map(|accept| accept.network.to_string())
.unwrap_or_default();
assert!(
network.starts_with("solana:"),
"solana accept, got {network}"
);
}
fn paid_app(facilitator: &str, upstream: &str) -> (axum::Router, crate::state::AppState) {
crate::http::app_and_state(
Config::from_toml_str(&paid_toml(facilitator, upstream)).expect("config"),
)
.expect("router")
}
async fn paid_stream(app: axum::Router, signature: &str) -> axum::http::Response<Body> {
paid_on(app, "/v1/chat/completions", signature, stream_chat_body()).await
}
async fn paid_exact_stream(app: axum::Router, signature: &str) -> axum::http::Response<Body> {
paid_on(app, "/v1/images/generations", signature, stream_chat_body()).await
}
async fn next_containing(
stream: &mut axum::body::BodyDataStream,
needle: &[u8],
) -> Option<Vec<u8>> {
let mut buf = Vec::new();
loop {
let chunk = stream.next().await?.ok()?;
buf.extend_from_slice(&chunk);
if buf.windows(needle.len()).any(|window| window == needle) {
return Some(buf);
}
}
}
async fn wait_path_count(server: &MockServer, want: &str, n: usize) {
let started = tokio::time::Instant::now();
loop {
if count_path(&server.received_requests().await.expect("recorded"), want) >= n {
return;
}
assert!(
started.elapsed() < Duration::from_secs(5),
"timed out waiting for {want} x{n}"
);
tokio::time::sleep(Duration::from_millis(20)).await;
}
}
fn spawn_held_sse() -> (
String,
std::sync::mpsc::Sender<()>,
std::thread::JoinHandle<()>,
) {
let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("bind");
let addr = listener.local_addr().expect("addr");
let (tx, rx) = std::sync::mpsc::channel();
let thread = std::thread::spawn(move || {
let (mut sock, _) = listener.accept().expect("accept");
sock.set_nodelay(true).expect("nodelay");
let mut buf = Vec::new();
let mut tmp = [0_u8; 1024];
loop {
let n = sock.read(&mut tmp).expect("read request");
if n == 0 {
break;
}
let Some(chunk) = tmp.get(..n) else {
break;
};
buf.extend_from_slice(chunk);
if buf.windows(4).any(|window| window == b"\r\n\r\n") {
break;
}
}
sock.write_all(
b"HTTP/1.1 200 OK\r\nContent-Type: text/event-stream\r\nCache-Control: no-cache\r\n\r\ndata: first\n\n",
)
.expect("headers");
sock.flush().expect("flush first");
rx.recv().expect("hold until first byte is observed");
sock.write_all(b"data: last\n\n").expect("last");
sock.flush().expect("flush last");
});
(format!("http://{addr}"), tx, thread)
}
#[tokio::test]
async fn exact_stream_first_sse_byte_precedes_settle() {
let facilitator = MockServer::start().await;
mount_supported(&facilitator).await;
mount_verify(&facilitator).await;
mount_settle(&facilitator, Some(Duration::from_millis(800))).await;
let (base, release, thread) = spawn_held_sse();
let (app, state) = paid_app(&facilitator.uri(), &base);
let (body, _) = unpaid_exact_402(app.clone()).await;
let signature = signature_from_402(&body);
let started = tokio::time::Instant::now();
let response = tokio::time::timeout(
Duration::from_millis(400),
paid_exact_stream(app, &signature),
)
.await
.expect("headers waited on settle or stream end");
assert_eq!(response.status(), StatusCode::OK, "paid stream status");
assert!(
response.headers().get(PAYMENT_RESPONSE).is_none(),
"no Payment-Response on stream"
);
assert!(
response
.headers()
.get("content-type")
.and_then(|value| value.to_str().ok())
.is_some_and(|ct| ct.contains("text/event-stream")),
"sse content-type"
);
let mut stream = response.into_body().into_data_stream();
let first = tokio::time::timeout(
Duration::from_millis(400),
next_containing(&mut stream, b"first"),
)
.await
.expect("first SSE byte waited on settle")
.expect("first chunk");
assert!(
first
.windows(b"first".len())
.any(|window| window == b"first"),
"first event: {}",
String::from_utf8_lossy(&first)
);
assert!(
started.elapsed() < Duration::from_millis(800),
"first byte arrived after settle delay"
);
release.send(()).expect("release");
let last = tokio::time::timeout(
Duration::from_secs(2),
next_containing(&mut stream, b"last"),
)
.await
.expect("last event")
.expect("last chunk");
assert!(
last.windows(b"last".len()).any(|window| window == b"last"),
"last event"
);
thread.join().expect("sse thread");
wait_path_count(&facilitator, "/settle", 1).await;
state
.inflight()
.wait_for_drain(Duration::from_secs(2))
.await;
}
#[tokio::test]
async fn exact_stream_duplicate_signature_is_402_until_spawn_finishes() {
let facilitator = MockServer::start().await;
mount_supported(&facilitator).await;
mount_verify(&facilitator).await;
mount_settle(&facilitator, Some(Duration::from_millis(800))).await;
let (base, release, thread) = spawn_held_sse();
let (app, state) = paid_app(&facilitator.uri(), &base);
let (body, _) = unpaid_exact_402(app.clone()).await;
let signature = signature_from_402(&body);
let response = tokio::time::timeout(
Duration::from_millis(400),
paid_exact_stream(app.clone(), &signature),
)
.await
.expect("headers waited on settle or stream end");
assert_eq!(response.status(), StatusCode::OK, "first stream");
let mut stream = response.into_body().into_data_stream();
let first = tokio::time::timeout(
Duration::from_millis(400),
next_containing(&mut stream, b"first"),
)
.await
.expect("first SSE byte")
.expect("first chunk");
assert!(
first
.windows(b"first".len())
.any(|window| window == b"first"),
"first event"
);
let second = paid_exact_stream(app.clone(), &signature).await;
assert_eq!(
second.status(),
StatusCode::PAYMENT_REQUIRED,
"duplicate in flight during stream"
);
let second_body = bytes_of(second).await;
let text = String::from_utf8_lossy(&second_body);
assert!(
text.contains("already in flight"),
"in-flight error: {text}"
);
release.send(()).expect("release");
let last = tokio::time::timeout(
Duration::from_secs(2),
next_containing(&mut stream, b"last"),
)
.await
.expect("last event")
.expect("last chunk");
assert!(
last.windows(b"last".len()).any(|window| window == b"last"),
"last event"
);
drop(stream);
thread.join().expect("sse thread");
state
.inflight()
.wait_for_drain(Duration::from_secs(2))
.await;
assert_eq!(
count_path(
&facilitator.received_requests().await.expect("recorded"),
"/settle"
),
1,
"single settle"
);
let retry = paid_exact_stream(app, &signature).await;
let retry_status = retry.status();
let retry_body = bytes_of(retry).await;
let retry_text = String::from_utf8_lossy(&retry_body);
assert!(
retry_status != StatusCode::PAYMENT_REQUIRED || !retry_text.contains("already in flight"),
"hash still held after spawn: {retry_status} {retry_text}"
);
}
#[tokio::test]
async fn exact_stream_upstream_4xx_does_not_settle() {
let facilitator = MockServer::start().await;
let upstream = MockServer::start().await;
mount_supported(&facilitator).await;
mount_verify(&facilitator).await;
mount_settle(&facilitator, None).await;
Mock::given(method("POST"))
.and(path("/v1/images/generations"))
.respond_with(
ResponseTemplate::new(400)
.set_body_string("data: error\n\n")
.insert_header("content-type", "text/event-stream"),
)
.mount(&upstream)
.await;
let app = router(&facilitator.uri(), &upstream.uri());
let (body, _) = unpaid_exact_402(app.clone()).await;
let signature = signature_from_402(&body);
let response = paid_exact_stream(app, &signature).await;
assert_eq!(response.status(), StatusCode::BAD_REQUEST, "passthrough");
assert!(
response.headers().get(PAYMENT_RESPONSE).is_none(),
"no receipt on 4xx stream"
);
let raw = bytes_of(response).await;
assert!(
raw.windows(b"error".len()).any(|window| window == b"error"),
"error body: {}",
String::from_utf8_lossy(&raw)
);
assert_eq!(
count_path(
&facilitator.received_requests().await.expect("recorded"),
"/settle"
),
0,
"no settle"
);
}
#[tokio::test]
async fn exact_stream_complete_then_drop_is_one_settle() {
let facilitator = MockServer::start().await;
let upstream = MockServer::start().await;
mount_supported(&facilitator).await;
mount_verify(&facilitator).await;
mount_settle(&facilitator, Some(Duration::from_millis(400))).await;
Mock::given(method("POST"))
.and(path("/v1/images/generations"))
.respond_with(
ResponseTemplate::new(200)
.set_body_string("data: done\n\n")
.insert_header("content-type", "text/event-stream"),
)
.mount(&upstream)
.await;
let (app, state) = paid_app(&facilitator.uri(), &upstream.uri());
let (body, _) = unpaid_exact_402(app.clone()).await;
let signature = signature_from_402(&body);
let response = paid_exact_stream(app, &signature).await;
assert_eq!(response.status(), StatusCode::OK, "stream");
let mut stream = response.into_body().into_data_stream();
let done = next_containing(&mut stream, b"done")
.await
.expect("done event");
assert!(
done.windows(b"done".len()).any(|window| window == b"done"),
"done"
);
drop(stream);
state
.inflight()
.wait_for_drain(Duration::from_secs(2))
.await;
assert_eq!(
count_path(
&facilitator.received_requests().await.expect("recorded"),
"/settle"
),
1,
"SettleOnce: one settle after complete then drop"
);
}
#[tokio::test]
async fn timeout_before_2xx_releases_hash() {
let facilitator = MockServer::start().await;
let upstream = MockServer::start().await;
mount_supported(&facilitator).await;
mount_verify(&facilitator).await;
mount_settle(&facilitator, None).await;
Mock::given(method("POST"))
.and(path("/v1/images/generations"))
.respond_with(
ResponseTemplate::new(200)
.set_delay(Duration::from_secs(3))
.set_body_string("data: late\n\n")
.insert_header("content-type", "text/event-stream"),
)
.up_to_n_times(1)
.mount(&upstream)
.await;
Mock::given(method("POST"))
.and(path("/v1/images/generations"))
.respond_with(
ResponseTemplate::new(200)
.set_body_string("data: ok\n\n")
.insert_header("content-type", "text/event-stream"),
)
.mount(&upstream)
.await;
let toml = paid_toml(&facilitator.uri(), &upstream.uri()).replace(
"base_url = \"https://o402.example.com\"",
"base_url = \"https://o402.example.com\"\nrequest_timeout_secs = 1",
);
let (app, state) =
crate::http::app_and_state(Config::from_toml_str(&toml).expect("config")).expect("router");
let (body, _) = unpaid_exact_402(app.clone()).await;
let signature = signature_from_402(&body);
let timed_out = paid_exact_stream(app.clone(), &signature).await;
assert_eq!(
timed_out.status(),
StatusCode::GATEWAY_TIMEOUT,
"timeout before 2xx"
);
assert_eq!(
count_path(
&facilitator.received_requests().await.expect("recorded"),
"/settle"
),
0,
"no settle on cancel before 2xx"
);
let retry = paid_exact_stream(app, &signature).await;
assert_eq!(retry.status(), StatusCode::OK, "retry after timeout");
wait_path_count(&facilitator, "/settle", 1).await;
state
.inflight()
.wait_for_drain(Duration::from_secs(2))
.await;
}
fn paid_upto_toml(facilitator: &str, upstream: &str) -> String {
format!(
r#"
[server]
base_url = "https://o402.example.com"
[payment]
enabled = true
missing_usage = "ceiling"
abort_usage = "ceiling"
[payment.facilitator]
url = "{facilitator}"
timeout_secs = 5
supported_cache_ttl_secs = 0
[payment.pay_to]
"eip155:*" = "0x0000000000000000000000000000000000000001"
[[payment.accepts]]
scheme = "exact"
network = "eip155:8453"
asset = "usdc"
[[payment.accepts]]
scheme = "upto"
network = "eip155:8453"
asset = "usdc"
[pricing.default]
scheme = "upto"
price = "0.001"
request_floor = "0.00001"
input_per_million = "0.15"
output_per_million = "0.60"
cached_input_per_million = "0.075"
reasoning_per_million = "0.60"
ceiling_multiplier = "1.0"
max_ceiling = "5.00"
max_input_tokens = 128000
default_max_output_tokens = 16384
[[upstreams]]
name = "stub"
base_url = "{upstream}"
api_key = "sk-upstream"
timeout_secs = 5
connect_timeout_secs = 1
[[models]]
id = "gpt-4o-mini"
upstream = "stub"
scheme = "upto"
input_per_million = "0.15"
output_per_million = "0.60"
cached_input_per_million = "0.075"
max_input_tokens = 128000
default_max_output_tokens = 16384
[[models]]
id = "text-embedding-3-small"
upstream = "stub"
scheme = "upto"
input_per_million = "0.02"
max_input_tokens = 8191
"#
)
}
fn upto_router(facilitator: &str, upstream: &str) -> axum::Router {
app(Config::from_toml_str(&paid_upto_toml(facilitator, upstream)).expect("config"))
.expect("router")
}
fn upto_app(facilitator: &str, upstream: &str) -> (axum::Router, crate::state::AppState) {
crate::http::app_and_state(
Config::from_toml_str(&paid_upto_toml(facilitator, upstream)).expect("config"),
)
.expect("router")
}
fn sse_usage_body() -> String {
"data: {\"id\":\"ok\",\"choices\":[{\"delta\":{\"content\":\"hi\"}}]}\n\n\
data: {\"id\":\"ok\",\"choices\":[],\"usage\":{\"prompt_tokens\":1000,\"completion_tokens\":100,\"prompt_tokens_details\":{\"cached_tokens\":200},\"completion_tokens_details\":{\"reasoning_tokens\":40}}}\n\n\
data: [DONE]\n\n"
.to_owned()
}
fn spawn_open_sse() -> (String, std::thread::JoinHandle<()>) {
let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("bind");
let addr = listener.local_addr().expect("addr");
let thread = std::thread::spawn(move || {
let (mut sock, _) = listener.accept().expect("accept");
sock.set_nodelay(true).expect("nodelay");
let mut buf = Vec::new();
let mut request_buf = [0_u8; 1024];
loop {
let n = sock.read(&mut request_buf).expect("read request");
if n == 0 {
break;
}
let Some(chunk) = request_buf.get(..n) else {
break;
};
buf.extend_from_slice(chunk);
if buf.windows(4).any(|window| window == b"\r\n\r\n") {
break;
}
}
sock.write_all(
b"HTTP/1.1 200 OK\r\nContent-Type: text/event-stream\r\nCache-Control: no-cache\r\n\r\ndata: first\n\n",
)
.expect("headers");
sock.flush().expect("flush");
let mut discard = [0_u8; 8];
drop(sock.read(&mut discard));
});
(format!("http://{addr}"), thread)
}
async fn mount_supported_upto(server: &MockServer) {
Mock::given(method("GET"))
.and(path("/supported"))
.respond_with(ResponseTemplate::new(200).set_body_json(json!({
"kinds": [{
"x402Version": 2,
"scheme": "upto",
"network": "eip155:8453"
}]
})))
.mount(server)
.await;
}
fn settle_amount(requests: &[wiremock::Request]) -> Option<String> {
let request = requests
.iter()
.find(|request| request.url.path() == "/settle")?;
let body: serde_json::Value = serde_json::from_slice(&request.body).ok()?;
body.pointer("/paymentRequirements/amount")?
.as_str()
.map(str::to_owned)
}
#[tokio::test]
async fn upto_402_amount_is_29040() {
let facilitator = MockServer::start().await;
let upstream = MockServer::start().await;
mount_supported_upto(&facilitator).await;
let (body, _) = unpaid_402(upto_router(&facilitator.uri(), &upstream.uri())).await;
let required: PaymentRequired = serde_json::from_slice(&body).expect("json");
let accept = required.accepts.first().expect("accept");
assert_eq!(accept.scheme.as_str(), "upto", "scheme");
assert_eq!(accept.amount.as_str(), "29040", "ceiling");
}
#[tokio::test]
async fn upto_sequential_settles_actual_from_usage() {
let facilitator = MockServer::start().await;
let upstream = MockServer::start().await;
mount_supported_upto(&facilitator).await;
mount_verify(&facilitator).await;
mount_settle(&facilitator, None).await;
Mock::given(method("POST"))
.and(path("/v1/chat/completions"))
.respond_with(ResponseTemplate::new(200).set_body_json(json!({
"id": "ok",
"usage": {
"prompt_tokens": 1000,
"completion_tokens": 100,
"prompt_tokens_details": {"cached_tokens": 200},
"completion_tokens_details": {"reasoning_tokens": 40}
}
})))
.mount(&upstream)
.await;
let app = upto_router(&facilitator.uri(), &upstream.uri());
let (body, _) = unpaid_402(app.clone()).await;
let signature = signature_from_402(&body);
let response = paid_chat(app, &signature).await;
assert_eq!(response.status(), StatusCode::OK, "paid status");
assert!(
response.headers().get(PAYMENT_RESPONSE).is_some(),
"Payment-Response"
);
let recorded = facilitator.received_requests().await.expect("recorded");
assert_eq!(count_path(&recorded, "/settle"), 1, "one settle");
assert_eq!(
settle_amount(&recorded).as_deref(),
Some("205"),
"actual meter"
);
}
#[tokio::test]
async fn upto_missing_usage_settles_ceiling() {
let facilitator = MockServer::start().await;
let upstream = MockServer::start().await;
mount_supported_upto(&facilitator).await;
mount_verify(&facilitator).await;
mount_settle(&facilitator, None).await;
Mock::given(method("POST"))
.and(path("/v1/chat/completions"))
.respond_with(ResponseTemplate::new(200).set_body_json(json!({"id":"ok"})))
.mount(&upstream)
.await;
let app = upto_router(&facilitator.uri(), &upstream.uri());
let (body, _) = unpaid_402(app.clone()).await;
let signature = signature_from_402(&body);
let response = paid_chat(app, &signature).await;
assert_eq!(response.status(), StatusCode::OK, "paid status");
assert!(
response.headers().get(PAYMENT_RESPONSE).is_some(),
"Payment-Response"
);
let recorded = facilitator.received_requests().await.expect("recorded");
assert_eq!(
settle_amount(&recorded).as_deref(),
Some("29040"),
"missing usage ceiling"
);
}
#[tokio::test]
async fn upto_upstream_400_does_not_settle() {
let facilitator = MockServer::start().await;
let upstream = MockServer::start().await;
mount_supported_upto(&facilitator).await;
mount_verify(&facilitator).await;
mount_settle(&facilitator, None).await;
Mock::given(method("POST"))
.and(path("/v1/chat/completions"))
.respond_with(ResponseTemplate::new(400).set_body_json(json!({
"error": {"message": "bad request", "type": "invalid_request_error"}
})))
.mount(&upstream)
.await;
let app = upto_router(&facilitator.uri(), &upstream.uri());
let (body, _) = unpaid_402(app.clone()).await;
let signature = signature_from_402(&body);
let response = paid_chat(app, &signature).await;
assert_eq!(response.status(), StatusCode::BAD_REQUEST, "passthrough");
assert!(
response.headers().get(PAYMENT_RESPONSE).is_none(),
"no receipt on 4xx"
);
assert_eq!(
count_path(
&facilitator.received_requests().await.expect("recorded"),
"/settle"
),
0,
"no settle"
);
}
#[tokio::test]
async fn upto_stream_first_sse_byte_precedes_settle() {
let facilitator = MockServer::start().await;
mount_supported_upto(&facilitator).await;
mount_verify(&facilitator).await;
mount_settle(&facilitator, Some(Duration::from_millis(800))).await;
let (base, release, thread) = spawn_held_sse();
let (app, state) = upto_app(&facilitator.uri(), &base);
let (body, _) = unpaid_402(app.clone()).await;
let signature = signature_from_402(&body);
let response = tokio::time::timeout(Duration::from_millis(400), paid_stream(app, &signature))
.await
.expect("headers waited on settle or stream end");
assert_eq!(response.status(), StatusCode::OK, "paid stream status");
assert!(
response.headers().get(PAYMENT_RESPONSE).is_none(),
"no Payment-Response on stream"
);
assert!(
response
.headers()
.get("content-type")
.and_then(|value| value.to_str().ok())
.is_some_and(|ct| ct.contains("text/event-stream")),
"sse content-type"
);
let mut stream = response.into_body().into_data_stream();
let first = tokio::time::timeout(
Duration::from_millis(400),
next_containing(&mut stream, b"first"),
)
.await
.expect("first SSE byte waited on settle")
.expect("first chunk");
assert!(
first
.windows(b"first".len())
.any(|window| window == b"first"),
"first event: {}",
String::from_utf8_lossy(&first)
);
assert_eq!(
count_path(
&facilitator.received_requests().await.expect("recorded"),
"/settle"
),
0,
"no settle before body end"
);
release.send(()).expect("release");
let last = tokio::time::timeout(
Duration::from_secs(2),
next_containing(&mut stream, b"last"),
)
.await
.expect("last event")
.expect("last chunk");
assert!(
last.windows(b"last".len()).any(|window| window == b"last"),
"last event"
);
drop(stream);
thread.join().expect("sse thread");
wait_path_count(&facilitator, "/settle", 1).await;
state
.inflight()
.wait_for_drain(Duration::from_secs(2))
.await;
}
#[tokio::test]
async fn upto_stream_duplicate_signature_is_402_until_spawn_finishes() {
let facilitator = MockServer::start().await;
mount_supported_upto(&facilitator).await;
mount_verify(&facilitator).await;
mount_settle(&facilitator, Some(Duration::from_millis(800))).await;
let (base, release, thread) = spawn_held_sse();
let (app, state) = upto_app(&facilitator.uri(), &base);
let (body, _) = unpaid_402(app.clone()).await;
let signature = signature_from_402(&body);
let response = tokio::time::timeout(
Duration::from_millis(400),
paid_stream(app.clone(), &signature),
)
.await
.expect("headers waited on settle or stream end");
assert_eq!(response.status(), StatusCode::OK, "first stream");
let mut stream = response.into_body().into_data_stream();
let first = tokio::time::timeout(
Duration::from_millis(400),
next_containing(&mut stream, b"first"),
)
.await
.expect("first SSE byte")
.expect("first chunk");
assert!(
first
.windows(b"first".len())
.any(|window| window == b"first"),
"first event"
);
assert_eq!(
count_path(
&facilitator.received_requests().await.expect("recorded"),
"/settle"
),
0,
"no settle while stream open"
);
let second = paid_stream(app.clone(), &signature).await;
assert_eq!(
second.status(),
StatusCode::PAYMENT_REQUIRED,
"duplicate in flight during stream"
);
let second_body = bytes_of(second).await;
let text = String::from_utf8_lossy(&second_body);
assert!(
text.contains("already in flight"),
"in-flight error: {text}"
);
release.send(()).expect("release");
let last = tokio::time::timeout(
Duration::from_secs(2),
next_containing(&mut stream, b"last"),
)
.await
.expect("last event")
.expect("last chunk");
assert!(
last.windows(b"last".len()).any(|window| window == b"last"),
"last event"
);
drop(stream);
thread.join().expect("sse thread");
state
.inflight()
.wait_for_drain(Duration::from_secs(2))
.await;
assert_eq!(
count_path(
&facilitator.received_requests().await.expect("recorded"),
"/settle"
),
1,
"single settle"
);
let retry = paid_stream(app, &signature).await;
let retry_status = retry.status();
let retry_body = bytes_of(retry).await;
let retry_text = String::from_utf8_lossy(&retry_body);
assert!(
retry_status != StatusCode::PAYMENT_REQUIRED || !retry_text.contains("already in flight"),
"hash still held after spawn: {retry_status} {retry_text}"
);
}
#[tokio::test]
async fn upto_stream_complete_then_drop_is_one_settle() {
let facilitator = MockServer::start().await;
let upstream = MockServer::start().await;
mount_supported_upto(&facilitator).await;
mount_verify(&facilitator).await;
mount_settle(&facilitator, Some(Duration::from_millis(400))).await;
Mock::given(method("POST"))
.and(path("/v1/chat/completions"))
.respond_with(
ResponseTemplate::new(200)
.set_body_string(sse_usage_body())
.insert_header("content-type", "text/event-stream"),
)
.mount(&upstream)
.await;
let (app, state) = upto_app(&facilitator.uri(), &upstream.uri());
let (body, _) = unpaid_402(app.clone()).await;
let signature = signature_from_402(&body);
let response = paid_stream(app, &signature).await;
assert_eq!(response.status(), StatusCode::OK, "stream");
assert!(
response.headers().get(PAYMENT_RESPONSE).is_none(),
"no Payment-Response on stream"
);
let mut stream = response.into_body().into_data_stream();
let done = next_containing(&mut stream, b"[DONE]")
.await
.expect("done event");
assert!(
done.windows(b"[DONE]".len())
.any(|window| window == b"[DONE]"),
"done"
);
while stream.next().await.is_some() {}
drop(stream);
state
.inflight()
.wait_for_drain(Duration::from_secs(2))
.await;
let recorded = facilitator.received_requests().await.expect("recorded");
assert_eq!(
count_path(&recorded, "/settle"),
1,
"SettleOnce: one settle after complete then drop"
);
assert_eq!(
settle_amount(&recorded).as_deref(),
Some("205"),
"actual from SSE usage"
);
}
#[tokio::test]
async fn upto_stream_forces_include_usage() {
let facilitator = MockServer::start().await;
let upstream = MockServer::start().await;
mount_supported_upto(&facilitator).await;
mount_verify(&facilitator).await;
mount_settle(&facilitator, None).await;
Mock::given(method("POST"))
.and(path("/v1/chat/completions"))
.respond_with(
ResponseTemplate::new(200)
.set_body_string(sse_usage_body())
.insert_header("content-type", "text/event-stream"),
)
.mount(&upstream)
.await;
let (app, state) = upto_app(&facilitator.uri(), &upstream.uri());
let (body, _) = unpaid_402(app.clone()).await;
let signature = signature_from_402(&body);
let response = app
.oneshot(
Request::post("/v1/chat/completions")
.header("content-type", "application/json")
.header(PAYMENT_SIGNATURE, signature)
.body(Body::from(
r#"{"model":"gpt-4o-mini","messages":[{"role":"user","content":"hi"}],"stream":true,"stream_options":{"include_usage":false}}"#,
))
.expect("request"),
)
.await
.expect("response");
assert_eq!(response.status(), StatusCode::OK, "stream");
drop(bytes_of(response).await);
state
.inflight()
.wait_for_drain(Duration::from_secs(2))
.await;
let received = upstream.received_requests().await.expect("recorded");
let request = received.first().expect("upstream request");
let forwarded: serde_json::Value = serde_json::from_slice(&request.body).expect("json");
assert_eq!(
forwarded.pointer("/stream_options/include_usage"),
Some(&serde_json::Value::Bool(true)),
"include_usage forced: {forwarded}"
);
}
#[tokio::test]
async fn upto_stream_abort_settles_abort_usage() {
let facilitator = MockServer::start().await;
mount_supported_upto(&facilitator).await;
mount_verify(&facilitator).await;
mount_settle(&facilitator, None).await;
let (base, thread) = spawn_open_sse();
let (app, state) = upto_app(&facilitator.uri(), &base);
let (body, _) = unpaid_402(app.clone()).await;
let signature = signature_from_402(&body);
let response = paid_stream(app, &signature).await;
assert_eq!(response.status(), StatusCode::OK, "stream");
let mut stream = response.into_body().into_data_stream();
let first = next_containing(&mut stream, b"first")
.await
.expect("first event");
assert!(
first
.windows(b"first".len())
.any(|window| window == b"first"),
"first"
);
assert_eq!(
count_path(
&facilitator.received_requests().await.expect("recorded"),
"/settle"
),
0,
"no settle before abort"
);
drop(stream);
state
.inflight()
.wait_for_drain(Duration::from_secs(2))
.await;
thread.join().expect("sse thread");
let recorded = facilitator.received_requests().await.expect("recorded");
assert_eq!(count_path(&recorded, "/settle"), 1, "one abort settle");
assert_eq!(
settle_amount(&recorded).as_deref(),
Some("29040"),
"abort_usage ceiling"
);
}
#[tokio::test]
async fn upto_stream_upstream_4xx_does_not_settle() {
let facilitator = MockServer::start().await;
let upstream = MockServer::start().await;
mount_supported_upto(&facilitator).await;
mount_verify(&facilitator).await;
mount_settle(&facilitator, None).await;
Mock::given(method("POST"))
.and(path("/v1/chat/completions"))
.respond_with(
ResponseTemplate::new(400)
.set_body_string("data: error\n\n")
.insert_header("content-type", "text/event-stream"),
)
.mount(&upstream)
.await;
let app = upto_router(&facilitator.uri(), &upstream.uri());
let (body, _) = unpaid_402(app.clone()).await;
let signature = signature_from_402(&body);
let response = paid_stream(app, &signature).await;
assert_eq!(response.status(), StatusCode::BAD_REQUEST, "passthrough");
assert!(
response.headers().get(PAYMENT_RESPONSE).is_none(),
"no receipt on 4xx stream"
);
assert_eq!(
count_path(
&facilitator.received_requests().await.expect("recorded"),
"/settle"
),
0,
"no settle"
);
}
#[tokio::test]
async fn embeddings_upto_settles_prompt_only() {
let facilitator = MockServer::start().await;
let upstream = MockServer::start().await;
mount_supported_upto(&facilitator).await;
mount_verify(&facilitator).await;
mount_settle(&facilitator, None).await;
Mock::given(method("POST"))
.and(path("/v1/embeddings"))
.respond_with(ResponseTemplate::new(200).set_body_json(json!({
"data": [],
"usage": {"prompt_tokens": 8191, "total_tokens": 8191}
})))
.mount(&upstream)
.await;
let app = upto_router(&facilitator.uri(), &upstream.uri());
let challenge = app
.clone()
.oneshot(
Request::post("/v1/embeddings")
.header("content-type", "application/json")
.body(Body::from(
r#"{"model":"text-embedding-3-small","input":"hi"}"#,
))
.expect("request"),
)
.await
.expect("response");
assert_eq!(
challenge.status(),
StatusCode::PAYMENT_REQUIRED,
"unpaid embeddings"
);
let body = bytes_of(challenge).await;
let required: PaymentRequired = serde_json::from_slice(&body).expect("json");
assert_eq!(
required
.accepts
.first()
.map(|accept| accept.amount.as_str()),
Some("173"),
"embeddings ceiling"
);
let signature = signature_from_402(&body);
let paid = app
.oneshot(
Request::post("/v1/embeddings")
.header("content-type", "application/json")
.header(PAYMENT_SIGNATURE, signature)
.body(Body::from(
r#"{"model":"text-embedding-3-small","input":"hi"}"#,
))
.expect("request"),
)
.await
.expect("response");
assert_eq!(paid.status(), StatusCode::OK, "paid embeddings");
let recorded = facilitator.received_requests().await.expect("recorded");
assert_eq!(
settle_amount(&recorded).as_deref(),
Some("173"),
"embeddings meter"
);
}