From 45b6b0de0009b0fdc4dfa0f7402522d195d83d0d Mon Sep 17 00:00:00 2001 From: karitham Date: Thu, 13 Aug 2026 18:35:11 +0200 Subject: [PATCH] tools/parse-plays: fix ci flakiness by abstracting the http client --- shared/src/tangled.gleam | 1 - shared/test/atproto_test.gleam | 5 +- tools/parse-plays/src/images.rs | 62 +++-------- tools/parse-plays/src/main.rs | 5 +- tools/parse-plays/src/net.rs | 33 ++++-- tools/parse-plays/src/sources.rs | 171 +++++++++++++----------------- tools/parse-plays/src/testutil.rs | 108 +++++++++++++++++++ 7 files changed, 224 insertions(+), 161 deletions(-) create mode 100644 tools/parse-plays/src/testutil.rs diff --git a/shared/src/tangled.gleam b/shared/src/tangled.gleam index 785dc4e..93b6882 100644 --- a/shared/src/tangled.gleam +++ b/shared/src/tangled.gleam @@ -11,7 +11,6 @@ import gen/actor/profile.{type ActorProfile} import gen/repo.{type Repo, Repo} import gleam/list import gleam/option.{Some, unwrap} -import gleam/string /// Fill in a Tangled repo's `name` from the URI rkey when the /// original is missing or empty. Records without a real name usually diff --git a/shared/test/atproto_test.gleam b/shared/test/atproto_test.gleam index a258cdb..992eea3 100644 --- a/shared/test/atproto_test.gleam +++ b/shared/test/atproto_test.gleam @@ -1,6 +1,5 @@ -import atproto.{type DecodedRecord, DecodedRecord} -import gen/repo.{type Repo} -import gleam/option.{type Option, None, Some} +import atproto +import gleam/option.{Some} import gleeunit/should // --- rkey_from_uri --- diff --git a/tools/parse-plays/src/images.rs b/tools/parse-plays/src/images.rs index 127eaef..2b0c1e3 100644 --- a/tools/parse-plays/src/images.rs +++ b/tools/parse-plays/src/images.rs @@ -11,7 +11,7 @@ //! the JSON so the tile still renders. use crate::model::Href; -use crate::net::HttpClient; +use crate::net::{HttpClient, HttpFetch}; use crate::sources::RateLimits; use rayon::prelude::*; use std::collections::{HashMap, HashSet}; @@ -73,7 +73,7 @@ fn scan_cache(dir: &Path) -> (HashMap, Vec) /// bodies are dropped so a truncated copy never becomes the cached /// artifact; writes go through a temp file + rename so a crash /// mid-write can't leave a partial file behind. -fn fetch_image(client: &HttpClient, dir: &Path, url: &Href) -> Option { +fn fetch_image(client: &dyn HttpFetch, dir: &Path, url: &Href) -> Option { let hash = short_hash(url.as_ref()); let (ct, buf) = client.get_bytes(url.as_ref(), MAX_IMAGE_BYTES as usize)?; if buf.is_empty() || buf.len() as u64 > MAX_IMAGE_BYTES { @@ -91,7 +91,7 @@ fn fetch_image(client: &HttpClient, dir: &Path, url: &Href) -> Option { /// per-host rate limiters; a failed download is simply absent from the /// map so the caller can keep the remote URL. pub fn download_many( - client: &HttpClient, + client: &dyn HttpFetch, limits: &RateLimits, urls: &[Href], dir: &Path, @@ -160,10 +160,6 @@ pub fn apply_rewrites(value: &mut serde_json::Value, rewrites: &HashMap Option { + fn get_json(&self, url: &str, params: &[(&str, &str)]) -> Option; + + /// GET a body, capped at `max` bytes, returning (content-type, + /// body). Callers check the length — an oversized response must + /// not become a cached artifact. + fn get_bytes(&self, url: &str, max: usize) -> Option<(String, Vec)>; + + /// Probe a URL for a 200 (e.g. Cover Art Archive's front-500 + /// endpoint). A non-2xx response (404 — no art) is a permanent + /// miss, not worth retrying. + fn check(&self, url: &str) -> bool; +} + +impl HttpFetch for HttpClient { + fn get_json(&self, url: &str, params: &[(&str, &str)]) -> Option { retry(&self.policy, || { let mut req = self.agent.get(url); for (key, value) in params { @@ -175,10 +196,7 @@ impl HttpClient { }) } - /// GET a body, capped at `max` bytes, returning (content-type, - /// body). Callers check the length — an oversized response must - /// not become a cached artifact. - pub fn get_bytes(&self, url: &str, max: usize) -> Option<(String, Vec)> { + fn get_bytes(&self, url: &str, max: usize) -> Option<(String, Vec)> { retry(&self.policy, || match self.agent.get(url).call() { Ok(resp) if resp.status() == 200 => { let ct = resp @@ -206,10 +224,7 @@ impl HttpClient { }) } - /// Probe a URL for a 200 (e.g. Cover Art Archive's front-500 - /// endpoint). A non-2xx response (404 — no art) is a permanent - /// miss, not worth retrying. - pub fn check(&self, url: &str) -> bool { + fn check(&self, url: &str) -> bool { retry(&self.policy, || match self.agent.get(url).call() { Ok(resp) if resp.status() == 200 => Attempt::Done(true), Ok(_) => Attempt::Stop, diff --git a/tools/parse-plays/src/sources.rs b/tools/parse-plays/src/sources.rs index 0354fff..dd485f3 100644 --- a/tools/parse-plays/src/sources.rs +++ b/tools/parse-plays/src/sources.rs @@ -6,7 +6,7 @@ //! endpoint round-trips are tested against a localhost stub server. use crate::model::{CommonsFilename, Href, MusicBrainzId, WikidataId}; -use crate::net::{HttpClient, Limiter}; +use crate::net::{HttpClient, HttpFetch, Limiter}; use crate::resolve::{Query, ResolvePlan}; use rayon::prelude::*; use serde_json::Value; @@ -79,8 +79,11 @@ impl RateLimits { } /// The impure gather boundary. Owns no cache — returns raw outcomes. +/// The HTTP client is behind the `HttpFetch` seam so tests substitute +/// a canned fake (see `testutil`); the network never leaks into the +/// decision logic. pub struct MusicSources { - pub(crate) client: HttpClient, + pub(crate) client: Box, pub(crate) limits: RateLimits, bases: EndpointBases, } @@ -88,20 +91,20 @@ pub struct MusicSources { impl MusicSources { pub fn new() -> Self { Self { - client: HttpClient::new(), + client: Box::new(HttpClient::new()), limits: RateLimits::production(), bases: EndpointBases::production(), } } - /// Tests only: all endpoints against one localhost stub, no rate + /// Tests only: all endpoints against one canned fake, no rate /// limiting. #[cfg(test)] - pub fn for_tests(port: u16) -> Self { + pub fn for_tests(client: Box) -> Self { Self { - client: HttpClient::new(), + client, limits: RateLimits::unthrottled(), - bases: EndpointBases::localhost(port), + bases: EndpointBases::localhost(0), } } @@ -350,11 +353,8 @@ fn percent_encode(s: &str) -> String { mod tests { use super::*; use crate::model::{AlbumKey, AlbumRef, ArtistKey, TrackKey, TrackRef}; + use crate::testutil::{Response, StubClient}; use serde_json::json; - use std::io::{Read, Write}; - use std::net::TcpListener; - use std::sync::Arc; - use std::sync::atomic::{AtomicUsize, Ordering}; // -------------------------------------------- pure parsers @@ -411,72 +411,21 @@ mod tests { assert!(p18_filename_from_json(&json!({ "claims": { "P18": [] } })).is_none()); } - // -------------------------------------------- stub server - - /// Serve canned responses, recording request lines. One request - /// per test keeps ordering trivially correct; a sequence reuses - /// the last response when exhausted. The thread owns a cloned - /// listener; the struct keeps the original so the port stays bound - /// for the test's lifetime. - struct Stub { - _listener: TcpListener, - requests: Arc, - } - - impl Stub { - fn serve(response: Vec) -> (Self, u16) { - Self::serve_sequence(vec![response]) - } - - fn serve_sequence(responses: Vec>) -> (Self, u16) { - let listener = TcpListener::bind("127.0.0.1:0").unwrap(); - let port = listener.local_addr().unwrap().port(); - let thread_listener = listener.try_clone().unwrap(); - let requests = Arc::new(AtomicUsize::new(0)); - let reqs = requests.clone(); - std::thread::spawn(move || { - for i in 0..16 { - let Ok((mut stream, _)) = thread_listener.accept() else { - return; - }; - reqs.fetch_add(1, Ordering::SeqCst); - let mut buf = [0u8; 4096]; - let _ = stream.read(&mut buf); - let response = &responses[i.min(responses.len() - 1)]; - let _ = stream.write_all(response); - } - }); - ( - Stub { - _listener: listener, - requests, - }, - port, - ) - } - - fn count(&self) -> usize { - self.requests.load(Ordering::SeqCst) - } - } - - fn http_response(status: &str, content_type: &str, body: &'static [u8]) -> Vec { - let mut resp = format!( - "HTTP/1.1 {status}\r\nContent-Type: {content_type}\r\nContent-Length: {}\r\nConnection: close\r\n\r\n", - body.len() - ) - .into_bytes(); - resp.extend_from_slice(body); - resp - } - - const OK_JSON: &[u8] = - br#"{"releases":[{"id":"aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee","title":"Flute"}]}"#; + // -------------------------------------------- endpoint round-trips #[test] fn roundtrip_release_search_and_cover_art() { - let (stub, port) = Stub::serve(http_response("200 OK", "application/json", OK_JSON)); - let sources = MusicSources::for_tests(port); + let stub = StubClient::new() + .route( + "ws/2/release", + Response::Json(json!({"releases": [{ + "id": "aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee", + "title": "Flute" + }]})), + ) + .route("front-500", Response::Status(200)); + let req_log = stub.request_log(); + let sources = MusicSources::for_tests(Box::new(stub)); let plan = ResolvePlan { resolved: vec![], queries: vec![Query::AlbumCover { @@ -491,18 +440,23 @@ mod tests { }], }; let results = sources.run_queries(&plan); - assert_eq!(stub.count(), 2); // release search + cover art probe let url = results[0].1.as_ref().unwrap().as_ref(); assert!( url.contains("/release/aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee/front-500"), "{url}" ); + // release search, then cover-art probe + let urls = req_log.lock().unwrap(); + assert_eq!(urls.len(), 2); + assert!(urls[0].contains("ws/2/release"), "{}", urls[0]); + assert!(urls[1].contains("front-500"), "{}", urls[1]); } #[test] fn roundtrip_cover_art_404_is_permanent_miss() { - let (stub, port) = Stub::serve(http_response("404 Not Found", "text/plain", b"nope")); - let sources = MusicSources::for_tests(port); + let stub = StubClient::new().route("front-500", Response::Status(404)); + let req_log = stub.request_log(); + let sources = MusicSources::for_tests(Box::new(stub)); let plan = ResolvePlan { resolved: vec![], queries: vec![Query::AlbumCover { @@ -520,22 +474,34 @@ mod tests { }; let results = sources.run_queries(&plan); // Provided MBID → single cover-art probe, no search, no retries. - assert_eq!(stub.count(), 1); assert!(results[0].1.is_none()); + assert_eq!(req_log.lock().unwrap().len(), 1); } #[test] fn roundtrip_artist_image_chain() { - let artist_json = br#"{"artists":[{"id":"bbbbbbbb-cccc-dddd-eeee-ffffffffffff"}]}"#; - let rels_json = br#"{"relations":[{"type":"wikidata","url":{"resource":"https://www.wikidata.org/wiki/Q130798"}}]}"#; - let claims_json = - br#"{"claims":{"P18":[{"mainsnak":{"datavalue":{"value":"Mac Miller 2017.jpg"}}}]}}"#; - let (stub, port) = Stub::serve_sequence(vec![ - http_response("200 OK", "application/json", artist_json), - http_response("200 OK", "application/json", rels_json), - http_response("200 OK", "application/json", claims_json), - ]); - let sources = MusicSources::for_tests(port); + let stub = StubClient::new() + .route( + "ws/2/artist?", + Response::Json(json!({"artists": [{ + "id": "bbbbbbbb-cccc-dddd-eeee-ffffffffffff" + }]})), + ) + .route( + "inc=url-rels", + Response::Json(json!({"relations": [{ + "type": "wikidata", + "url": { "resource": "https://www.wikidata.org/wiki/Q130798" } + }]})), + ) + .route( + "w/api.php", + Response::Json(json!({"claims": {"P18": [{ + "mainsnak": { "datavalue": { "value": "Mac Miller 2017.jpg" } } + }]}})), + ); + let req_log = stub.request_log(); + let sources = MusicSources::for_tests(Box::new(stub)); let plan = ResolvePlan { resolved: vec![], queries: vec![Query::ArtistImage { @@ -544,23 +510,32 @@ mod tests { }], }; let results = sources.run_queries(&plan); - assert_eq!(stub.count(), 3); let url = results[0].1.as_ref().unwrap().as_ref(); assert!( url.contains("Special:FilePath/Mac%20Miller%202017.jpg?width=600"), "{url}" ); + // artist search, url-rels lookup, wikidata claims + assert_eq!(req_log.lock().unwrap().len(), 3); } #[test] fn roundtrip_recording_and_artist_urls() { - let recording_json = br#"{"recordings":[{"id":"cccccccc-dddd-eeee-ffff-000000000000"}]}"#; - let artist_json = br#"{"artists":[{"id":"dddddddd-eeee-ffff-0000-111111111111"}]}"#; - let (stub, port) = Stub::serve_sequence(vec![ - http_response("200 OK", "application/json", recording_json), - http_response("200 OK", "application/json", artist_json), - ]); - let sources = MusicSources::for_tests(port); + let stub = StubClient::new() + .route( + "ws/2/recording", + Response::Json(json!({"recordings": [{ + "id": "cccccccc-dddd-eeee-ffff-000000000000" + }]})), + ) + .route( + "ws/2/artist", + Response::Json(json!({"artists": [{ + "id": "dddddddd-eeee-ffff-0000-111111111111" + }]})), + ); + let req_log = stub.request_log(); + let sources = MusicSources::for_tests(Box::new(stub)); let plan = ResolvePlan { resolved: vec![], queries: vec![ @@ -579,7 +554,6 @@ mod tests { ], }; let results = sources.run_queries(&plan); - assert_eq!(stub.count(), 2); assert!( results[0] .1 @@ -596,5 +570,8 @@ mod tests { .as_ref() .contains("/artist/dddddddd-eeee-ffff-0000-111111111111") ); + // Two parallel queries; the fake routes by URL, so accept order + // can't cross-wire the responses. + assert_eq!(req_log.lock().unwrap().len(), 2); } } diff --git a/tools/parse-plays/src/testutil.rs b/tools/parse-plays/src/testutil.rs new file mode 100644 index 0000000..b824cf4 --- /dev/null +++ b/tools/parse-plays/src/testutil.rs @@ -0,0 +1,108 @@ +//! Canned HTTP boundary for tests. No sockets, no threads: responses +//! are keyed by URL substring, and every request is recorded for +//! assertions. `ureq` is assumed correct — what we test is our logic +//! around it, not the wire. + +use crate::net::HttpFetch; +use serde_json::Value; +use std::sync::{Arc, Mutex}; + +/// What a matched route serves. +pub enum Response { + /// 200 with a JSON body (for `get_json`). + Json(Value), + /// 200 with raw bytes and a content type (for `get_bytes`). + Bytes(&'static [u8], &'static str), + /// A non-200 status: a permanent miss for `check`/`get_json`. + Status(u16), +} + +/// A fake `HttpFetch`: the first route whose needle matches the URL +/// wins; no match is a transport failure (`None`/`false`). Routing on +/// the URL — not arrival order — keeps parallel-query tests +/// deterministic. +pub struct StubClient { + routes: Vec<(String, Response)>, + requests: Arc>>, +} + +impl StubClient { + pub fn new() -> Self { + Self { + routes: Vec::new(), + requests: Arc::new(Mutex::new(Vec::new())), + } + } + + pub fn route(mut self, needle: &str, response: Response) -> Self { + self.routes.push((needle.to_string(), response)); + self + } + + /// Requests made so far, in order. + pub fn requests(&self) -> Vec { + self.requests.lock().unwrap().clone() + } + + /// A handle on the request log, for tests that hand the client to + /// `MusicSources::for_tests` and inspect it afterwards. + pub fn request_log(&self) -> Arc>> { + self.requests.clone() + } + + fn record(&self, url: &str) { + self.requests.lock().unwrap().push(url.to_string()); + } + + /// Match/record the URL the real client would send: `url` plus the + /// query params (ureq appends them itself). + fn full_url(url: &str, params: &[(&str, &str)]) -> String { + if params.is_empty() { + url.to_string() + } else { + let query: Vec = params + .iter() + .map(|(key, value)| format!("{key}={value}")) + .collect(); + format!("{url}?{}", query.join("&")) + } + } + + fn route_for(&self, url: &str) -> Option<&Response> { + self.routes + .iter() + .find(|(needle, _)| url.contains(needle)) + .map(|(_, response)| response) + } +} + +impl HttpFetch for StubClient { + fn get_json(&self, url: &str, params: &[(&str, &str)]) -> Option { + let url = Self::full_url(url, params); + self.record(&url); + match self.route_for(&url) { + Some(Response::Json(value)) => Some(value.clone()), + _ => None, + } + } + + fn get_bytes(&self, url: &str, _max: usize) -> Option<(String, Vec)> { + self.record(url); + match self.route_for(url) { + Some(Response::Bytes(body, content_type)) => { + Some(((*content_type).to_string(), body.to_vec())) + } + _ => None, + } + } + + fn check(&self, url: &str) -> bool { + self.record(url); + matches!( + self.route_for(url), + Some(Response::Status(200)) + | Some(Response::Json(_)) + | Some(Response::Bytes(_, _)) + ) + } +} -- 2.51.2