From e06346e43af29c8405fe0db111209c97abb93d03 Mon Sep 17 00:00:00 2001 From: Bendik Aagaard Lynghaug Date: Tue, 22 Sep 2026 16:51:15 +0200 Subject: [PATCH] tests/jetstream.rs: the store, event log and projection against a real JetStream Until now nothing outside a live site exercised the NATS side: the aggregate engine was tested by in-memory replay only. This test submits an answer, moves it from a desk through the engine, checks that an undeclared move and an unknown record are refused, and reads the record back from its own event log. It runs when PORTAL_TEST_NATS_URL is set; CI starts a throwaway nats:2.14.6-alpine with JetStream for it and tears it down. Without the variable the test passes as skipped, so `cargo test` stays green offline. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_01L4jrCgLiKKHAEFuZUJjckH --- .gitea/workflows/test.yml | 17 ++++++++++ Cargo.toml | 4 +++ tests/jetstream.rs | 66 +++++++++++++++++++++++++++++++++++++++ 3 files changed, 87 insertions(+) create mode 100644 tests/jetstream.rs diff --git a/.gitea/workflows/test.yml b/.gitea/workflows/test.yml index ca389d8..96191e4 100644 --- a/.gitea/workflows/test.yml +++ b/.gitea/workflows/test.yml @@ -22,5 +22,22 @@ jobs: steps: - uses: actions/checkout@v4 + # A throwaway JetStream for tests/jetstream.rs. Best effort: if the + # runner cannot use docker the variable stays unset and that one + # test reports itself skipped rather than failing the build. + - name: Start a throwaway JetStream + run: | + docker rm -f portal-test-nats >/dev/null 2>&1 || true + if docker run -d --rm --name portal-test-nats -p 127.0.0.1:14223:4222 nats:2.14.6-alpine -js >/dev/null 2>&1; then + sleep 2 + echo "PORTAL_TEST_NATS_URL=nats://127.0.0.1:14223" >> "$GITHUB_ENV" + else + echo "::warning::no docker on this runner - the JetStream integration test will be skipped" + fi + - name: Test run: cargo test --features ssr + + - name: Stop the throwaway JetStream + if: always() + run: docker rm -f portal-test-nats >/dev/null 2>&1 || true diff --git a/Cargo.toml b/Cargo.toml index 7afded9..d35c5c4 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -106,6 +106,10 @@ ssr = [ [dev-dependencies] proptest = "1" +[[test]] +name = "jetstream" +required-features = ["ssr"] + [[bin]] name = "question_lint" path = "src/bin/question_lint.rs" diff --git a/tests/jetstream.rs b/tests/jetstream.rs new file mode 100644 index 0000000..9dc373d --- /dev/null +++ b/tests/jetstream.rs @@ -0,0 +1,66 @@ +//! The one test that meets a real JetStream: a submission becomes an +//! aggregate, a desk move is accepted, an undeclared move is refused, +//! and the record reads back from its own event log. Everything else +//! in the suite replays in memory; this is what proves the store, the +//! event log and the projection agree with each other. +//! +//! Runs only when `PORTAL_TEST_NATS_URL` points at a JetStream (CI +//! starts a throwaway one, see `.gitea/workflows/test.yml`); without +//! it the test passes as skipped, so `cargo test` stays green offline. +#![cfg(feature = "ssr")] + +use portal::aggregates::{self, parse_aggregates_yaml, TransitionError}; +use portal::answers::{store_answer, Answer}; +use portal::events::store::{ensure_stream, load_events}; +use serde_json::json; + +#[tokio::test] +async fn a_submission_is_created_moved_refused_and_read_back_through_the_real_engine() { + let Ok(url) = std::env::var("PORTAL_TEST_NATS_URL") else { + eprintln!("skipped: PORTAL_TEST_NATS_URL is not set"); + return; + }; + let js = async_nats::jetstream::new(async_nats::connect(&url).await.expect("connect")); + ensure_stream(&js).await.expect("event stream"); + + // A bucket of its own per run, so reruns and parallel runs never + // read each other's records. + let bucket = format!("it_{}", chrono::Utc::now().timestamp_millis()); + let schemas = parse_aggregates_yaml(&format!( + "aggregates:\n - bucket: {bucket}\n initial: open\n states:\n open: {{ event: received }}\n quoted: {{ event: quoted }}\n won: {{ event: won }}\n lost: {{ event: lost }}\n transitions:\n open: [quoted, lost]\n quoted: [won, lost]\n" + )) + .expect("schema"); + let schema = &schemas[&bucket]; + let now = chrono::Utc::now().timestamp_millis(); + + // The submission: a KV record in the initial state, plus the first event. + store_answer(&js, &schemas, &bucket, "rec-1".into(), "/", "Hire us", &json!({"email": "a@b.example"}), now) + .await + .expect("store"); + let store = js.get_key_value(&bucket).await.expect("bucket exists"); + let stored: Answer = serde_json::from_slice(&store.get("rec-1").await.unwrap().expect("record")).unwrap(); + assert_eq!(stored.state, "open"); + assert_eq!(stored.responses["email"], "a@b.example"); + + // A desk move the machine allows. + let moved = aggregates::transition(&js, schema, "rec-1", "quoted", json!({"by": "test"}), now + 1) + .await + .expect("open -> quoted is declared"); + assert_eq!(moved.state, "quoted"); + + // Moves the machine does not allow are refused, whoever asks. + let back = aggregates::transition(&js, schema, "rec-1", "open", json!({}), now + 2).await; + assert!(matches!(back, Err(TransitionError::DisallowedTransition { .. })), "{back:?}"); + let skip = aggregates::transition(&js, schema, "rec-1", "won", json!({}), now + 2).await; + assert!(skip.is_ok(), "quoted -> won is declared: {skip:?}"); + let nobody = aggregates::transition(&js, schema, "rec-9", "quoted", json!({}), now + 3).await; + assert!(matches!(nobody, Err(TransitionError::UnknownAggregate)), "{nobody:?}"); + + // The record reads back from its own event log, in the state the + // accepted moves left it, with a sequence that moved forward. + let events = load_events(&js, &bucket, "rec-1").await.expect("events"); + assert_eq!(events.len(), 3, "received, quoted, won"); + let replayed = aggregates::replay(schema, "rec-1", &events).expect("replays"); + assert_eq!(replayed.state, "won"); + assert!(replayed.sequence > moved.sequence); +}