Tests for Signals' judging, pacing, icons and API (#398)

signals.rs's pass, judge, pace, back_off, check_key and pick_icon ran
only against a live Jev. In cargo tests ask now goes to a stand-in for
it, served from a thread of its own so every test shares it, answering
by key as TypeSafe would: a pass judges each new item once, keeps its
verdicts and bills its owner's tokens on their own key; a refused key
stops its owner's judging, not anyone else's, and is not sent again; a
429 stops the pass and the next call to Jev waits out Retry-After; a
422 stops the pass rather than repeating on every item; a key check
costs tokens; an icon is one Jev was offered, on the server's key.

The API: a Signal made, its words trimmed and deduplicated, bad names,
words, icons and descriptions refused, renamed and re-worded, its icon
cleared, its picked items listed, deleted with its verdicts; someone
else's is 404 every way. In the browser, a Signal renamed and deleted
from its own page. The coverage run runs the browser Signals tests too
(IPX_SIGNALS_TESTS=1): Jev there is the fixture server's.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

crates/ipx/src/metrics.rs

@@ -158,6 +158,12 @@ impl Stats {
         }
     }
 
+    /// What `who` has been billed on `key`: items judged, Signals fit, input and output tokens.
+    #[cfg(test)]
+    pub fn billed(&self, who: &str, key: &'static str) -> [f64; 4] {
+        self.people.lock().unwrap().get(&(who.to_owned(), key)).copied().unwrap_or_default()
+    }
+
     /// Marks the worker as on a job of `kind` until what it returns is dropped, however the job
     /// ends: finished, failed or stopped.
     pub fn job(&'static self, kind: &'static str) -> Job {

crates/ipx/src/signals.rs

@@ -405,6 +405,9 @@ pub async fn check_key(ctx: &Ctx, who: &str, key: &str) -> Result<()> {
 /// Puts `body` to Jev on `key` and gives its answer, or what TypeSafe refused it with. Carries
 /// this trace's `traceparent`, as every call out of ipx does.
 async fn ask(ctx: &Ctx, key: &str, body: &serde_json::Value) -> Result<serde_json::Value> {
+    #[cfg(test)]
+    let url = tests::jev();
+    #[cfg(not(test))]
     let url = std::env::var("IPX_TYPESAFE_URL").unwrap_or_else(|_| feed::JEV_URL.to_owned());
     let mut req = ctx.client.post(url).bearer_auth(key).json(body).timeout(feed::FEED_TIMEOUT);
     if let Some(tp) = ipx_directory::otel::traceparent() {
@@ -535,8 +538,172 @@ fn attr<'a>(tag: &'a str, name: &str) -> Option<&'a str> {
 }
 
 #[cfg(test)]
-mod tests {
+pub(crate) mod tests {
     use super::*;
+    use std::sync::{Mutex, OnceLock};
+
+    /// Calls the stand-in has had, by the key they came with.
+    static CALLS: Mutex<BTreeMap<String, usize>> = Mutex::new(BTreeMap::new());
+
+    /// A stand-in for TypeSafe's Jev, as the browser suite's fixture server is: on a thread and
+    /// runtime of its own, so it outlives any one test's. It answers by the key: `refused-` as a
+    /// key it does not know, `slow-` as over its limits, `malformed-` as a request it cannot read,
+    /// `offlist-` with an icon it was not offered; anything else judges an item to fit when its
+    /// text says "fits", and picks the icon labelled robot.
+    pub(crate) fn jev() -> String {
+        static AT: OnceLock<String> = OnceLock::new();
+        AT.get_or_init(|| {
+            let l = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
+            l.set_nonblocking(true).unwrap();
+            let at = l.local_addr().unwrap();
+            std::thread::spawn(move || {
+                let rt = tokio::runtime::Builder::new_current_thread().enable_all().build().unwrap();
+                rt.block_on(async move {
+                    let app = axum::Router::new().route("/", axum::routing::post(stand_in));
+                    axum::serve(tokio::net::TcpListener::from_std(l).unwrap(), app).await.unwrap();
+                });
+            });
+            format!("http://{at}/")
+        })
+        .clone()
+    }
+
+    async fn stand_in(
+        headers: axum::http::HeaderMap,
+        axum::Json(body): axum::Json<serde_json::Value>,
+    ) -> axum::response::Response {
+        use axum::response::IntoResponse;
+        let key =
+            headers.get("authorization").and_then(|v| v.to_str().ok()).unwrap_or("").trim_start_matches("Bearer ");
+        *CALLS.lock().unwrap().entry(key.to_owned()).or_default() += 1;
+        let refuse = |status: u16, detail: &str| {
+            (
+                axum::http::StatusCode::from_u16(status).unwrap(),
+                [("retry-after", "1")],
+                format!(r#"{{"detail":"{detail}"}}"#),
+            )
+                .into_response()
+        };
+        if key.starts_with("refused-") {
+            return refuse(401, "invalid api key");
+        } else if key.starts_with("slow-") {
+            return refuse(429, "rate limited");
+        } else if key.starts_with("malformed-") {
+            return refuse(422, "criteria should be an object");
+        }
+        let fits = body["state"].as_str().is_some_and(|s| s.contains("fits"));
+        let mut answers = serde_json::Map::new();
+        for (id, q) in body["questions"].as_object().unwrap() {
+            let a = if q["type"] == "choice" {
+                let robot = q["criteria"].as_object().unwrap().iter().find(|(_, l)| l.as_str() == Some("robot"));
+                let pick = if key.starts_with("offlist-") { "solid/nope" } else { robot.map_or("", |(k, _)| k) };
+                serde_json::json!({ "choice": pick })
+            } else {
+                serde_json::json!({ "noul": if fits { 0.9 } else { 0.1 } })
+            };
+            answers.insert(id.clone(), a);
+        }
+        axum::Json(serde_json::json!({ "answers": answers, "usage": { "input_tokens": 100, "output_tokens": 3 } }))
+            .into_response()
+    }
+
+    fn calls_with(key: &str) -> usize {
+        CALLS.lock().unwrap().get(key).copied().unwrap_or(0)
+    }
+
+    /// Someone with a key, a Signal and a feed of their own holding these items; their Signal's id.
+    async fn person(ctx: &Ctx, name: &str, key: &str, items: &[(&str, &str)]) -> i64 {
+        let id = ctx.db.create_user(name, None, false).await.unwrap();
+        ctx.db.set_typesafe_key(id, Some(key)).await.unwrap();
+        let feed = format!("feed-{name}");
+        let mut sql = format!("INSERT INTO subscriptions (user_id, feed_id) VALUES ({id}, '{feed}');");
+        for (guid, title) in items {
+            sql += &format!(
+                "INSERT INTO entries (feed_id, guid, title, first_seen) VALUES ('{feed}', '{guid}', '{title}', {});",
+                db::now()
+            );
+        }
+        ctx.db.exec_for_test(&sql).await.unwrap();
+        ctx.db.add_signal(id, "Rust", &["rust".into()], None).await.unwrap()
+    }
+
+    async fn verdicts(ctx: &Ctx, signal: i64) -> Vec<String> {
+        ctx.db
+            .strings_for_test(&format!(
+                "SELECT guid || CASE WHEN hit THEN ' fits' ELSE ' not' END FROM signal_verdicts
+                  WHERE signal_id = {signal} ORDER BY guid"
+            ))
+            .await
+    }
+
+    #[tokio::test]
+    async fn a_pass_judges_each_new_item_once_keeps_its_verdicts_and_bills_its_owner() {
+        let ctx = crate::test_util::test_ctx(crate::config::Config::default()).await;
+        let signal = person(&ctx, "sig-ann", "good-ann", &[("a", "this one fits"), ("b", "about gardening")]).await;
+        assert!(!pass(&ctx).await.unwrap(), "two items is not a full batch");
+        assert_eq!(verdicts(&ctx, signal).await, ["a fits", "b not"]);
+        assert_eq!(calls_with("good-ann"), 2, "a call an item");
+        assert!(!pass(&ctx).await.unwrap());
+        assert_eq!(calls_with("good-ann"), 2, "nothing is asked twice");
+        // Items judged, Signals fit, input and output tokens, on the person's own key (#278).
+        assert_eq!(STATS.billed("sig-ann", "own"), [2.0, 1.0, 200.0, 6.0]);
+        assert_eq!(STATS.billed("sig-ann", "server"), [0.0; 4]);
+    }
+
+    #[tokio::test]
+    async fn a_refused_key_stops_its_owners_judging_and_nobody_elses() {
+        let ctx = crate::test_util::test_ctx(crate::config::Config::default()).await;
+        let ann = person(&ctx, "sig-ann2", "good-ann2", &[("a", "fits")]).await;
+        let bob = person(&ctx, "sig-bob", "refused-bob", &[("b", "fits")]).await;
+        let refused_before = STATS.jev_calls[1].load(Relaxed);
+        assert!(pass(&ctx).await.unwrap(), "goes round again for whoever is left");
+        assert_eq!(verdicts(&ctx, ann).await, ["a fits"]);
+        assert!(verdicts(&ctx, bob).await.is_empty(), "left unjudged, not judged a miss");
+        let bob_id = ctx.db.user_by_name("sig-bob").await.unwrap().unwrap().id;
+        assert!(ctx.signals_refused.lock().unwrap().contains(&bob_id));
+        assert!(STATS.jev_calls[1].load(Relaxed) > refused_before, "counted as key_refused");
+        assert!(!pass(&ctx).await.unwrap());
+        assert_eq!(calls_with("refused-bob"), 1, "not sent again until a new key is saved");
+    }
+
+    #[tokio::test]
+    async fn over_typesafes_limits_the_pass_stops_and_the_next_call_waits_its_turn() {
+        let ctx = crate::test_util::test_ctx(crate::config::Config::default()).await;
+        let signal = person(&ctx, "sig-cy", "slow-cy", &[("c", "fits")]).await;
+        let limited_before = STATS.jev_calls[3].load(Relaxed);
+        assert!(!pass(&ctx).await.unwrap(), "stops rather than going straight round");
+        assert!(verdicts(&ctx, signal).await.is_empty());
+        assert!(STATS.jev_calls[3].load(Relaxed) > limited_before, "counted as rate_limited");
+        // Retry-After said a second: the next call to Jev, anyone's, waits for it.
+        let started = std::time::Instant::now();
+        check_key(&ctx, "sig-cy", "good-cy").await.unwrap();
+        assert!(started.elapsed() >= Duration::from_millis(900), "{:?}", started.elapsed());
+    }
+
+    #[tokio::test]
+    async fn a_request_typesafe_cannot_read_stops_the_pass_and_is_counted() {
+        let ctx = crate::test_util::test_ctx(crate::config::Config::default()).await;
+        let signal = person(&ctx, "sig-dee", "malformed-dee", &[("d", "fits"), ("e", "fits")]).await;
+        let malformed_before = STATS.jev_calls[2].load(Relaxed);
+        assert!(!pass(&ctx).await.unwrap(), "every other item's would fail the same way (#257)");
+        assert!(verdicts(&ctx, signal).await.is_empty());
+        assert!(STATS.jev_calls[2].load(Relaxed) > malformed_before);
+    }
+
+    #[tokio::test]
+    async fn a_key_is_checked_and_an_icon_is_one_jev_was_offered() {
+        let ctx = crate::test_util::test_ctx(crate::config::Config::default()).await;
+        check_key(&ctx, "sig-eve", "good-eve").await.unwrap();
+        assert!(refuses_key(&check_key(&ctx, "sig-eve", "refused-eve").await.unwrap_err()));
+        assert_eq!(STATS.billed("sig-eve", "own")[2], 100.0, "a check costs tokens too");
+
+        let offered = [("solid/robot".to_owned(), "robot".to_owned()), ("solid/leaf".to_owned(), "leaf".to_owned())];
+        let words = ["ai".to_owned()];
+        let icon = pick_icon(&ctx, "sig-eve", "good-eve", "AI", &words, None, &offered).await.unwrap();
+        assert_eq!(icon, "solid/robot");
+        assert!(pick_icon(&ctx, "sig-eve", "offlist-eve", "AI", &words, None, &offered).await.is_err(), "not offered");
+        assert_eq!(STATS.billed("sig-eve", "server")[2], 200.0, "icons are on the server's key");
+    }
 
     #[test]
     fn a_noul_of_a_half_or_more_is_a_fit() {

crates/ipx/src/web.rs

@@ -1422,6 +1422,97 @@ mod tests {
         assert_eq!(ask(&app, "GET", "/api/me", "10.0.0.5", &[("cookie", "ipx_session=forged")], None).await.0, 401);
     }
 
+    #[tokio::test]
+    async fn a_signal_is_made_changed_listed_and_removed_by_its_owner_alone() {
+        let (app, ctx) = app(proxied()).await;
+        // As the proxy signs each in; the name is all that changes.
+        let call = |who: &'static str, m: &'static str, uri: String, b: Option<serde_json::Value>| {
+            let app = app.clone();
+            async move { ask(&app, m, &uri, CADDY, &[("Cf-Access-Authenticated-User-Email", who)], b).await }
+        };
+        let (dee, sam) = ("dee@x.test", "sam@x.test");
+        let json = |b: &str| serde_json::from_str::<serde_json::Value>(b).unwrap();
+
+        let make = serde_json::json!({ "name": "Rust", "keywords": ["rust", " Rust ", "cargo"], "icon": "solid/gear" });
+        let (s, _, b) = call(dee, "POST", "/api/signals".into(), Some(make)).await;
+        assert_eq!(s, 200, "{b}");
+        let id = json(&b)["id"].as_i64().unwrap();
+        for (bad, why) in [
+            (serde_json::json!({ "name": "", "keywords": ["x"] }), "no name"),
+            (serde_json::json!({ "name": "X", "keywords": [] }), "no words"),
+            (serde_json::json!({ "name": "X", "keywords": ["x"], "icon": "solid/../x" }), "an icon that is not a name"),
+            (
+                serde_json::json!({ "name": "X", "keywords": ["x"], "description": "d".repeat(501) }),
+                "a long description",
+            ),
+        ] {
+            assert_eq!(call(dee, "POST", "/api/signals".into(), Some(bad)).await.0, 400, "{why}");
+        }
+        let (_, _, list) = call(dee, "GET", "/api/signals".into(), None).await;
+        let list = json(&list);
+        assert_eq!(list.as_array().unwrap().len(), 1);
+        assert_eq!(list[0]["keywords"], serde_json::json!(["rust", "cargo"]), "trimmed, once each");
+        assert_eq!(list[0]["icon"], "solid/gear");
+
+        let change =
+            serde_json::json!({ "name": "Rust lang", "keywords": ["rustlang"], "description": "the language" });
+        assert_eq!(call(dee, "PATCH", format!("/api/signals/{id}"), Some(change)).await.0, 204);
+        let row =
+            ctx.db.signal(ctx.db.user_by_name("dee@x.test").await.unwrap().unwrap().id, id).await.unwrap().unwrap();
+        assert_eq!(
+            (row.name.as_str(), row.keywords.as_slice(), row.description.as_deref()),
+            ("Rust lang", &["rustlang".to_owned()][..], Some("the language"))
+        );
+        assert_eq!(
+            call(dee, "PATCH", format!("/api/signals/{id}"), Some(serde_json::json!({ "icon": null }))).await.0,
+            204
+        );
+        assert!(
+            ctx.db.signal(row_owner(&ctx, "dee@x.test").await, id).await.unwrap().unwrap().icon.is_none(),
+            "cleared"
+        );
+
+        // What it picked out is listed under it.
+        let owner = row_owner(&ctx, "dee@x.test").await;
+        ctx.db
+            .exec_for_test(&format!(
+                "INSERT INTO subscriptions (user_id, feed_id) VALUES ({owner}, 'f');
+                 INSERT INTO entries (feed_id, guid, title, first_seen) VALUES ('f', 'g1', 'Rust 2.0', 1), ('f', 'g2', 'Gardening', 1);"
+            ))
+            .await
+            .unwrap();
+        ctx.db.record_verdicts("f", "g1", &[(id, true)]).await.unwrap();
+        ctx.db.record_verdicts("f", "g2", &[(id, false)]).await.unwrap();
+        let (s, _, page) = call(dee, "GET", format!("/api/signals/{id}/entries"), None).await;
+        let page = json(&page);
+        assert_eq!((s, page["total"].as_i64()), (200, Some(1)), "{page}");
+        assert_eq!(page["entries"][0]["guid"], "g1");
+
+        // Someone else's Signal is not there for them, in any way.
+        assert_eq!(call(sam, "GET", format!("/api/signals/{id}/entries"), None).await.0, 404);
+        assert_eq!(
+            call(sam, "PATCH", format!("/api/signals/{id}"), Some(serde_json::json!({ "name": "Mine" }))).await.0,
+            404
+        );
+        assert_eq!(call(sam, "DELETE", format!("/api/signals/{id}"), None).await.0, 404);
+        let (_, _, theirs) = call(sam, "GET", "/api/signals".into(), None).await;
+        assert_eq!(json(&theirs), serde_json::json!([]));
+
+        assert_eq!(call(dee, "DELETE", format!("/api/signals/{id}"), None).await.0, 204);
+        assert_eq!(call(dee, "DELETE", format!("/api/signals/{id}"), None).await.0, 404, "already gone");
+        assert!(
+            ctx.db
+                .strings_for_test(&format!("SELECT guid FROM signal_verdicts WHERE signal_id = {id}"))
+                .await
+                .is_empty(),
+            "its verdicts with it"
+        );
+    }
+
+    async fn row_owner(ctx: &Ctx, name: &str) -> i64 {
+        ctx.db.user_by_name(name).await.unwrap().unwrap().id
+    }
+
     #[tokio::test]
     async fn the_block_list_and_the_typesafe_key_are_checked_before_they_are_saved() {
         let (app, ctx) = app(proxied()).await;