Something went wrong. Try again.
Live STIB-MIVB departures on the Pebble Time 2. It covers the bus, tram, and metro network of Brussels
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850use crate::{ provider::{FeedFailure, FeedStage, RawFeed}, quota::Quota, storage::{Store, TrainDeparturesBatch, TrainStationsBatch},};use anyhow::{Context, Result};use chrono::Utc;use serde::Deserialize;use std::{collections::HashSet, future::Future, sync::Arc};
#[derive(Clone, Debug, PartialEq)]pub struct TrainStation { pub id: String, pub name: String, pub lat: f64, pub lon: f64,}
#[derive(Clone, Debug, PartialEq)]pub struct TrainDeparture { pub label: String, pub destination: String, pub in_seconds: i64, pub delayed: bool, pub platform: Option<String>, pub delay_seconds: Option<i64>,}
#[derive(Deserialize)]struct StationsDoc { station: Vec<RawStation>,}
#[derive(Deserialize)]struct RawStation { id: String, name: String, standardname: String, #[serde(rename = "locationX")] location_x: String, #[serde(rename = "locationY")] location_y: String,}
pub fn parse_stations(body: &str) -> Result<Vec<TrainStation>> { let doc: StationsDoc = serde_json::from_str(body).context("stations envelope")?; Ok(doc .station .into_iter() .filter_map(|s| { let lon = s.location_x.parse::<f64>().ok()?; let lat = s.location_y.parse::<f64>().ok()?; let name = if s.standardname.trim().is_empty() { s.name } else { s.standardname }; Some(TrainStation { id: s.id, name, lat, lon }) }) .collect())}
#[derive(Deserialize)]struct LiveboardDoc { departures: DeparturesDoc,}
#[derive(Deserialize)]struct DeparturesDoc { departure: Vec<RawDeparture>,}
#[derive(Deserialize)]struct RawDeparture { station: Option<String>, time: Option<String>, delay: Option<String>, platform: Option<String>, vehicleinfo: Option<RawVehicleInfo>,}
#[derive(Deserialize)]struct RawVehicleInfo { shortname: Option<String>, #[serde(rename = "type")] kind: Option<String>,}
fn label_for(info: &Option<RawVehicleInfo>) -> String { if let Some(info) = info { if let Some(kind) = &info.kind { if !kind.trim().is_empty() { return kind.clone(); } } if let Some(shortname) = &info.shortname { if let Some(first) = shortname.split_whitespace().next() { return first.to_string(); } } } "T".to_string()}
pub fn parse_liveboard(body: &str, now_unix: i64) -> Result<Vec<TrainDeparture>> { let doc: LiveboardDoc = serde_json::from_str(body).context("liveboard envelope")?; let mut departures: Vec<TrainDeparture> = doc .departures .departure .into_iter() .filter_map(|d| { let time: i64 = d.time.as_deref()?.parse().ok()?; let delay_seconds: Option<i64> = d.delay.as_deref().and_then(|s| s.parse().ok()); let delay = delay_seconds.unwrap_or(0); let in_seconds = time + delay - now_unix; if in_seconds < -30 { return None; } Some(TrainDeparture { label: label_for(&d.vehicleinfo), destination: d.station.unwrap_or_default(), in_seconds, delayed: delay > 0, platform: d .platform .as_deref() .map(str::trim) .filter(|p| !p.is_empty() && *p != "?") .map(String::from), delay_seconds, }) }) .collect(); departures.sort_by_key(|d| d.in_seconds); departures.truncate(4); Ok(departures)}
const STATIONS_URL: &str = "https://api.irail.be/stations/?format=json";const LIVEBOARD_URL: &str = "https://api.irail.be/liveboard/";
async fn fetch_raw( http: &reqwest::Client, url: &str, feed: &'static str, scope: &str,) -> std::result::Result<RawFeed, FeedFailure> { let response = http .get(url) .send() .await .with_context(|| format!("failed to fetch {url}")) .and_then(|response| { response .error_for_status() .with_context(|| format!("failed to fetch {url}")) }) .map_err(|source| FeedFailure { stage: FeedStage::Provider, source, })?; let content_type = response .headers() .get(reqwest::header::CONTENT_TYPE) .with_context(|| format!("missing content type for {url}")) .and_then(|value| { value .to_str() .with_context(|| format!("invalid content type for {url}")) }) .map_err(|source| FeedFailure { stage: FeedStage::Provider, source, })? .to_owned() .into_boxed_str(); let body = response .bytes() .await .with_context(|| format!("failed to read {url}")) .map_err(|source| FeedFailure { stage: FeedStage::Provider, source, })? .to_vec(); Ok(RawFeed { provider: "irail", feed, scope: scope.into(), captured_at: Utc::now(), source_at: None, content_type, body, })}
async fn fetch_stations_at( http: &reqwest::Client, quota: &Quota, url: &str,) -> std::result::Result<TrainStationsBatch, FeedFailure> { if !quota.try_take(Utc::now()) { return Err(FeedFailure { stage: FeedStage::Provider, source: anyhow::anyhow!("irail quota exhausted"), }); } let raw = fetch_raw(http, url, "stations", "").await?; let stations = std::str::from_utf8(&raw.body) .context("irail stations encoding") .and_then(parse_stations) .and_then(|stations| { anyhow::ensure!(!stations.is_empty(), "irail stations empty"); Ok(stations) }) .map_err(|source| FeedFailure { stage: FeedStage::Parse, source, })?; Ok(TrainStationsBatch { raw, stations })}
pub async fn refresh_stations( http: &reqwest::Client, quota: &Quota,) -> std::result::Result<TrainStationsBatch, FeedFailure> { fetch_stations_at(http, quota, STATIONS_URL).await}
async fn fetch_departures_at( http: &reqwest::Client, url: &str, station_id: &str,) -> std::result::Result<TrainDeparturesBatch, FeedFailure> { let raw = fetch_raw(http, url, "liveboard", station_id).await?; let departures = std::str::from_utf8(&raw.body) .context("irail liveboard encoding") .and_then(|body| parse_liveboard(body, raw.captured_at.timestamp())) .map_err(|source| FeedFailure { stage: FeedStage::Parse, source, })?; Ok(TrainDeparturesBatch { raw, station_id: station_id.into(), departures, })}
async fn fetch_departures_admitted( http: &reqwest::Client, station_id: &str,) -> std::result::Result<TrainDeparturesBatch, FeedFailure> { let url = format!("{LIVEBOARD_URL}?id={station_id}&format=json&arrdep=departure"); fetch_departures_at(http, &url, station_id).await}
pub async fn fetch_departures( http: &reqwest::Client, quota: &Quota, station_id: &str,) -> std::result::Result<TrainDeparturesBatch, FeedFailure> { if !quota.try_take(Utc::now()) { return Err(FeedFailure { stage: FeedStage::Provider, source: anyhow::anyhow!("irail quota exhausted"), }); } fetch_departures_admitted(http, station_id).await}
async fn run_liveboard_worker_with<F, Fut>( store: Store, quota: Arc<Quota>, mut requests: tokio::sync::mpsc::Receiver<Vec<String>>, fetch: F,) -> anyhow::Result<()>where F: Fn(String) -> Fut, Fut: Future<Output = std::result::Result<TrainDeparturesBatch, FeedFailure>>,{ let mut next = Vec::new(); loop { let initial = if next.is_empty() { match requests.recv().await { Some(station_ids) => station_ids, None => return Ok(()), } } else { std::mem::take(&mut next) }; let mut seen = HashSet::new(); let mut pending = Vec::new(); for station_id in initial { if seen.insert(station_id.clone()) { pending.push(station_id); } } while let Ok(station_ids) = requests.try_recv() { for station_id in station_ids { if seen.insert(station_id.clone()) { pending.push(station_id); } } } let admitted = pending .into_iter() .filter(|_| quota.try_take(Utc::now())); let results = futures::future::join_all(admitted.map(&fetch)).await; let mut failures = Vec::new(); for result in results { let applied = match result { Ok(batch) => store .apply_train_departures(batch) .await .map_err(|source| FeedFailure { stage: FeedStage::Commit, source, }), Err(failure) => Err(failure), }; if let Err(failure) = applied { tracing::warn!("irail liveboard refresh failed: {failure}"); failures.push(failure); } } if !failures.is_empty() { let stage = if failures .iter() .any(|failure| failure.stage == FeedStage::Commit) { FeedStage::Commit } else if failures .iter() .any(|failure| failure.stage == FeedStage::Parse) { FeedStage::Parse } else { FeedStage::Provider }; let mut messages = failures .iter() .map(|failure| { format!("{}: {:#}", failure.stage.as_str(), failure.source) }) .collect::<Vec<_>>(); messages.sort_unstable(); messages.dedup(); if let Err(error) = store .record_feed_failure( "irail", "liveboard", stage, &messages.join("; "), Utc::now(), ) .await { tracing::warn!("irail liveboard failure recording failed: {error:#}"); } } let mut queued = HashSet::new(); while let Ok(station_ids) = requests.try_recv() { for station_id in station_ids { if !seen.contains(&station_id) && queued.insert(station_id.clone()) { next.push(station_id); } } } if next.is_empty() && requests.is_closed() { return Ok(()); } }}
pub async fn run_liveboard_worker( store: Store, http: reqwest::Client, quota: Arc<Quota>, requests: tokio::sync::mpsc::Receiver<Vec<String>>,) -> anyhow::Result<()> { run_liveboard_worker_with(store, quota, requests, move |station_id| { let http = http.clone(); async move { fetch_departures_admitted(&http, &station_id).await } }) .await}
#[cfg(test)]mod tests { use super::*;
const STATIONS: &str = r#"{"station":[ {"id":"BE.NMBS.008812005","name":"Brussels-North","standardname":"Brussel-Noord/Bruxelles-Nord","locationX":"4.36","locationY":"50.85"}, {"id":"BE.NMBS.008813003","name":"Brussels-Central","standardname":"","locationX":"4.356801","locationY":"50.845658"}, {"id":"BE.NMBS.999","name":"Bad","standardname":"Bad","locationX":"bogus","locationY":"50.0"} ]}"#;
#[test] fn parses_stations_and_prefers_standardname() { let stations = parse_stations(STATIONS).unwrap(); assert_eq!(stations.len(), 2); assert_eq!(stations[0].id, "BE.NMBS.008812005"); assert_eq!(stations[0].name, "Brussel-Noord/Bruxelles-Nord"); assert!((stations[0].lon - 4.36).abs() < 1e-9); assert!((stations[0].lat - 50.85).abs() < 1e-9); }
#[test] fn falls_back_to_name_when_standardname_empty() { let stations = parse_stations(STATIONS).unwrap(); assert_eq!(stations[1].name, "Brussels-Central"); }
#[test] fn skips_station_with_unparsable_coordinates() { let stations = parse_stations(STATIONS).unwrap(); assert!(!stations.iter().any(|s| s.id == "BE.NMBS.999")); }
fn liveboard(departures: &str) -> String { format!(r#"{{"departures":{{"number":"0","departure":[{departures}]}}}}"#) }
#[test] fn parses_liveboard_with_documented_shape() { let body = liveboard( r#"{"station":"Leuven","time":"1755733200","delay":"60","platform":"3","vehicleinfo":{"shortname":"IC 1234","type":"IC"}}"#, ); let departures = parse_liveboard(&body, 1755733000).unwrap(); assert_eq!(departures.len(), 1); assert_eq!(departures[0].label, "IC"); assert_eq!(departures[0].destination, "Leuven"); assert_eq!(departures[0].in_seconds, 260); assert!(departures[0].delayed); assert_eq!(departures[0].platform.as_deref(), Some("3")); assert_eq!(departures[0].delay_seconds, Some(60)); }
#[test] fn platform_and_delay_absent_from_the_feed_stay_none() { let body = liveboard(r#"{"station":"A","time":"100"}"#); let departures = parse_liveboard(&body, 0).unwrap(); assert_eq!(departures[0].platform, None); assert_eq!(departures[0].delay_seconds, None); }
#[test] fn unknown_or_blank_platform_marker_becomes_none() { let body = liveboard( r#"{"station":"A","time":"100","delay":"0","platform":"?"}, {"station":"B","time":"100","delay":"0","platform":" "}, {"station":"C","time":"100","delay":"0","platform":" ? "}, {"station":"D","time":"100","delay":"0","platform":" 3 "}"#, ); let departures = parse_liveboard(&body, 0).unwrap(); assert_eq!(departures[0].platform, None); assert_eq!(departures[1].platform, None); assert_eq!(departures[2].platform, None); assert_eq!(departures[3].platform.as_deref(), Some("3")); assert_eq!(departures[0].delay_seconds, Some(0)); }
#[test] fn label_falls_back_to_first_token_of_shortname_then_t() { let body = liveboard( r#"{"station":"A","time":"100","delay":"0","vehicleinfo":{"shortname":"P 7008"}}, {"station":"B","time":"100","delay":"0"}"#, ); let departures = parse_liveboard(&body, 0).unwrap(); assert_eq!(departures[0].label, "P"); assert_eq!(departures[1].label, "T"); }
#[test] fn drops_unparsable_time_and_past_departures() { let body = liveboard( r#"{"station":"A","time":"not-a-number","delay":"0"}, {"station":"B","time":"50","delay":"0"}"#, ); let departures = parse_liveboard(&body, 100).unwrap(); assert!(departures.is_empty()); }
#[test] fn keeps_departures_within_the_grace_window() { let body = liveboard(r#"{"station":"A","time":"71","delay":"0"}"#); let departures = parse_liveboard(&body, 100).unwrap(); assert_eq!(departures.len(), 1); assert_eq!(departures[0].in_seconds, -29); }
#[test] fn sorts_ascending_and_truncates_to_four() { let entries: Vec<String> = (0..6) .map(|i| format!(r#"{{"station":"S{i}","time":"{}","delay":"0"}}"#, 1000 - i * 10)) .collect(); let body = liveboard(&entries.join(",")); let departures = parse_liveboard(&body, 0).unwrap(); assert_eq!(departures.len(), 4); assert_eq!(departures[0].destination, "S5"); assert_eq!(departures[3].destination, "S2"); assert!(departures.windows(2).all(|w| w[0].in_seconds <= w[1].in_seconds)); }
fn departure(destination: &str) -> TrainDeparture { TrainDeparture { label: "IC".into(), destination: destination.into(), in_seconds: 300, delayed: false, platform: None, delay_seconds: None, } }
fn raw( feed: &'static str, scope: &str, captured_at: chrono::DateTime<Utc>, body: &[u8], ) -> crate::provider::RawFeed { crate::provider::RawFeed { provider: "irail", feed, scope: scope.into(), captured_at, source_at: None, content_type: "application/json".into(), body: body.to_vec(), } }
async fn worker_store() -> (tempfile::TempDir, crate::storage::Store) { let directory = tempfile::tempdir().unwrap(); let path = directory.path().join("stibble.sqlite"); let extension = std::env::var("SPATIALITE_EXTENSION").unwrap(); let store = crate::storage::Store::open(path, extension).await.unwrap(); let captured_at = Utc::now(); store .apply_train_stations(crate::storage::TrainStationsBatch { raw: raw("stations", "", captured_at, STATIONS.as_bytes()), stations: vec![ TrainStation { id: "first".into(), name: "FIRST".into(), lat: 50.84665, lon: 4.35243, }, TrainStation { id: "second".into(), name: "SECOND".into(), lat: 50.85, lon: 4.35243, }, ], }) .await .unwrap(); (directory, store) }
fn worker_batch(station_id: String) -> crate::storage::TrainDeparturesBatch { let captured_at = Utc::now(); crate::storage::TrainDeparturesBatch { raw: raw("liveboard", &station_id, captured_at, b"{}"), station_id: station_id.clone(), departures: vec![departure(&station_id)], } }
#[tokio::test] async fn liveboard_worker_deduplicates_pending_station_ids_and_commits_each() { let (_directory, store) = worker_store().await; let (tx, rx) = tokio::sync::mpsc::channel(8); tx.send(vec!["first".into(), "first".into()]) .await .unwrap(); tx.send(vec!["first".into(), "second".into()]) .await .unwrap(); drop(tx); let calls = Arc::new(tokio::sync::Mutex::new(Vec::new())); let fetched = calls.clone();
run_liveboard_worker_with( store.clone(), Arc::new(Quota::new(10, 100)), rx, move |station_id| { let fetched = fetched.clone(); async move { fetched.lock().await.push(station_id.clone()); Ok(worker_batch(station_id)) } }, ) .await .unwrap();
assert_eq!( calls.lock().await.as_slice(), ["first".to_string(), "second".to_string()], ); let board = store .nearby_trains(50.84665, 4.35243, Utc::now()) .await .unwrap(); assert_eq!(board.stops[0].departures[0].destination, "first"); assert_eq!(board.stops[1].departures[0].destination, "second"); }
#[tokio::test] async fn liveboard_worker_admits_only_the_available_quota_in_request_order() { let (_directory, store) = worker_store().await; let (tx, rx) = tokio::sync::mpsc::channel(8); tx.send(vec!["first".into(), "second".into()]) .await .unwrap(); drop(tx); let calls = Arc::new(tokio::sync::Mutex::new(Vec::new())); let fetched = calls.clone();
run_liveboard_worker_with( store, Arc::new(Quota::new(1, 100)), rx, move |station_id| { let fetched = fetched.clone(); async move { fetched.lock().await.push(station_id.clone()); Ok(worker_batch(station_id)) } }, ) .await .unwrap();
assert_eq!(calls.lock().await.as_slice(), ["first".to_string()]); }
#[tokio::test] async fn liveboard_worker_fetches_admitted_stations_concurrently() { use std::sync::atomic::{AtomicUsize, Ordering};
let (_directory, store) = worker_store().await; let (tx, rx) = tokio::sync::mpsc::channel(8); tx.send(vec!["first".into(), "second".into()]) .await .unwrap(); drop(tx); let in_flight = Arc::new(AtomicUsize::new(0)); let peak = Arc::new(AtomicUsize::new(0)); let observed_in_flight = in_flight.clone(); let observed_peak = peak.clone();
run_liveboard_worker_with( store, Arc::new(Quota::new(10, 100)), rx, move |station_id| { let in_flight = observed_in_flight.clone(); let peak = observed_peak.clone(); async move { let current = in_flight.fetch_add(1, Ordering::SeqCst) + 1; peak.fetch_max(current, Ordering::SeqCst); tokio::time::sleep(std::time::Duration::from_millis(50)).await; in_flight.fetch_sub(1, Ordering::SeqCst); Ok(worker_batch(station_id)) } }, ) .await .unwrap();
assert_eq!(peak.load(Ordering::SeqCst), 2); }
async fn assert_mixed_liveboard_batch_retains_failure(station_ids: [&str; 2]) { let (directory, store) = worker_store().await; let (tx, rx) = tokio::sync::mpsc::channel(8); tx.send( station_ids .into_iter() .map(String::from) .collect(), ) .await .unwrap(); drop(tx);
run_liveboard_worker_with( store.clone(), Arc::new(Quota::new(10, 100)), rx, |station_id| async move { if station_id == "first" { Err(crate::provider::FeedFailure { stage: crate::provider::FeedStage::Provider, source: anyhow::anyhow!("failed first"), }) } else { Ok(worker_batch(station_id)) } }, ) .await .unwrap();
let connection = deadpool_sqlite::rusqlite::Connection::open( directory.path().join("stibble.sqlite"), ) .unwrap(); let failure_count = connection .query_row( "SELECT count(*) FROM feed_failure JOIN provider_feed ON provider_feed.id = feed_failure.feed_id WHERE provider_feed.provider = 'irail' AND provider_feed.feed = 'liveboard'", [], |row| row.get::<_, i64>(0), ) .unwrap(); assert_eq!(failure_count, 1); let stage = connection .query_row( "SELECT stage FROM feed_failure JOIN provider_feed ON provider_feed.id = feed_failure.feed_id WHERE provider_feed.provider = 'irail' AND provider_feed.feed = 'liveboard'", [], |row| row.get::<_, String>(0), ) .unwrap(); assert_eq!(stage, "provider"); let board = store .nearby_trains(50.84665, 4.35243, Utc::now()) .await .unwrap(); assert!(board.stops[0].departures.is_empty()); assert_eq!(board.stops[1].departures[0].destination, "second"); }
#[tokio::test] async fn liveboard_worker_retains_failure_before_success() { assert_mixed_liveboard_batch_retains_failure(["first", "second"]).await; }
#[tokio::test] async fn liveboard_worker_retains_failure_after_success() { assert_mixed_liveboard_batch_retains_failure(["second", "first"]).await; }
async fn irail_server() -> (String, tokio::task::JoinHandle<()>) { let listener = tokio::net::TcpListener::bind("127.0.0.1:0") .await .unwrap(); let address = listener.local_addr().unwrap(); let departure_at = Utc::now().timestamp() + 300; let liveboard = liveboard(&format!( r#"{{"station":"LEUVEN","time":"{departure_at}","delay":"0","platform":"3","vehicleinfo":{{"shortname":"IC 1","type":"IC"}}}}"#, )); let app = axum::Router::new() .route( "/stations", axum::routing::get(|| async { ( [(axum::http::header::CONTENT_TYPE, "application/json")], STATIONS, ) }), ) .route( "/liveboard", axum::routing::get(move || { let liveboard = liveboard.clone(); async move { ( [(axum::http::header::CONTENT_TYPE, "application/json")], liveboard, ) } }), ); let server = tokio::spawn(async move { axum::serve(listener, app).await.unwrap(); }); (format!("http://{address}"), server) }
#[tokio::test] async fn fetches_station_and_liveboard_raw_parsed_batches_with_exact_scopes() { let (base, server) = irail_server().await; let http = reqwest::Client::new(); let quota = Quota::new(10, 100); let stations = fetch_stations_at(&http, "a, &format!("{base}/stations")) .await .unwrap(); assert_eq!(stations.raw.provider, "irail"); assert_eq!(stations.raw.feed, "stations"); assert!(stations.raw.scope.is_empty()); assert_eq!(stations.raw.body, STATIONS.as_bytes()); assert_eq!(stations.stations.len(), 2);
let departures = fetch_departures_at(&http, &format!("{base}/liveboard"), "first") .await .unwrap(); server.abort(); assert_eq!(departures.raw.provider, "irail"); assert_eq!(departures.raw.feed, "liveboard"); assert_eq!(departures.raw.scope.as_ref(), "first"); assert_eq!(departures.station_id, "first"); assert_eq!(departures.departures[0].destination, "LEUVEN"); }
#[tokio::test] async fn station_fetch_maps_http_and_parser_failures_to_their_stages() { let listener = tokio::net::TcpListener::bind("127.0.0.1:0") .await .unwrap(); let address = listener.local_addr().unwrap(); let app = axum::Router::new().route( "/invalid", axum::routing::get(|| async { ( [(axum::http::header::CONTENT_TYPE, "application/json")], "invalid", ) }), ); let server = tokio::spawn(async move { axum::serve(listener, app).await.unwrap(); }); let http = reqwest::Client::new(); let quota = Quota::new(10, 100); let provider = match fetch_stations_at(&http, "a, &format!("http://{address}/missing")).await { Ok(_) => panic!("expected provider failure"), Err(failure) => failure, }; let parsed = match fetch_stations_at(&http, "a, &format!("http://{address}/invalid")).await { Ok(_) => panic!("expected parse failure"), Err(failure) => failure, }; server.abort(); assert_eq!(provider.stage, crate::provider::FeedStage::Provider); assert_eq!(parsed.stage, crate::provider::FeedStage::Parse); }}