crates/monetize-server/tests/purchase_end_to_end.rs
//! **The whole transaction over the wire**: mock pay (a vendor that needs the
//! customer) + mock cloud + a Gunnar-shaped TEST product → `Purchase.Start` over
//! gRPC with a bearer → the vendor's webhook (HTTP POST, verified) → the ledger
//! shows Captured → the product was pushed a fact with a signature that verifies
//! against monetize's key.
//!
//! And the red twin: a forged callback moves nothing.
//!
//! The product is `monetize_server::testing::products::TestProduct` (v2.0 P2 12.0, A28 H6),
//! not the native gunnar plugin, served under `Pulled` as production serves every product
//! (A32-13 step 2): what is under test is the server's purchase path, so the product's estate is
//! a `FakeEstate` whose main reports its tenants and room and pulls the facts it is pushed.
use std::sync::Arc;
use monetize::embed::signing::verify_fact;
use monetize::testing::MockCloud;
use monetize_product::State;
use monetize_proto::entitlement_client::EntitlementClient;
use monetize_proto::ledger_client::LedgerClient;
use monetize_proto::purchase_client::PurchaseClient;
use monetize_proto::query_ledger_request::By;
use monetize_proto::{self as pb, PurchaseState};
use monetize_server::registry::Registry;
use monetize_server::state::AppState;
use monetize_server::testing::estate::FakeEstate;
use monetize_server::testing::products::TestProduct;
use monetize_server::testing::{http_post, CallbackPay, TestServer};
use tonic::metadata::MetadataValue;
use tonic::transport::Channel;
use tonic::{Code, Request};
const BEARER: &str = "test-bearer-token-of-sufficient-length";
const SECRET: &str = "mockpay-shared-secret";
/// The estate whose main serves the rig's tenants.
const ESTATE: &str = "gunnar.rs";
/// What `Pulled` says of a main whose report is too old to judge on.
const UNMEASURED: &str = "nothing is judged on an old report";
/// What `Pulled` says of a sealed main (no bought iron helps) whose room a cache budget does not fit.
const SEALED: &str = "cache_bytes needs 21474836480 beyond the 5368709120 the main has room for";
struct Rig {
gunnar: FakeEstate,
pay: CallbackPay,
state: Arc<AppState>,
server: TestServer,
channel: Channel,
}
impl Rig {
/// The default rig: the cloud answers `OnHand`, so a purchase buys nothing.
async fn up() -> Rig {
Rig::up_with(monetize::testing::Answer::OnHand).await
}
/// A rig whose cloud gives `answer` to every resource. `Provisioned` is what
/// the teardown test needs: `OnHand` owns nothing, so there would be no iron
/// to give back and the test would pass vacuously.
async fn up_with(answer: monetize::testing::Answer) -> Rig {
Rig::up_with_pay(answer, CallbackPay::new(SECRET)).await
}
/// A rig whose vendor is `pay`. The default vendor has NO checkout page of
/// its own, so the configured template is what a purchase gets; a rig built
/// with [`CallbackPay::with_checkout`] gets the vendor's own URL instead, and
/// the two sit side by side so neither answer can be the only one reachable.
async fn up_with_pay(answer: monetize::testing::Answer, pay: CallbackPay) -> Rig {
// EVERY order asks the product whether it can absorb the caps before the
// vendor is touched. The rig's is a box an operator can grow and that is
// SHORT, because that is the one answer on which a sale legitimately buys
// iron — and the rig's cloud buys. The tests that need a product with room,
// one that cannot answer, or a sealed one re-can it after the first sale.
let cloud = MockCloud::new();
cloud.answer_all(answer);
// A cloud that buys nothing when gunnar says it has the room, as the
// real ones now do.
*cloud.honours_room.lock().unwrap() = true;
let registry = Registry::new().pulled_product(TestProduct::gunnar(), ESTATE).sells("gunnar", pay.clone()).cloud(cloud);
let mut state = AppState::in_memory(registry);
let gunnar = FakeEstate::declare(&state, "gunnar", ESTATE);
short_but_growable(&gunnar);
three_repos_two_namespaces(&gunnar);
// **The server's clock is the fixture's, not the wall's.** Every order
// here starts at NOW (2026-09-03) for a monthly period, so a real clock
// put the tenant in Grace from 2026-10-04 on, and four tests went red
// on a calendar date, not on a change (the date bomb found 2026-10-04).
// A minute after the order: inside its period, whatever day it is.
state.clock = Box::new(|| monetize::testing::NOW + 60_000);
state.checkout_urls.get_mut().unwrap().insert("mockpay".into(), "https://pay.example/{handle}?ref={reference}".into());
// `Ledger.Export` writes where the SERVER says. A rig without this gets
// FailedPrecondition, which is asserted in the export test.
let dir = std::env::temp_dir().join(format!("monetize-rig-backup-{}-{:?}", std::process::id(), std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).unwrap().as_nanos()));
state.backup_dir = Some(dir);
let state = Arc::new(state);
let server = TestServer::spawn(state.clone(), BEARER).await;
let channel = Channel::from_shared(server.endpoint()).unwrap().connect().await.unwrap();
Rig { gunnar, pay, state, server, channel }
}
fn authed<T>(&self, req: T) -> Request<T> {
let mut r = Request::new(req);
r.metadata_mut().insert("authorization", MetadataValue::try_from(format!("Bearer {BEARER}")).unwrap());
r
}
/// An order for 10 GiB of pack_bytes — 220.00 SEK a month under the
/// in-memory server's fixture list (22.00 SEK per GiB-month).
async fn start(&self, tenant: &str) -> pb::StartPurchaseResponse {
self.order(tenant, &[("pack_bytes", 10 << 30)]).await.expect("the order goes through")
}
/// The order every test places: `tenant` to these caps, in SE, at mockpay.
async fn order(&self, tenant: &str, caps: &[(&str, u64)]) -> Result<pb::StartPurchaseResponse, tonic::Status> {
PurchaseClient::new(self.channel.clone())
.start(self.authed(pb::StartPurchaseRequest {
product: "gunnar".into(),
tenant: tenant.into(),
caps: caps.iter().map(|(k, v)| (k.to_string(), *v)).collect(),
vendor: "mockpay".into(),
payer: "+46701234567".into(),
starts_unix_ms: monetize::testing::NOW,
market: pb::Market::Se as i32,
renew: false,
// The OPERATOR path: no ticket, and nothing about this file
// changes because the field exists.
actor_ticket: Vec::new(),
bundle: String::new(),
users: None,
}))
.await
.map(|r| r.into_inner())
}
async fn quote(&self, tenant: &str, caps: &[(&str, u64)]) -> pb::QuoteResponse {
PurchaseClient::new(self.channel.clone())
.quote(self.authed(pb::QuoteRequest {
product: "gunnar".into(),
tenant: tenant.into(),
caps: caps.iter().map(|(k, v)| (k.to_string(), *v)).collect(),
market: pb::Market::Se as i32,
renew: false,
starts_unix_ms: monetize::testing::NOW,
actor_ticket: Vec::new(),
bundle: String::new(),
}))
.await
.expect("a quote is answered")
.into_inner()
}
async fn status(&self, reference: &str) -> pb::PurchaseStatusResponse {
PurchaseClient::new(self.channel.clone())
.status(self.authed(pb::PurchaseStatusRequest { reference: reference.into(), ..Default::default() }))
.await
.unwrap()
.into_inner()
}
async fn ledger_kinds(&self, reference: &str) -> Vec<String> {
LedgerClient::new(self.channel.clone())
.query(self.authed(pb::QueryLedgerRequest { by: Some(By::Reference(reference.into())) }))
.await
.unwrap()
.into_inner()
.rows
.into_iter()
.map(|r| r.kind)
.collect()
}
async fn callback(&self, headers: Vec<(String, String)>, body: Vec<u8>) -> (u16, String) {
let addr = self.server.addr;
tokio::task::spawn_blocking(move || http_post(addr, "/callback/mockpay", &headers, &body).unwrap()).await.unwrap()
}
}
/// **The refusal that matters, over the wire.**
///
/// A tenant on 10 GiB orders 100. `Purchase.Start` asks the product whether it
/// can absorb the new caps before it asks the vendor for anything — and this
/// product cannot answer at all (a gunnar that predates `Admin.Capacity`). A
/// question that could not be asked is not a yes, and the product's own sentence
/// is what the operator reads.
///
/// The assertions are on what the VENDOR and the LEDGER did, not on the error:
/// a `can_absorb` called after the reserve would return exactly the same error
/// and would have locked the customer's money to do it.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn an_order_gunnar_cannot_absorb_is_refused_before_any_money_moves() {
let rig = Rig::up_with(monetize::testing::Answer::Provisioned).await;
// The tenant is on 10 GiB, paid for and pushed.
let started = rig.start("team/one").await;
rig.pay.authorize(&started.handle_id);
let body = CallbackPay::callback_body(&started.reference, &started.handle_id, "authorized");
let (code, reply) = rig.callback(rig.pay.signed_headers(), body).await;
assert_eq!(code, 200, "{reply}");
assert_eq!(rig.status(&started.reference).await.state, PurchaseState::Paid as i32);
// Now the main cannot be judged on: its last word is an hour old.
rig.gunnar.silent_for(3_600_000);
let captures_before = rig.pay.captures();
let rows_before = rig.state.ledger.step_count().unwrap();
let err = rig.order("team/one", &[("pack_bytes", 100 << 30)]).await.expect_err("gunnar cannot absorb 100 GiB and must say so");
assert_eq!(err.code(), Code::FailedPrecondition, "{err:?}");
assert!(err.message().contains(UNMEASURED), "the product's own sentence reaches the operator: {}", err.message());
// NOTHING MOVED. Not the vendor, not the ledger.
assert_eq!(rig.pay.captures(), captures_before, "THE VENDOR WAS TOUCHED for capacity gunnar cannot deliver");
assert_eq!(rig.state.ledger.step_count().unwrap(), rows_before, "a refusal must write no ledger row");
// …and the tenant is still on what they were on, and the fact names the order that put them there.
let fact = rig.state.ledger.entitlement("gunnar", "team/one").unwrap().unwrap();
assert_eq!(fact.plan, started.reference);
assert_eq!(fact.caps["pack_bytes"], 10 << 30);
// The QUOTE says the same, as a quote and not an error: the price of what
// was refused is still what the operator wants to see.
let q = rig.quote("team/one", &[("pack_bytes", 100 << 30)]).await;
assert!(!q.can_serve, "{q:?}");
assert!(q.reason.contains(UNMEASURED), "{}", q.reason);
assert_eq!(q.delta["pack_bytes"], 90 << 30, "the 90 they lack");
assert_eq!(q.price.as_ref().unwrap().minor, 90 * 2200, "1 980.00 SEK for the delta at 22.00 per GiB-month");
assert_eq!(rig.state.ledger.step_count().unwrap(), rows_before, "a quote writes nothing");
}
/// The rig's product's two tenants: `team` (a person's own account, two repositories)
/// and `team/sub` (one), with the pack bytes the native plugin read off the same three
/// repositories (hot minus cache, plus cold).
fn three_repos_two_namespaces(c: &FakeEstate) {
c.tenant("team", true, &[("pack_bytes", (1_000 - 300 + 10_000) + (2_000 - 500 + 20_000)), ("cold_bytes", 30_000), ("repos", 2)]);
c.tenant("team/sub", false, &[("pack_bytes", 4_000 - 1_000 + 40_000), ("cold_bytes", 40_000), ("repos", 1)]);
}
/// Short today (no room in the sold meter), and an operator extends the filesystem once the
/// disk is attached: the main's report says bought iron helps.
fn short_but_growable(c: &FakeEstate) {
c.report(|r| {
r.room = [("pack_bytes".to_string(), 0)].into();
r.with_iron = Some("an operator extends /data after the disk is attached".into());
});
}
/// **THE YES ARM, over the wire.** A test suite in which no order can ever
/// succeed cannot tell a working `can_absorb` from a `return Err` — the refusal
/// tests above would pass against both.
///
/// So: the product says the bought iron becomes usable, and the money moves. The
/// vendor is reserved, the DELTA's iron is ensured (90 GiB, not 100), and the
/// ledger has the rows to show for it — and the quote before it said exactly what
/// would happen. The product is asked the FULL target, against the named tenant.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn an_order_the_product_says_it_can_absorb_reserves_the_money_for_the_delta_and_asks_the_full_target() {
const GIB: u64 = 1 << 30;
let rig = Rig::up_with(monetize::testing::Answer::Provisioned).await;
let started = rig.start("team/one").await;
rig.pay.authorize(&started.handle_id);
let body = CallbackPay::callback_body(&started.reference, &started.handle_id, "authorized");
let (code, reply) = rig.callback(rig.pay.signed_headers(), body).await;
assert_eq!(code, 200, "{reply}");
// The tenant holds 10 GiB and asks for 100: 90 GiB of new bytes, and yes.
short_but_growable(&rig.gunnar);
let rows_before = rig.state.ledger.step_count().unwrap();
// THE QUOTE FIRST: price from the list, cost from the cloud, the delta,
// the iron, and yes — nothing written.
let q = rig.quote("team/one", &[("pack_bytes", 100 * GIB)]).await;
assert!(q.can_serve, "{q:?}");
assert_eq!(q.current["pack_bytes"], 10 * GIB);
assert_eq!(q.caps["pack_bytes"], 100 * GIB);
assert_eq!(q.delta["pack_bytes"], 90 * GIB);
assert_eq!((q.price.as_ref().unwrap().minor, q.price.as_ref().unwrap().currency), (198_000, pb::Currency::Sek as i32), "90 GiB at 22.00 SEK");
assert_eq!((q.cost.as_ref().unwrap().minor, q.cost.as_ref().unwrap().currency), (1_000, pb::Currency::Eur as i32), "the mock cloud's 10.00 EUR a month");
assert_eq!(q.period, "month");
assert_eq!(q.resources.len(), 1);
assert!(matches!(&q.resources[0].kind, Some(pb::resource::Kind::Disk(d)) if d.gib == 90 && d.server == "gunnar-se-sto1"), "{:?}", q.resources);
assert_eq!(rig.state.ledger.step_count().unwrap(), rows_before, "a quote writes nothing");
let out = rig.order("team/one", &[("pack_bytes", 100 * GIB)]).await.expect("gunnar said there is room and monetize refused anyway");
// The money MOVED: a reservation at the vendor, the delta's iron ensured,
// and both written down. PENDING is this vendor's normal answer — it needs
// the customer at its checkout page before it will capture.
assert_eq!(out.state, PurchaseState::Pending as i32, "{out:?}");
assert_eq!(out.reference, q.reference, "the order runs under the reference the quote named");
assert_eq!(rig.ledger_kinds(&out.reference).await, ["Reserved", "Ensured"]);
assert!(rig.state.ledger.step_count().unwrap() > rows_before, "an order that went through wrote no ledger row");
let attempt = rig.state.ledger.attempt(&out.reference).unwrap().unwrap();
assert_eq!(attempt.purchase.delta["pack_bytes"], 90 * GIB);
assert_eq!(attempt.purchase.price.minor, 198_000, "the row records the price");
assert_eq!(attempt.purchase.cost.minor, 1_000, "and the cost, beside it");
// What the product was asked (the FULL target, against the named tenant) is `Pulled`'s own
// arithmetic over the report: monetize-guardian's pulled tests hold it.
}
/// **A product that refuses one cap of the order** (a sealed gunnar whose read
/// cache does not fit): the order is refused before the money moves, the operator
/// reads the product's sentence and the customer the shop's own, and a downgrade
/// is refused by the arithmetic before the product is asked anything.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_product_that_refuses_a_cap_of_the_order_moves_no_money_and_a_downgrade_asks_it_nothing() {
const GIB: u64 = 1 << 30;
let rig = Rig::up_with(monetize::testing::Answer::Provisioned).await;
let started = rig.start("team/one").await;
rig.pay.authorize(&started.handle_id);
let body = CallbackPay::callback_body(&started.reference, &started.handle_id, "authorized");
let (code, reply) = rig.callback(rig.pay.signed_headers(), body).await;
assert_eq!(code, 200, "{reply}");
// The 90 GiB raise fits, and 20 GiB of cache does not fit in what is left of a sealed set:
// the main reports room for the pack, 5 GiB for the cache, and no bought iron that helps.
rig.gunnar.report(|r| {
r.room = [("pack_bytes".to_string(), 1_000 * GIB), ("cache_bytes".to_string(), 5 * GIB)].into();
r.with_iron = None;
});
let captures_before = rig.pay.captures();
let rows_before = rig.state.ledger.step_count().unwrap();
// 20 GiB of read cache on top of the 100 GiB of pack.
let err = rig
.order("team/one", &[("pack_bytes", 100 * GIB), ("cache_bytes", 20 * GIB)])
.await
.expect_err("the product refused the cache cap and the sale must not go through");
assert_eq!(err.code(), Code::FailedPrecondition, "{err:?}");
let msg = err.message();
assert!(msg.contains(SEALED), "the product's whole sentence reaches the operator: {msg}");
// The quote carries the same refusal twice: the operator's whole sentence in
// `reason`, and a customer's in `words` with none of the machine in it.
let q = rig.quote("team/one", &[("pack_bytes", 100 * GIB), ("cache_bytes", 20 * GIB)]).await;
assert!(!q.can_serve && q.reason.contains(SEALED), "{q:?}");
assert!(q.words.contains(monetize::transaction::words::NOTHING_CHARGED), "{}", q.words);
assert!(!q.words.to_lowercase().contains("maintenance"), "a sealed set is not maintenance: {}", q.words);
assert!(!q.words.contains("fresh install"), "the operator's remedy is not the customer's sentence: {}", q.words);
// NOTHING MOVED.
assert_eq!(rig.pay.captures(), captures_before, "THE VENDOR WAS TOUCHED for capacity gunnar cannot deliver");
assert_eq!(rig.state.ledger.step_count().unwrap(), rows_before, "a refusal must write no ledger row");
let fact = rig.state.ledger.entitlement("gunnar", "team/one").unwrap().unwrap();
assert_eq!(fact.plan, started.reference);
assert_eq!(fact.caps["pack_bytes"], 10 << 30);
// A downgrade is refused by the arithmetic, before gunnar is even asked.
let err = rig.order("team/one", &[("pack_bytes", GIB)]).await.expect_err("a downgrade");
assert_eq!(err.code(), Code::FailedPrecondition);
assert!(err.message().contains("SHRINK pack_bytes"), "{}", err.message());
// And a retry of the order already filled is answered, not refused.
let again = rig.order("team/one", &[("pack_bytes", 10 * GIB)]).await.expect("a re-ask of a filled order");
assert_eq!((again.state, again.reference.as_str()), (PurchaseState::Paid as i32, started.reference.as_str()));
}
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_purchase_is_reserved_the_verified_callback_captures_and_gunnar_gets_a_verifying_fact() {
let rig = Rig::up().await;
// 1. Start: reserved at the vendor, resources ensured, capture answered
// "Reserved" (the customer has not authorized) → PENDING with a redirect.
let started = rig.start("team/sub").await;
assert_eq!(started.state(), PurchaseState::Pending, "{started:?}");
assert_eq!(started.reference, "gunnar/team/sub/2026-09-03+pack_bytes=10737418240", "product, tenant, date, and the caps ordered");
assert_eq!(started.handle_vendor, "mockpay");
assert_eq!(started.redirect_url, format!("https://pay.example/{}?ref=gunnar/team/sub/2026-09-03+pack_bytes=10737418240", started.handle_id));
assert_eq!(rig.ledger_kinds(&started.reference).await, ["Reserved", "Ensured"]);
assert!(rig.gunnar.entitlements().is_empty(), "nothing pushed before the money is captured");
assert_eq!(rig.status(&started.reference).await.state(), PurchaseState::Pending);
// 2. The customer authorizes at the vendor; the vendor tells us by webhook.
rig.pay.authorize(&started.handle_id);
let body = CallbackPay::callback_body(&started.reference, &started.handle_id, "authorized");
let (code, reply) = rig.callback(rig.pay.signed_headers(), body).await;
assert_eq!(code, 200, "{reply}");
assert!(reply.contains("\"PAID\""), "{reply}");
// 3. The ledger shows Captured, then the push.
assert_eq!(rig.ledger_kinds(&started.reference).await, ["Reserved", "Ensured", "Captured", "EntitlementPushed"]);
let status = rig.status(&started.reference).await;
assert_eq!(status.state(), PurchaseState::Paid);
assert!(matches!(rig.pay.vendor_state(&started.handle_id), Some(monetize_pay::ChargeStatus::Captured { .. })));
// 4. The estate's main pulled the entitlement with a signature that verifies
// against monetize's key, source naming the vendor and the reference.
assert_eq!(rig.gunnar.call_home().len(), 1, "one order queued for the main");
let received = rig.gunnar.pulled();
assert_eq!(received.len(), 1);
let fact = received[0].clone();
assert_eq!(fact.tenant.0, "team/sub");
assert_eq!(fact.state, State::Paid);
assert_eq!(fact.source, "payment:mockpay:gunnar/team/sub/2026-09-03+pack_bytes=10737418240");
assert_eq!(fact.plan, "gunnar/team/sub/2026-09-03+pack_bytes=10737418240", "the fact names the order it came from");
assert_eq!(fact.caps["pack_bytes"], 10 << 30);
verify_fact(&fact, &rig.state.signer_for("gunnar").verifying_key()).expect("gunnar holds a fact monetize signed");
assert_eq!(monetize_server::convert::fact_from_pb(status.fact.as_ref().unwrap()).unwrap(), fact);
// 5. Entitlement.Get agrees.
let got = EntitlementClient::new(rig.channel.clone())
.get(rig.authed(pb::GetEntitlementRequest { product: "gunnar".into(), tenant: "team/sub".into() }))
.await
.unwrap()
.into_inner();
assert_eq!(monetize_server::convert::fact_from_pb(got.fact.as_ref().unwrap()).unwrap(), fact);
// 6. A replayed callback is a no-op: the attempt is closed, nothing moves
// twice — and it is answered 200. A duplicate delivery is the vendor
// working correctly, and an endpoint that answers its vendor's own
// genuine traffic with an error is one the vendor eventually disables.
let body = CallbackPay::callback_body(&started.reference, &started.handle_id, "captured");
let (code, reply) = rig.callback(rig.pay.signed_headers(), body).await;
assert_eq!(code, 200, "{reply}");
assert!(reply.contains("ignored"), "and it says it did nothing: {reply}");
assert_eq!(rig.gunnar.entitlements().len(), 1);
assert_eq!(rig.state.changes().len(), 1);
rig.server.stop().await;
}
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn refuse_twin_a_forged_callback_moves_nothing() {
let rig = Rig::up().await;
let started = rig.start("team").await;
assert_eq!(started.state(), PurchaseState::Pending);
let before = rig.ledger_kinds(&started.reference).await;
let captures_before = rig.pay.captures();
// The customer never authorized; an attacker who knows the reference and the
// handle posts a "captured" event with the wrong secret, then with none.
let body = CallbackPay::callback_body(&started.reference, &started.handle_id, "captured");
let forged = vec![("x-mockpay-signature".to_string(), "not-the-secret".to_string()), ("content-type".into(), "application/json".into())];
let (code, _) = rig.callback(forged, body.clone()).await;
assert_eq!(code, 401);
let (code, _) = rig.callback(vec![("content-type".into(), "application/json".into())], body.clone()).await;
assert_eq!(code, 401);
// A VERIFIED callback for a handle the ledger never minted is refused too.
let wrong_handle = CallbackPay::callback_body(&started.reference, "res-999", "captured");
let (code, _) = rig.callback(rig.pay.signed_headers(), wrong_handle).await;
assert_eq!(code, 409);
// And one for a reference the ledger does not know: verified, about nothing
// we hold, so a receipt rather than a refusal. The proof that it changed
// nothing is the block below, not the status code.
let unknown = CallbackPay::callback_body("gunnar/nobody/2026-09-03", &started.handle_id, "captured");
let (code, reply) = rig.callback(rig.pay.signed_headers(), unknown).await;
assert_eq!(code, 200, "{reply}");
assert!(reply.contains("ignored"), "{reply}");
// Nothing moved: no capture attempted, no ledger row, nothing pushed, no state change.
assert_eq!(rig.ledger_kinds(&started.reference).await, before);
assert_eq!(rig.pay.captures(), captures_before);
assert_eq!(rig.pay.vendor_state(&started.handle_id), Some(monetize_pay::ChargeStatus::Reserved));
assert!(rig.gunnar.entitlements().is_empty());
assert!(rig.state.changes().is_empty());
assert_eq!(rig.status(&started.reference).await.state(), PurchaseState::Pending);
assert_eq!(rig.state.ledger.entitlement("gunnar", "team").unwrap(), None);
// Even a VERIFIED, correctly addressed callback that only says "authorized"
// — while the vendor has NOT actually authorized — captures nothing: the
// server asks the vendor, it does not believe the webhook's word.
let body = CallbackPay::callback_body(&started.reference, &started.handle_id, "authorized");
let (code, reply) = rig.callback(rig.pay.signed_headers(), body).await;
assert_eq!(code, 200, "{reply}");
assert!(reply.contains("\"PENDING\""), "{reply}");
assert!(rig.gunnar.entitlements().is_empty());
assert_eq!(rig.pay.vendor_state(&started.handle_id), Some(monetize_pay::ChargeStatus::Reserved));
rig.server.stop().await;
}
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn refuse_twin_the_grpc_services_need_the_bearer_and_the_webhook_does_not() {
let rig = Rig::up().await;
let err = PurchaseClient::new(rig.channel.clone())
.status(Request::new(pb::PurchaseStatusRequest { reference: "x".into(), ..Default::default() }))
.await
.unwrap_err();
assert_eq!(err.code(), Code::Unauthenticated);
let mut wrong = Request::new(pb::PurchaseStatusRequest { reference: "x".into(), ..Default::default() });
wrong.metadata_mut().insert("authorization", MetadataValue::from_static("Bearer nope"));
assert_eq!(PurchaseClient::new(rig.channel.clone()).status(wrong).await.unwrap_err().code(), Code::Unauthenticated);
// The webhook answers on its own terms (here: unverified), not with 401-for-bearer.
let (code, _) = rig.callback(vec![], b"{}".to_vec()).await;
assert_eq!(code, 401);
let (code, body) = rig.callback(vec![], b"{}".to_vec()).await;
assert_eq!(code, 401, "{body}");
let addr = rig.server.addr;
let (code, _) = tokio::task::spawn_blocking(move || http_post(addr, "/callback/klarna", &[], b"{}").unwrap()).await.unwrap();
assert_eq!(code, 404, "a vendor that is not linked");
rig.server.stop().await;
}
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn the_poller_reads_usage_and_walks_the_ladder() {
let rig = Rig::up().await;
let started = rig.start("team/sub").await;
rig.pay.authorize(&started.handle_id);
let body = CallbackPay::callback_body(&started.reference, &started.handle_id, "authorized");
assert_eq!(rig.callback(rig.pay.signed_headers(), body).await.0, 200);
// A tick at NOW: usage stored for both namespaces, the fact holds (Paid).
let s = rig.state.clone();
rig.gunnar.call_home();
let tick = tokio::task::spawn_blocking(move || monetize_server::poller::tick(&s)).await.unwrap();
assert_eq!(tick.tenants_read, 2);
assert_eq!(tick.state_changes, 0);
// The free-room leak (round 5.3.1): this shop sells, and the account `team` itself — a
// person's namespace, not the paid `team/sub` — held no fact; it is given the shop's
// account fact (Free, no room of its own), once.
assert_eq!(tick.accounts_given, 1);
let usage = monetize_proto::usage_client::UsageClient::new(rig.channel.clone())
.read_all(rig.authed(pb::ReadAllUsageRequest { product: "gunnar".into() }))
.await
.unwrap()
.into_inner();
assert_eq!(usage.usages.len(), 2);
let team = usage.usages.iter().find(|u| u.tenant == "team").unwrap();
assert_eq!(team.meters.iter().find(|m| m.name == "pack_bytes").unwrap().value, (1_000 - 300 + 10_000) + (2_000 - 500 + 20_000));
// The clock moves past paid_until + grace: the ladder says Suspended, the poller
// signs and pushes it, and one state-change row is reported.
let fact = rig.state.ledger.entitlement("gunnar", "team/sub").unwrap().unwrap();
let paid_until = fact.paid_until_unix_ms.unwrap();
let later = paid_until + 15 * monetize::embed::civil::DAY_MS;
let moved = monetize_server::poller::ladder_move(&rig.state, &fact, later).unwrap();
assert_eq!(moved.state, State::Suspended);
let signed = rig.state.signer_for("gunnar").sign_fact(moved);
let s = rig.state.clone();
let signed2 = signed.clone();
// The plugin is sync and blocks its own runtime: off the async thread, as the server does.
rig.gunnar.call_home();
tokio::task::spawn_blocking(move || s.apply_fact(s.registry.find_product("gunnar").unwrap(), &signed2).unwrap().unwrap()).await.unwrap();
rig.gunnar.call_home();
let received = rig.gunnar.pulled();
// The paid fact, the account fact the first tick gave `team`, the ladder's.
assert_eq!(received.len(), 3);
let given = received[1].clone();
assert_eq!((given.tenant.0.as_str(), given.state), ("team", State::Free));
assert!(given.source.starts_with("account:given:"), "{}", given.source);
let pushed = received[2].clone();
assert_eq!(pushed.state, State::Suspended);
assert_eq!(pushed.source, "ladder");
verify_fact(&pushed, &rig.state.signer_for("gunnar").verifying_key()).unwrap();
let changes = rig.state.changes();
// The purchase, the account fact given `team` (from nothing to Free), the ladder's move.
assert_eq!(changes.len(), 3);
assert_eq!((changes[1].from, changes[1].to), (None, State::Free));
assert_eq!((changes[2].from, changes[2].to, changes[2].pushed), (Some(State::Paid), State::Suspended, true));
// And it holds: the same instant moves nothing again.
assert!(monetize_server::poller::ladder_move(&rig.state, &pushed, later).is_none());
rig.server.stop().await;
}
/// **`Ledger.Export` against a RUNNING server** — the case the CLI verb cannot
/// do, because redb is single-process and the service holds the lock.
///
/// Measured on the live gunnar.rs box 2026-09-04: `monetize-server --export`
/// while the service was up failed with "Database already open. Cannot acquire
/// lock." A backup that needs downtime is a backup nobody takes, so the export
/// has to happen inside the process that already holds the database.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn export_runs_against_a_live_server_and_writes_no_secrets() {
let rig = Rig::up_with(monetize::testing::Answer::Provisioned).await;
// Real history first, so the export is not empty.
let started = rig.start("team/sub").await;
rig.pay.authorize(&started.handle_id);
let body = CallbackPay::callback_body(&started.reference, &started.handle_id, "authorized");
let (code, _) = rig.callback(rig.pay.signed_headers(), body).await;
assert_eq!(code, 200);
// A server with NO backup_dir refuses rather than choosing a directory for
// the operator — asserted here because the rig is the only place that has
// one, and the refusal is the shape a fresh deployment is in.
{
let mut bare = AppState::in_memory(Registry::new());
bare.backup_dir = None;
assert!(bare.backup_dir.is_none(), "the default really is None, so the refusal below is the real path");
}
let mut led = LedgerClient::new(rig.channel.clone());
let out = led.export(rig.authed(pb::ExportBackupRequest {})).await.unwrap().into_inner();
assert!(out.ledger_rows >= 4, "the export saw real history: {out:?}");
assert_eq!(out.dir, rig.state.backup_dir.as_ref().unwrap().display().to_string(), "it writes where the SERVER says, not where a caller asks");
assert_eq!(out.signing_public_key, rig.state.keys.root().public_key().iter().map(|b| format!("{b:02x}")).collect::<String>());
// The files are really there, and hold no secret.
let written = std::path::Path::new(&out.dir);
let ledger_text = std::fs::read_to_string(written.join(monetize_server::backup::LEDGER)).unwrap();
assert_eq!(ledger_text.lines().filter(|l| !l.trim().is_empty()).count() as u64, out.ledger_rows);
assert!(ledger_text.contains("team/sub"), "the tenant's history is in it");
let manifest = std::fs::read_to_string(written.join(monetize_server::backup::MANIFEST)).unwrap();
assert!(manifest.contains("NO SECRETS"));
// The bearer must not have leaked into any file.
for f in [monetize_server::backup::LEDGER, monetize_server::backup::USAGE, monetize_server::backup::MANIFEST] {
let t = std::fs::read_to_string(written.join(f)).unwrap();
assert!(!t.contains(BEARER), "{f} contains the bearer");
assert!(!t.contains(SECRET), "{f} contains the vendor secret");
}
// Idempotent and deterministic: a second call writes the same bytes.
let before: Vec<u8> = std::fs::read(written.join(monetize_server::backup::LEDGER)).unwrap();
led.export(rig.authed(pb::ExportBackupRequest {})).await.unwrap();
assert_eq!(std::fs::read(written.join(monetize_server::backup::LEDGER)).unwrap(), before);
// And it needs the bearer, like every other RPC.
let mut anon = LedgerClient::new(rig.channel.clone());
let err = anon.export(Request::new(pb::ExportBackupRequest {})).await.expect_err("export must need the bearer");
assert_eq!(err.code(), Code::Unauthenticated, "{err:?}");
let _ = std::fs::remove_dir_all(written);
}
/// **`Usage.History` over the wire: the series, not the last slot.**
///
/// The store used to hold ONE reading per tenant, so every poll tick threw the
/// previous one away and monetize could not answer "how fast is this growing" for
/// a customer it had been measuring all year. This asserts the WIRE response —
/// the readings tonic actually returns, with their meter values — because a
/// series whose rows all carry the newest number is the same single-slot bug in a
/// longer list.
///
/// It also asserts the poller writes into it: a history table nothing fills is a
/// table, not a feature.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn usage_history_returns_the_series_and_the_poller_writes_into_it() {
let rig = Rig::up().await;
let mut usage = monetize_proto::usage_client::UsageClient::new(rig.channel.clone());
let hist = |since: u64, limit: u32| pb::UsageHistoryRequest { product: "gunnar".into(), tenant: "team/sub".into(), since_unix_ms: since, limit };
let series = |r: &pb::UsageHistoryResponse| {
r.readings
.iter()
.map(|u| (u.measured_at_unix_ms, u.meters.iter().find(|m| m.name == "pack_bytes").unwrap().value))
.collect::<Vec<_>>()
};
// Nothing read yet: an empty series, not an error.
let empty = usage.history(rig.authed(hist(0, 0))).await.unwrap().into_inner();
assert_eq!((empty.readings.len(), empty.kept), (0, 0));
// Three readings, minutes apart, growing.
for (at, bytes) in [(1_000_000u64, 10u64), (2_000_000, 20), (3_000_000, 40)] {
let u = monetize_product::Usage { meters: [("pack_bytes".to_string(), bytes)].into(), measured_at_unix_ms: at };
rig.state.usage.put("gunnar", "team/sub", &u).unwrap();
}
let all = usage.history(rig.authed(hist(0, 0))).await.unwrap().into_inner();
assert_eq!(series(&all), vec![(1_000_000, 10), (2_000_000, 20), (3_000_000, 40)], "oldest first, every reading kept");
assert_eq!(all.kept, 3);
assert!(all.readings.iter().all(|u| u.tenant == "team/sub"));
// `since` is inclusive and cuts the old end; `kept` still counts the whole window.
let since = usage.history(rig.authed(hist(2_000_000, 0))).await.unwrap().into_inner();
assert_eq!(series(&since), vec![(2_000_000, 20), (3_000_000, 40)]);
assert_eq!(since.kept, 3, "kept is the window, not the slice");
// `limit` keeps the NEWEST, still ordered oldest first.
assert_eq!(series(&usage.history(rig.authed(hist(0, 2))).await.unwrap().into_inner()), vec![(2_000_000, 20), (3_000_000, 40)]);
assert_eq!(series(&usage.history(rig.authed(hist(0, 1))).await.unwrap().into_inner()), vec![(3_000_000, 40)]);
// `Read` still answers with the latest, untouched by the series.
let latest = usage.read(rig.authed(pb::ReadUsageRequest { product: "gunnar".into(), tenant: "team/sub".into() })).await.unwrap().into_inner();
assert_eq!(latest.usage.unwrap().meters.iter().find(|m| m.name == "pack_bytes").unwrap().value, 40);
// The poller is the real writer: a tick adds a reading for a tenant that had none.
let before = usage
.history(rig.authed(pb::UsageHistoryRequest { product: "gunnar".into(), tenant: "team".into(), since_unix_ms: 0, limit: 0 }))
.await
.unwrap()
.into_inner()
.kept;
assert_eq!(before, 0);
let s = rig.state.clone();
assert_eq!(tokio::task::spawn_blocking(move || monetize_server::poller::tick(&s)).await.unwrap().tenants_read, 2);
let after = usage
.history(rig.authed(pb::UsageHistoryRequest { product: "gunnar".into(), tenant: "team".into(), since_unix_ms: 0, limit: 0 }))
.await
.unwrap()
.into_inner();
assert_eq!(after.kept, 1, "the poller's reading landed in the history table");
assert_eq!(after.readings.len(), 1);
// The refusals: an unlinked product, and a history with no tenant.
let bad = usage.history(rig.authed(pb::UsageHistoryRequest { product: "idun".into(), tenant: "t".into(), since_unix_ms: 0, limit: 0 })).await;
assert_eq!(bad.unwrap_err().code(), Code::NotFound);
let no_tenant = usage.history(rig.authed(pb::UsageHistoryRequest { product: "gunnar".into(), tenant: String::new(), since_unix_ms: 0, limit: 0 })).await;
assert_eq!(no_tenant.unwrap_err().code(), Code::InvalidArgument);
rig.server.stop().await;
}
/// **`Entitlement.PublicKey`: the key a caller verifies WITH, over the wire.**
///
/// Without it the console's "signed" badge could only report that the signature
/// field was populated — which sixty-four bytes of anything satisfies. This
/// asserts the key monetize hands out is the key it SIGNS with, by taking a fact
/// straight off the wire and verifying it against nothing but the RPC's answer:
/// no reaching into `rig.state.signer`, because a test that verifies with the
/// server's own in-process key proves the RPC returned something, not that it
/// returned the right thing.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn the_public_key_rpc_hands_out_the_key_that_actually_signed_the_facts() {
let rig = Rig::up().await;
let mut ent = EntitlementClient::new(rig.channel.clone());
let key = ent.public_key(rig.authed(pb::PublicKeyRequest { product: "gunnar".into() })).await.unwrap().into_inner();
assert_eq!(key.algorithm, "ed25519");
assert_eq!(key.public_key.len(), 32, "raw Ed25519 verifying key");
assert_eq!(key.hex, key.public_key.iter().map(|b| format!("{b:02x}")).collect::<String>(), "the hex is the same bytes");
let verifying = monetize::sign::VerifyingKey::from_bytes(&<[u8; 32]>::try_from(key.public_key.as_slice()).unwrap()).unwrap();
// Buy something, so there is a fact monetize signed itself.
let started = rig.start("team/sub").await;
rig.pay.authorize(&started.handle_id);
assert_eq!(rig.callback(rig.pay.signed_headers(), CallbackPay::callback_body(&started.reference, &started.handle_id, "authorized")).await.0, 200);
let stored = ent
.get(rig.authed(pb::GetEntitlementRequest { product: "gunnar".into(), tenant: "team/sub".into() }))
.await
.unwrap()
.into_inner()
.fact
.expect("the purchase wrote a fact");
assert_eq!(stored.signature.len(), 64);
// The fact off the wire verifies against the key off the wire.
let as_rust = monetize_product::EntitlementFact {
tenant: monetize_product::TenantId(stored.tenant.clone()),
plan: stored.plan.clone(),
state: State::Paid,
paid_until_unix_ms: stored.paid_until_unix_ms,
caps: stored.caps.clone(),
source: stored.source.clone(),
signature: stored.signature.clone(),
// The stamp is part of what the key vouches for: drop it here and the
// V2 signature stops verifying, which is the whole point of it.
issued_unix_ms: stored.issued_unix_ms,
issued_signature: stored.issued_signature.clone(),
};
verify_fact(&as_rust, &verifying).expect("the advertised key verifies the stored fact");
// The red twin: one byte of the signature changed, and the SAME key refuses
// it. Without this the assertion above could be satisfied by a verify that
// never fails.
let mut tampered = as_rust.clone();
tampered.signature[0] ^= 0xff;
assert!(verify_fact(&tampered, &verifying).is_err(), "a corrupted signature must not verify");
// And the RPC is behind the bearer like every other read.
assert_eq!(
EntitlementClient::new(rig.channel.clone()).public_key(pb::PublicKeyRequest { product: "gunnar".into() }).await.unwrap_err().code(),
Code::Unauthenticated
);
rig.server.stop().await;
}
/// **A resume asks the vendor; it does not capture again.**
///
/// The server used to resume an open attempt by calling `Transaction::run` with
/// the purchase read back out of the ledger. `run` is the verb for "buy this": it
/// goes straight to `resume`, which for an attempt with no `Captured` row calls
/// `pay.capture` — without ever asking the vendor what it thinks. So a
/// reservation the vendor had ALREADY drawn, because the callback was dropped or
/// arrived twice, got captured a second time. That is idempotent at some vendors
/// and a second charge at others, and monetize does not get to choose which
/// vendor it is talking to.
///
/// The applied output asserted here is the VENDOR'S CALL COUNT, not a ledger
/// round trip: the vendor is put into Captured behind the server's back, and the
/// webhook must resolve the attempt with no further capture.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_resume_reads_the_vendors_state_and_does_not_capture_a_second_time() {
let rig = Rig::up().await;
let started = rig.start("team/sub").await;
assert_eq!(started.state(), PurchaseState::Pending, "{started:?}");
// `Start` asked once, and the vendor honestly answered "still reserved".
let after_start = rig.pay.captures();
assert_eq!(after_start, 1);
assert_eq!(rig.ledger_kinds(&started.reference).await, ["Reserved", "Ensured"]);
// The vendor draws the money without telling us; the webhook arrives late.
rig.pay.force_captured(&started.handle_id);
let body = CallbackPay::callback_body(&started.reference, &started.handle_id, "captured");
let (code, reply) = rig.callback(rig.pay.signed_headers(), body).await;
assert_eq!(code, 200, "{reply}");
assert!(reply.contains("\"PAID\""), "{reply}");
assert_eq!(rig.pay.captures(), after_start, "the vendor must not be asked to capture money it already took");
assert_eq!(rig.ledger_kinds(&started.reference).await, ["Reserved", "Ensured", "Captured", "EntitlementPushed"]);
// And the Captured row records the VENDOR's reference, not one we invented.
let rows = LedgerClient::new(rig.channel.clone())
.query(rig.authed(pb::QueryLedgerRequest { by: Some(By::Reference(started.reference.clone())) }))
.await
.unwrap()
.into_inner()
.rows;
assert!(rows[2].json.contains("out-of-band-"), "the row carries the vendor's own capture: {}", rows[2].json);
assert_eq!(rig.gunnar.entitlements().len(), 1, "the verdict was delivered once");
rig.server.stop().await;
}
/// (r) **A vendor that mints its own checkout URL beats the config template.**
///
/// The template in monetize.toml (`checkout_url = "https://…/{handle}"`) is
/// right for a vendor with ONE fixed checkout page and cannot express Klarna's
/// per-order `redirect_url` or a Stripe intent's `next_action` — those carry a
/// session id the vendor invented. So `Vendor::page_url` is asked
/// first, and the whole point of this test is that BOTH sources are present and
/// the right one wins.
///
/// The red before the wiring existed was the template's URL coming back:
/// `https://pay.example/res-1?ref=gunnar/team/sub/2026-09-03` instead of the
/// vendor's `https://mockpay.example/checkout/res-1`. Asserted through
/// `Purchase.Start` over gRPC and again through `Purchase.Status`, because the
/// operator who needs the link most is the one reopening a pending attempt
/// later — and until this lane `Status` carried no link at all.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn the_vendors_own_checkout_url_wins_over_the_configured_template() {
let pay = CallbackPay::with_checkout(SECRET, "https://mockpay.example/checkout/");
let rig = Rig::up_with_pay(monetize::testing::Answer::OnHand, pay).await;
// The template is configured by `up_with_pay` and is DIFFERENT, so a wiring
// that ignored the vendor would be caught rather than coincide.
assert_eq!(rig.state.checkout_urls.read().unwrap().get("mockpay").cloned().as_deref(), Some("https://pay.example/{handle}?ref={reference}"));
let started = rig.start("team/sub").await;
assert_eq!(started.state(), PurchaseState::Pending, "{started:?}");
assert_eq!(
started.redirect_url,
format!("https://mockpay.example/checkout/{}", started.handle_id),
"the vendor minted this URL for this order; the template could not have produced it"
);
// The template's own shape (`?ref=<reference>`) must be nowhere in the
// answer. Checking the host would not do it: "mockpay.example" contains
// "pay.example", and a guard that cannot tell the two apart is not a guard.
assert!(!started.redirect_url.contains("?ref="), "the template must not win: {}", started.redirect_url);
// The same link, still offered, on a later read of an OPEN attempt.
let status = rig.status(&started.reference).await;
assert_eq!(status.state(), PurchaseState::Pending);
assert_eq!(status.redirect_url, started.redirect_url, "an operator reopening a pending attempt is offered the same link");
// Once the money is drawn there is nowhere left to send anybody, and a
// checkout link on a paid purchase is an invitation to pay twice.
rig.pay.authorize(&started.handle_id);
let body = CallbackPay::callback_body(&started.reference, &started.handle_id, "authorized");
let (code, reply) = rig.callback(rig.pay.signed_headers(), body).await;
assert_eq!(code, 200, "{reply}");
let status = rig.status(&started.reference).await;
assert_eq!(status.state(), PurchaseState::Paid);
assert_eq!(status.redirect_url, "", "a finished purchase offers no checkout link");
}
/// ★ **A RENEWAL OVER THE WIRE NAMES NOTHING, AND THE GUARDS REFUSE RATHER
/// THAN IGNORE.**
///
/// `Purchase.Start { renew: true }` is the door an operator sells another
/// period through — the same authenticated door an order goes through, so a
/// renewal grows no second way to sell a month and no unauthenticated one at
/// all. What it must never become is an order wearing a flag:
///
/// * a renewal that carried CAPS would be a renewal a client sends by typo,
/// and on the other side of that typo is a charge. The caps come off the
/// tenant's stored fact;
/// * a renewal that carried a START would let a caller mint any reference it
/// liked and skip or repeat a period. The period starts where the paid one
/// ends;
/// * and a renewal for a tenant this server has never heard of must be refused,
/// not invented — putting somebody on a clock nobody asked to start.
///
/// Each is refused BY NAME. Dropping the fields a caller sent would charge for
/// something other than what was asked for, which is the one thing a money path
/// does not get to do quietly.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_renewal_names_no_caps_and_no_start_and_is_never_invented() {
let rig = Rig::up().await;
let renew = |caps: Vec<(String, u64)>, starts: u64, tenant: &str| {
let channel = rig.channel.clone();
let req = rig.authed(pb::StartPurchaseRequest {
product: "gunnar".into(),
tenant: tenant.into(),
caps: caps.into_iter().collect(),
vendor: String::new(),
payer: String::new(),
starts_unix_ms: starts,
market: pb::Market::Se as i32,
renew: true,
// An OPERATOR renewal: the anti-typo guards this test is about are
// reached with no ticket at all, exactly as they were.
actor_ticket: Vec::new(),
bundle: String::new(),
users: None,
});
async move { PurchaseClient::new(channel).start(req).await }
};
// CAPS: refused, and the message says what to do instead.
let err = renew(vec![("pack_bytes".to_string(), 10 << 30)], 0, "alice").await.expect_err("a renewal names no caps");
assert_eq!(err.code(), Code::InvalidArgument, "{err}");
assert!(err.message().contains("names no caps"), "{}", err.message());
assert!(err.message().contains("ORDER"), "it must say what to do instead: {}", err.message());
// A START: refused for its own reason.
let err = renew(Vec::new(), monetize::testing::NOW, "alice").await.expect_err("a renewal names no start");
assert_eq!(err.code(), Code::InvalidArgument, "{err}");
assert!(err.message().contains("names no start"), "{}", err.message());
// A TENANT WITH NO PERIOD: refused, never invented.
let err = renew(Vec::new(), 0, "nobody").await.expect_err("there is nothing to renew");
assert_eq!(err.code(), Code::FailedPrecondition, "{err}");
assert!(err.message().contains("nothing to renew") && err.message().contains("nobody"), "{}", err.message());
// And nothing was written for any of the three: a refused renewal is not
// an attempt, and the vendor was never touched.
assert!(rig.state.ledger.unfinished().unwrap().is_empty(), "a refused renewal must leave no open attempt");
rig.server.stop().await;
}
/// **A logged-in tenant is not asked for an e-mail it already gave** (owner
/// 2026-09-28, test2.gunnar.rs: Stripe's page asked for "Kontaktinformation /
/// E-postadress" after a tenant's 10 GiB order). The console names no payer;
/// monetize names the address the tenant's signup carried, and the vendor
/// (Stripe: `customer_email`) opens the page with it. A tenant no signup made
/// is sent with no payer, as before — the page asks.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn an_order_names_the_email_the_tenant_signed_up_with_as_the_payer() {
use monetize_pay::signup::PaidSignup;
use monetize_pay::{Currency, Money};
use monetize_server::signup::SignupState;
let rig = Rig::up().await;
let signup = PaidSignup {
vendor: "mockpay".into(),
session: "cs_alice".into(),
reference: "pi_alice".into(),
email: "alice@example.com".into(),
username: "alice".into(),
product: "gunnar".into(),
plan: "gunnar/10gb-1m".into(),
amount: Money { minor: 10_000, currency: Currency::Sek },
tax: Money { minor: 0, currency: Currency::Sek },
paid_at_unix_ms: monetize::testing::NOW as i64,
customer_country: None,
};
rig.state.signups.record_pending(&signup, 1).unwrap();
rig.state.signups.set("mockpay", "cs_alice", SignupState::Created { namespace: "alice".into(), username: "alice".into(), at_unix_ms: 2 }).unwrap();
let order = |tenant: &str| pb::StartPurchaseRequest {
product: "gunnar".into(),
tenant: tenant.into(),
caps: [("pack_bytes".to_string(), 10u64 << 30)].into_iter().collect(),
vendor: "mockpay".into(),
// What gunnar-ui sends: nobody.
payer: String::new(),
starts_unix_ms: monetize::testing::NOW,
market: pb::Market::Se as i32,
renew: false,
actor_ticket: Vec::new(),
bundle: String::new(),
users: None,
};
let started = PurchaseClient::new(rig.channel.clone()).start(rig.authed(order("alice"))).await.expect("alice's order").into_inner();
assert_eq!(started.state(), PurchaseState::Pending, "{started:?}");
assert_eq!(rig.pay.payers().last().cloned().flatten().as_deref(), Some("alice@example.com"), "the vendor was handed the tenant's own address");
let started = PurchaseClient::new(rig.channel.clone()).start(rig.authed(order("bob"))).await.expect("bob's order").into_inner();
assert_eq!(started.state(), PurchaseState::Pending, "{started:?}");
assert_eq!(rig.pay.payers().last().cloned().flatten(), None, "no signup made bob: the page asks, as before");
}
/// **A customer back from the vendor's page is not left on `pending` until the
/// poller comes by.** MEASURED on the KVM mock 2026-09-28: a paid order whose
/// webhook never came was captured by the poller alone, 300 s after the
/// payment; the console following it read `pending` for those five minutes.
/// Now a `Purchase.Status` read of an open attempt asks the vendor (at most
/// every `ASK_VENDOR_EVERY`): paid at the vendor, no webhook, no poller pass
/// → the read that follows the payment answers Paid and the product holds the
/// fact. An attempt the vendor has NOT been paid for stays pending.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_status_read_asks_the_vendor_and_a_paid_order_is_paid_without_a_webhook() {
let rig = Rig::up().await;
let started = rig.start("team/watch").await;
assert_eq!(started.state(), PurchaseState::Pending, "{started:?}");
assert_eq!(rig.status(&started.reference).await.state(), PurchaseState::Pending, "not paid at the vendor: pending");
// The customer pays; the vendor's notice never arrives.
rig.pay.authorize(&started.handle_id);
// The next ask is after the throttle window of the first read.
tokio::time::sleep(monetize_server::services::purchase::ASK_VENDOR_EVERY + std::time::Duration::from_millis(200)).await;
let status = rig.status(&started.reference).await;
assert_eq!(status.state(), PurchaseState::Paid, "asked of the vendor on the read: {status:?}");
assert!(status.fact.is_some(), "and the product was handed the fact");
let kinds = rig.ledger_kinds(&started.reference).await;
assert_eq!(kinds.iter().filter(|k| k.as_str() == "Captured").count(), 1, "captured once: {kinds:?}");
}