crates/monetize-ui-server/tests/screens_feed.rs
//! **The feed, on the wire.** An event the mock monetize publishes must come
//! out of `/api/events` as SSE bytes with its contents, and a monetize that
//! cannot stream must make the console SAY it dropped to polling — the frame
//! is announced, never silent.
//!
//! `tests/api.rs` proves this for a `LedgerAppended`; these hold the two other
//! payloads a screen renders — a usage reading with the fixture's distinct
//! number in it, and a verdict moving from grace to paid with BOTH sides of the
//! move — and that the very first frame of every connection is the remembered
//! mode line, before any delta.
#[path = "../../../uitest/src/fixture.rs"]
mod fixture;
#[path = "../../../uitest/src/mock.rs"]
mod mock;
mod screens_rig;
use fixture as fx;
use http::{Method, StatusCode};
use monetize_proto as pb;
use monetize_ui_server::role::Role;
use screens_rig::{sse_until, Auth, Rig};
async fn wait_for_a_watcher(rig: &Rig, deadline: tokio::time::Instant) {
// Publishing into a broadcast nobody holds is publishing into nothing, and
// the test would then be asserting on a race rather than on the feed.
while rig.mock.watchers.load(std::sync::atomic::Ordering::SeqCst) == 0 {
assert!(tokio::time::Instant::now() < deadline, "the BFF never opened Events.Watch");
tokio::time::sleep(std::time::Duration::from_millis(10)).await;
}
}
/// (r) RED first: with `seq: 3` asserted for the reading's meter value, it
/// fails on `10952200000` — the fixture's number is what travels, and a
/// translator emitting `{"kind":"usage"}` alone would satisfy an `is_some`.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_usage_reading_and_a_verdict_move_arrive_on_the_feed_with_their_contents() {
let rig = Rig::up().await;
let viewer = rig.cookie(Role::Viewer);
let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(20);
wait_for_a_watcher(&rig, deadline).await;
let mock = rig.mock.clone();
let pump = tokio::spawn(async move {
for i in 0..200u64 {
let _ = mock.events.send(pb::Event::new(
10 + i,
fx::T0 + 1,
pb::event::Payload::UsageMeasured(pb::UsageMeasured {
product: "gunnar".into(),
tenant: "alice".into(),
meters: vec![pb::Meter { name: "pack_bytes".into(), value: fx::ALICE_PACK.raw }, pb::Meter { name: "repos".into(), value: fx::ALICE_REPOS }],
measured_at_unix_ms: fx::T0 + 2,
}),
));
let _ = mock.events.send(pb::Event::new(
300 + i,
fx::T0 + 3,
pb::event::Payload::EntitlementChanged(pb::EntitlementChanged {
product: "gunnar".into(),
tenant: "team/sub".into(),
order: "gunnar/team/sub/2026-09-05+pack_bytes=10737418240".into(),
from_state: pb::State::Grace as i32,
had_previous: true,
to_state: pb::State::Paid as i32,
source: "payment:invoice:gunnar/team/sub/2026-09-05".into(),
pushed: true,
}),
));
tokio::time::sleep(std::time::Duration::from_millis(25)).await;
}
});
let (usage, seen) = sse_until(&rig.router, &viewer, deadline, |v| v["kind"] == "usage").await;
assert_eq!((usage["product"].as_str(), usage["tenant"].as_str(), usage["at_unix_ms"].as_u64()), (Some("gunnar"), Some("alice"), Some(fx::T0 + 2)), "the reading's own time: {usage}");
assert_eq!(usage["meters"][0]["name"], "pack_bytes");
assert_eq!(usage["meters"][0]["value"].as_u64(), Some(fx::ALICE_PACK.raw), "the fixture's distinct number, off the stream");
assert_eq!(usage["meters"][1]["value"].as_u64(), Some(fx::ALICE_REPOS));
// The FIRST frame of the connection is the remembered mode line — a tab
// opened mid-outage must be told what it is looking at before any delta.
assert_eq!(seen.first().and_then(|v| v["kind"].as_str()), Some("feed"), "the opening frame is the mode line: {seen:?}");
assert!(seen.iter().any(|v| v["kind"] == "feed" && v["mode"] == "stream"), "told it is LIVE before a streamed delta: {seen:?}");
let (moved, _) = sse_until(&rig.router, &viewer, deadline, |v| v["kind"] == "entitlement").await;
assert_eq!((moved["tenant"].as_str(), moved["from_state"].as_str(), moved["state"].as_str()), (Some("team/sub"), Some("grace"), Some("paid")), "both sides of the move: {moved}");
assert_eq!((moved["source"].as_str(), moved["pushed"].as_bool()), (Some("payment:invoice:gunnar/team/sub/2026-09-05"), Some(true)));
pump.abort();
}
/// (r) **A monetize without `Events` makes the console announce it is polling
/// — with monetize's own refusal in the frame.** A hub that never tried also
/// says `polling`, so the message is the load-bearing half.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_monetize_that_cannot_stream_is_announced_as_polling_not_silently_stale() {
let rig = Rig::up_with(mock::Mock::refusing_events(), None).await;
let viewer = rig.cookie(Role::Viewer);
let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(20);
loop {
assert!(tokio::time::Instant::now() < deadline, "the console never announced the fallback");
let (line, _) = sse_until(&rig.router, &viewer, deadline, |v| v["kind"] == "feed").await;
if line["message"].as_str().unwrap_or("").contains("Events.Watch refused") {
assert_eq!(line["mode"], "polling", "{line}");
assert!(line["message"].as_str().unwrap().contains("no Events service"), "monetize's own words: {line}");
assert!(line["at_unix_ms"].as_u64().is_some_and(|t| t > 0));
break;
}
tokio::time::sleep(std::time::Duration::from_millis(50)).await;
}
// Still a working console: the screens answer off the polled path, with
// the fixture's numbers.
let (status, out, _) = rig.call(Method::GET, "/api/products/gunnar/tenants", Auth::Cookie(&viewer), None).await;
assert_eq!(status, StatusCode::OK, "{out}");
assert_eq!(out["tenants"].as_array().map(Vec::len), Some(7));
assert_eq!(rig.mock.watchers.load(std::sync::atomic::Ordering::SeqCst), 0, "no stream was ever open");
}