diff --git a/README.md b/README.md index 25b8097..31439e0 100644 --- a/README.md +++ b/README.md @@ -34,6 +34,9 @@ The default format is NDJSON. The settled horizon and optional next cursor go to stderr. Repeat with `--cursor` and the same window and `--group-by` to continue. CSV repeats its header on every page. A failed page exits nonzero; never treat partial output from several invocations as a complete export. +`--all --output FILE` follows every cursor and splits any `--from`/`--to` range into +adjacent 31-day windows, writing one header. FILE is replaced only after every page +succeeds; any inconsistency fails. The earliest settled horizon goes to stderr. Build from source with `cargo build --locked --release`. The pinned Rust toolchain is installed by rustup. Run `cargo fmt --check`, `cargo clippy --locked --all-targets -- -D warnings`, diff --git a/src/lib.rs b/src/lib.rs index 6096711..8f51fa9 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -113,7 +113,7 @@ enum Command { Guidance(RepositoryArgs), /// Read a bounded, untrusted finding fix prompt as JSON. FixPrompt { finding_id: Uuid }, - /// Export one page of analytics as CSV or NDJSON. + /// Export one page of analytics as CSV or NDJSON, or with --all every page into a file. Export(tools::ExportArgs), /// Install the newest official release over this installer-managed binary. Update, diff --git a/src/tools.rs b/src/tools.rs index 962f633..1dc28d7 100644 --- a/src/tools.rs +++ b/src/tools.rs @@ -1,5 +1,7 @@ //! Native operations on the existing token API. +use std::collections::HashSet; use std::io::{IsTerminal as _, Write as _}; +use std::path::{Path, PathBuf}; use anyhow::{Context as _, bail}; use clap::{Args, Subcommand, ValueEnum}; @@ -72,6 +74,13 @@ pub(crate) struct ExportArgs { /// Grouping for `finding_outcomes` only. #[arg(long)] group_by: Option, + /// Follow every page of every window of at most 31 days into --output; + /// --from and --to may then span any length. + #[arg(long, requires = "output", conflicts_with = "cursor")] + all: bool, + /// File that --all replaces only after every page succeeded. + #[arg(long, requires = "all")] + output: Option, } pub(crate) fn json(value: &serde_json::Value) -> anyhow::Result { @@ -111,17 +120,66 @@ pub(crate) async fn export(api: &Api, args: &ExportArgs) -> anyhow::Result, +} + +const fn export_media(format: Format) -> (&'static str, &'static str) { + match format { Format::Csv => ("csv", "text/csv"), Format::Ndjson => ("ndjson", "application/x-ndjson"), - }; + } +} + +async fn export_page( + api: &Api, + args: &ExportArgs, + from: &str, + to: &str, + cursor: Option<&str>, +) -> anyhow::Result { + let (format, accept) = export_media(args.format); let mut query = vec![ - ("from", args.from.clone()), - ("to", args.to.clone()), + ("from", from.to_owned()), + ("to", to.to_owned()), ("format", format.to_owned()), ]; - if let Some(cursor) = &args.cursor { - query.push(("cursor", cursor.clone())); + if let Some(cursor) = cursor { + query.push(("cursor", cursor.to_owned())); } if let Some(group) = &args.group_by { query.push(("group_by", group.clone())); @@ -145,27 +203,312 @@ pub(crate) async fn export(api: &Api, args: &ExportArgs) -> anyhow::Result anyhow::Result { + let windows = export_windows(&args.from, &args.to)?; + let directory = match output.parent() { + Some(parent) if !parent.as_os_str().is_empty() => parent, + _ => Path::new("."), + }; + let temporary = tempfile::Builder::new() + .prefix(".proofline-export-") + .tempfile_in(directory) + .with_context(|| format!("could not create a file next to {}", output.display()))?; + let mut file = ExportFile::new(args.format, temporary); + let mut pages = 0; + let mut earliest: Option<(Instant, String)> = None; + for (from, to) in &windows { + let mut cursor: Option = None; + let mut cursors = HashSet::new(); + let mut window_horizon: Option = None; + loop { + pages += 1; + if pages > MAX_EXPORT_PAGES { + bail!( + "the export exceeded {MAX_EXPORT_PAGES} pages; split --from and --to into smaller ranges" + ); + } + let page = export_page(api, args, from, to, cursor.as_deref()).await?; + match &window_horizon { + None => window_horizon = Some(page.horizon.clone()), + Some(horizon) if *horizon != page.horizon => bail!( + "the settled horizon changed within the window {from} to {to}, so its pages describe different snapshots" + ), + Some(_) => {} + } + file.append(&page.body)?; + match page.next_cursor { + None => break, + Some(next) if next.is_empty() || !cursors.insert(next.clone()) => { + bail!("the server returned an empty or repeated export cursor") + } + Some(next) => cursor = Some(next), + } + } + let horizon = window_horizon.context("the export window returned no page")?; + let instant = + parse_rfc3339(&horizon).context("the export has an invalid settled horizon")?; + if earliest.as_ref().is_none_or(|(first, _)| instant < *first) { + earliest = Some((instant, horizon)); + } } - say(&format!( - "settled-until: {}", - crate::text::terminal_safe(horizon) - )); - if let Some(cursor) = reply.headers.get("x-proofline-export-next-cursor") { + file.persist(output)?; + if let Some((_, horizon)) = earliest { say(&format!( - "next-cursor: {}", - crate::text::terminal_safe(cursor.to_str()?) + "settled-until: {}", + crate::text::terminal_safe(&horizon) )); } Ok(Verdict::Done) } +/// The export as it is written: a temporary file beside the destination that +/// replaces it only once complete, so a failure leaves the destination as it was. +struct ExportFile { + format: Format, + temporary: tempfile::NamedTempFile, + csv_header: Option, +} + +impl ExportFile { + const fn new(format: Format, temporary: tempfile::NamedTempFile) -> Self { + Self { + format, + temporary, + csv_header: None, + } + } + + fn append(&mut self, page: &str) -> anyhow::Result<()> { + let records = match self.format { + Format::Ndjson => page, + Format::Csv => { + let (header, records) = page.split_once('\n').unwrap_or((page, "")); + let header = header.strip_suffix('\r').unwrap_or(header); + if header.is_empty() { + bail!("an export page has no CSV header"); + } + match &self.csv_header { + None => { + self.csv_header = Some(header.to_owned()); + page + } + Some(first) if first == header => records, + Some(_) => bail!("an export page has a different CSV header"), + } + } + }; + if records.is_empty() { + return Ok(()); + } + let file = self.temporary.as_file_mut(); + file.write_all(records.as_bytes())?; + if !records.ends_with('\n') { + // A page without a final newline would join its last record to the next page's first. + let newline = match self.format { + Format::Csv => "\r\n", + Format::Ndjson => "\n", + }; + file.write_all(newline.as_bytes())?; + } + Ok(()) + } + + fn persist(self, output: &Path) -> anyhow::Result<()> { + self.temporary.as_file().sync_all()?; + self.temporary + .persist(output) + .map_err(|error| error.error) + .with_context(|| format!("could not write {}", output.display()))?; + Ok(()) + } +} + +/// Splits `[from, to)` into adjacent windows the server accepts. Each window's +/// exclusive end is the next window's inclusive start, so no record is read +/// twice or skipped. A range that already fits is sent exactly as typed. +fn export_windows(from: &str, to: &str) -> anyhow::Result> { + let start = parse_rfc3339(from).context("--from is not an RFC3339 timestamp")?; + let end = parse_rfc3339(to).context("--to is not an RFC3339 timestamp")?; + if end <= start { + bail!("--to must be later than --from"); + } + let mut windows = Vec::new(); + let mut window_from = from.to_owned(); + let mut window_start = start; + loop { + let window_end = window_start.plus_seconds(MAX_WINDOW_SECONDS); + if window_end >= end { + windows.push((window_from, to.to_owned())); + return Ok(windows); + } + let boundary = window_end.to_rfc3339(); + windows.push((window_from, boundary.clone())); + window_from = boundary; + window_start = window_end; + } +} + +/// A UTC instant. Field order makes the derived ordering chronological. +#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)] +struct Instant { + seconds: i64, + nanoseconds: u32, +} + +impl Instant { + const fn plus_seconds(self, seconds: i64) -> Self { + Self { + seconds: self.seconds + seconds, + nanoseconds: self.nanoseconds, + } + } + + fn to_rfc3339(self) -> String { + let days = self.seconds.div_euclid(86_400); + let second_of_day = self.seconds.rem_euclid(86_400); + let (year, month, day) = civil_from_days(days); + let mut text = format!( + "{year:04}-{month:02}-{day:02}T{:02}:{:02}:{:02}", + second_of_day / 3600, + second_of_day % 3600 / 60, + second_of_day % 60 + ); + if self.nanoseconds != 0 { + let fraction = format!("{:09}", self.nanoseconds); + text.push('.'); + text.push_str(fraction.trim_end_matches('0')); + } + text.push('Z'); + text + } +} + +/// Parses `YYYY-MM-DDTHH:MM:SS[.fraction](Z|±HH:MM)`. Leap seconds and +/// fractions finer than a nanosecond are refused rather than rounded. +fn parse_rfc3339(text: &str) -> Option { + let bytes = text.as_bytes(); + let year = digits(bytes, 0, 4)?; + let month = digits(bytes, 5, 2)?; + let day = digits(bytes, 8, 2)?; + let hour = digits(bytes, 11, 2)?; + let minute = digits(bytes, 14, 2)?; + let second = digits(bytes, 17, 2)?; + if bytes.get(4) != Some(&b'-') + || bytes.get(7) != Some(&b'-') + || !matches!(bytes.get(10), Some(b'T' | b't')) + || bytes.get(13) != Some(&b':') + || bytes.get(16) != Some(&b':') + || !(1..=12).contains(&month) + || day < 1 + || day > days_in_month(year, month) + || hour > 23 + || minute > 59 + || second > 59 + { + return None; + } + let mut position = 19; + let mut nanoseconds = 0; + if bytes.get(position) == Some(&b'.') { + position += 1; + let first = position; + while bytes.get(position).is_some_and(u8::is_ascii_digit) { + position += 1; + } + let count = position - first; + if count == 0 || count > 9 { + return None; + } + let fraction = u32::try_from(digits(bytes, first, count)?).ok()?; + nanoseconds = fraction * 10_u32.pow(u32::try_from(9 - count).ok()?); + } + let offset = match bytes.get(position..)? { + [b'Z' | b'z'] => 0, + [sign @ (b'+' | b'-'), _, _, b':', _, _] => { + let hours = digits(bytes, position + 1, 2)?; + let minutes = digits(bytes, position + 4, 2)?; + if hours > 23 || minutes > 59 { + return None; + } + let magnitude = hours * 3600 + minutes * 60; + if *sign == b'-' { -magnitude } else { magnitude } + } + _ => return None, + }; + let seconds = + days_from_civil(year, month, day) * 86_400 + hour * 3600 + minute * 60 + second - offset; + Some(Instant { + seconds, + nanoseconds, + }) +} + +/// The decimal value of `count` ASCII digits starting at `start`. +fn digits(bytes: &[u8], start: usize, count: usize) -> Option { + bytes + .get(start..start + count)? + .iter() + .try_fold(0, |value, byte| { + byte.is_ascii_digit() + .then(|| value * 10 + i64::from(byte - b'0')) + }) +} + +const fn days_in_month(year: i64, month: i64) -> i64 { + match month { + 2 if year % 4 == 0 && (year % 100 != 0 || year % 400 == 0) => 29, + 2 => 28, + 4 | 6 | 9 | 11 => 30, + _ => 31, + } +} + +/// Days since 1970-01-01 of a proleptic Gregorian date (Howard Hinnant's +/// `days_from_civil`). +const fn days_from_civil(year: i64, month: i64, day: i64) -> i64 { + let year = if month <= 2 { year - 1 } else { year }; + let era = year.div_euclid(400); + let year_of_era = year - era * 400; + let day_of_year = (153 * ((month + 9) % 12) + 2) / 5 + day - 1; + let day_of_era = year_of_era * 365 + year_of_era / 4 - year_of_era / 100 + day_of_year; + era * 146_097 + day_of_era - 719_468 +} + +/// The proleptic Gregorian date of days since 1970-01-01 (Howard Hinnant's +/// `civil_from_days`). +const fn civil_from_days(days: i64) -> (i64, i64, i64) { + let days = days + 719_468; + let era = days.div_euclid(146_097); + let day_of_era = days - era * 146_097; + let year_of_era = + (day_of_era - day_of_era / 1460 + day_of_era / 36_524 - day_of_era / 146_096) / 365; + let day_of_year = day_of_era - (365 * year_of_era + year_of_era / 4 - year_of_era / 100); + let shifted_month = (5 * day_of_year + 2) / 153; + let day = day_of_year - (153 * shifted_month + 2) / 5 + 1; + let month = if shifted_month < 10 { + shifted_month + 3 + } else { + shifted_month - 9 + }; + let year = year_of_era + era * 400; + (if month <= 2 { year + 1 } else { year }, month, day) +} + #[derive(Debug, Args)] pub(crate) struct ListArgs { #[arg(long)] @@ -213,3 +556,88 @@ pub(crate) async fn session(api: &Api, command: SessionCommand) -> anyhow::Resul }; json(&value) } + +#[cfg(test)] +mod tests { + use super::*; + + fn pair(from: &str, to: &str) -> (String, String) { + (from.to_owned(), to.to_owned()) + } + + #[test] + fn windows_share_each_exclusive_end_as_the_next_start() { + let windows = export_windows("2026-08-01T00:00:00Z", "2026-10-10T00:00:00Z").unwrap(); + assert_eq!( + windows, + [ + pair("2026-08-01T00:00:00Z", "2026-09-01T00:00:00Z"), + pair("2026-09-01T00:00:00Z", "2026-10-02T00:00:00Z"), + pair("2026-10-02T00:00:00Z", "2026-10-10T00:00:00Z"), + ] + ); + // Offsets and fractions are kept at the edges and carried into UTC boundaries. + let windows = + export_windows("2024-02-01T01:30:00.25+02:00", "2024-03-10T00:00:00Z").unwrap(); + assert_eq!( + windows, + [ + pair("2024-02-01T01:30:00.25+02:00", "2024-03-02T23:30:00.25Z"), + pair("2024-03-02T23:30:00.25Z", "2024-03-10T00:00:00Z"), + ] + ); + for window in windows.windows(2) { + assert_eq!( + parse_rfc3339(&window[0].1).unwrap(), + parse_rfc3339(&window[1].0).unwrap() + ); + } + } + + #[test] + fn a_fitting_span_is_sent_as_typed() { + for (from, to) in [ + ("2026-10-01T02:00:00+02:00", "2026-10-25T00:00:00z"), + ("2026-10-01T00:00:00Z", "2026-11-01T00:00:00Z"), + ("2026-10-01t00:00:00.000Z", "2026-10-01T00:00:00.000000001Z"), + ] { + assert_eq!(export_windows(from, to).unwrap(), [pair(from, to)]); + } + } + + #[test] + fn malformed_or_empty_ranges_are_refused() { + for (from, to) in [ + ("2026-10-01T00:00:00Z", "2026-10-01T00:00:00Z"), + ("2026-10-02T00:00:00Z", "2026-10-01T00:00:00Z"), + ("2026-10-01T02:00:00+02:00", "2026-10-01T00:00:00Z"), + ("2026-10-01", "2026-10-25T00:00:00Z"), + ("2026-10-01T00:00:00", "2026-10-25T00:00:00Z"), + ("2026-02-29T00:00:00Z", "2026-10-25T00:00:00Z"), + ("2026-06-30T23:59:60Z", "2026-10-25T00:00:00Z"), + ("2026-10-01T00:00:00.1234567891Z", "2026-10-25T00:00:00Z"), + ("2026-10-01T00:00:00.Z", "2026-10-25T00:00:00Z"), + ("2026-10-01T00:00:00+2:00", "2026-10-25T00:00:00Z"), + ("2026-10-01T00:00:00Z ", "2026-10-25T00:00:00Z"), + ("2026-10-01T00:00:00Z", "２026-10-25T00:00:00Z"), + ("", ""), + ] { + assert!(export_windows(from, to).is_err(), "{from} {to}"); + } + } + + #[test] + fn civil_dates_round_trip() { + assert_eq!(days_from_civil(1970, 1, 1), 0); + assert_eq!(days_from_civil(2000, 3, 1), 11_017); + assert_eq!(civil_from_days(-1), (1969, 12, 31)); + for days in [-719_468, -1, 0, 11_016, 11_017, 20_000, 2_932_896] { + let (year, month, day) = civil_from_days(days); + assert_eq!(days_from_civil(year, month, day), days); + } + assert_eq!( + parse_rfc3339("2024-02-29T12:00:00Z").unwrap().to_rfc3339(), + "2024-02-29T12:00:00Z" + ); + } +} diff --git a/tests/http.rs b/tests/http.rs index b65ed37..5a19ffb 100644 --- a/tests/http.rs +++ b/tests/http.rs @@ -2188,3 +2188,366 @@ async fn fix_commands_preserve_context_and_never_retry_verification() { assert_eq!(calls.lock().unwrap().len(), 4); server.abort(); } + +type ExportQuery = std::collections::BTreeMap; + +/// A console whose export pages come from `page`, recording every query. +async fn export_console( + page: impl Fn(&ExportQuery) -> Response + Clone + Send + Sync + 'static, +) -> ( + String, + Arc>>, + tokio::task::JoinHandle<()>, +) { + let queries = Arc::new(Mutex::new(Vec::new())); + let seen = Arc::clone(&queries); + let app = Router::new().route( + &format!("/api/v1/organizations/{ORG}/exports/reviews"), + get(move |uri: axum::http::Uri| { + let seen = Arc::clone(&seen); + let page = page.clone(); + async move { + let query: ExportQuery = + url::form_urlencoded::parse(uri.query().unwrap_or_default().as_bytes()) + .into_owned() + .collect(); + let reply = page(&query); + seen.lock().unwrap().push(query); + reply + } + }), + ); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let origin = format!("http://{}", listener.local_addr().unwrap()); + let server = tokio::spawn(async move { axum::serve(listener, app).await.unwrap() }); + (origin, queries, server) +} + +fn export_reply(content_type: &str, horizon: &str, next: Option<&str>, body: &str) -> Response { + let mut headers = HeaderMap::new(); + headers.insert(header::CONTENT_TYPE, content_type.parse().unwrap()); + headers.insert("x-proofline-export-version", "1".parse().unwrap()); + headers.insert("x-proofline-export-settled-until", horizon.parse().unwrap()); + if let Some(next) = next { + headers.insert("x-proofline-export-next-cursor", next.parse().unwrap()); + } + (headers, body.to_owned()).into_response() +} + +fn export_args<'a>(from: &'a str, to: &'a str, format: &'a str, output: &'a str) -> Vec<&'a str> { + vec![ + "export", "reviews", "--from", from, "--to", to, "--format", format, "--all", "--output", + output, + ] +} + +fn directory_entries(directory: &std::path::Path) -> Vec { + let mut names: Vec = std::fs::read_dir(directory) + .unwrap() + .map(|entry| entry.unwrap().file_name().into_string().unwrap()) + .collect(); + names.sort(); + names +} + +#[tokio::test] +async fn export_all_follows_every_cursor_into_one_ndjson_file() { + let (origin, queries, server) = export_console(|query| { + let (next, body) = match query.get("cursor").map(String::as_str) { + None => (Some("c/1+"), "{\"n\":1}\n"), + // A page without its final newline must not join the next page's first record. + Some("c/1+") => (Some("c2"), "{\"n\":2}"), + Some(_) => (None, "{\"n\":3}\n"), + }; + export_reply("application/x-ndjson", "2026-10-25T00:00:00Z", next, body) + }) + .await; + let directory = tempfile::tempdir().unwrap(); + let output = cli_in( + &origin, + &export_args( + "2026-10-01T00:00:00Z", + "2026-10-25T00:00:00Z", + "ndjson", + "reviews.ndjson", + ), + directory.path(), + ) + .await; + assert!( + output.status.success(), + "{}", + String::from_utf8_lossy(&output.stderr) + ); + assert!(output.stdout.is_empty()); + assert_eq!( + String::from_utf8_lossy(&output.stderr), + "settled-until: 2026-10-25T00:00:00Z\n" + ); + assert_eq!( + std::fs::read_to_string(directory.path().join("reviews.ndjson")).unwrap(), + "{\"n\":1}\n{\"n\":2}\n{\"n\":3}\n" + ); + assert_eq!(directory_entries(directory.path()), ["reviews.ndjson"]); + let queries = queries.lock().unwrap(); + let cursors: Vec<_> = queries.iter().map(|query| query.get("cursor")).collect(); + assert_eq!( + cursors, + [None, Some(&"c/1+".to_owned()), Some(&"c2".to_owned())] + ); + for query in queries.iter() { + assert_eq!(query["from"], "2026-10-01T00:00:00Z"); + assert_eq!(query["to"], "2026-10-25T00:00:00Z"); + assert_eq!(query["format"], "ndjson"); + } + server.abort(); +} + +#[tokio::test] +async fn export_all_splits_a_long_range_into_adjacent_windows_with_one_csv_header() { + let (origin, queries, server) = export_console(|query| { + let from = query["from"].as_str(); + let horizon = match from { + "2026-09-01T00:00:00Z" => "2026-11-01T00:00:00Z", + "2026-10-02T00:00:00Z" => "2026-11-03T00:00:00Z", + _ => "2026-11-02T00:00:00Z", + }; + let (next, body) = match (from, query.get("cursor")) { + ("2026-08-01T00:00:00Z", None) => { + (Some("more"), format!("from,claim\r\n{from},\"a,b\"\r\n")) + } + ("2026-08-01T00:00:00Z", Some(_)) => (None, format!("from,claim\r\n{from},second")), + _ => (None, format!("from,claim\r\n{from},\"a,b\"\r\n")), + }; + export_reply("text/csv", horizon, next, &body) + }) + .await; + let directory = tempfile::tempdir().unwrap(); + let output = cli_in( + &origin, + &export_args( + "2026-08-01T00:00:00Z", + "2026-10-10T00:00:00Z", + "csv", + "reviews.csv", + ), + directory.path(), + ) + .await; + assert!( + output.status.success(), + "{}", + String::from_utf8_lossy(&output.stderr) + ); + let sent: Vec<_> = queries + .lock() + .unwrap() + .iter() + .map(|query| { + ( + query["from"].clone(), + query["to"].clone(), + query.get("cursor").cloned(), + ) + }) + .collect(); + let window = |from: &str, to: &str, cursor: Option<&str>| { + (from.to_owned(), to.to_owned(), cursor.map(str::to_owned)) + }; + assert_eq!( + sent, + [ + window("2026-08-01T00:00:00Z", "2026-09-01T00:00:00Z", None), + window("2026-08-01T00:00:00Z", "2026-09-01T00:00:00Z", Some("more")), + window("2026-09-01T00:00:00Z", "2026-10-02T00:00:00Z", None), + window("2026-10-02T00:00:00Z", "2026-10-10T00:00:00Z", None), + ] + ); + assert_eq!( + std::fs::read_to_string(directory.path().join("reviews.csv")).unwrap(), + "from,claim\r\n\ + 2026-08-01T00:00:00Z,\"a,b\"\r\n\ + 2026-08-01T00:00:00Z,second\r\n\ + 2026-09-01T00:00:00Z,\"a,b\"\r\n\ + 2026-10-02T00:00:00Z,\"a,b\"\r\n" + ); + // The earliest window horizon is the guarantee for the whole file. + assert_eq!( + String::from_utf8_lossy(&output.stderr), + "settled-until: 2026-11-01T00:00:00Z\n" + ); + assert_eq!(directory_entries(directory.path()), ["reviews.csv"]); + server.abort(); +} + +#[tokio::test] +async fn export_all_failure_leaves_no_file_and_an_existing_file_untouched() { + let (origin, _, server) = export_console(|query| { + if query.contains_key("cursor") { + return ( + StatusCode::INTERNAL_SERVER_ERROR, + Json(json!({"detail": "boom"})), + ) + .into_response(); + } + export_reply( + "text/csv", + "2026-10-25T00:00:00Z", + Some("next"), + "from,claim\r\nnew,row\r\n", + ) + }) + .await; + let directory = tempfile::tempdir().unwrap(); + let args = export_args( + "2026-10-01T00:00:00Z", + "2026-10-25T00:00:00Z", + "csv", + "reviews.csv", + ); + let output = cli_in(&origin, &args, directory.path()).await; + assert_eq!(output.status.code(), Some(1)); + assert!(directory_entries(directory.path()).is_empty()); + std::fs::write(directory.path().join("reviews.csv"), "previous\r\n").unwrap(); + let output = cli_in(&origin, &args, directory.path()).await; + assert_eq!(output.status.code(), Some(1)); + assert!(String::from_utf8_lossy(&output.stderr).contains("HTTP 500")); + assert_eq!( + std::fs::read_to_string(directory.path().join("reviews.csv")).unwrap(), + "previous\r\n" + ); + assert_eq!(directory_entries(directory.path()), ["reviews.csv"]); + server.abort(); +} + +#[tokio::test] +async fn export_all_refuses_inconsistent_pages() { + type Page = fn(Option<&str>) -> (&'static str, Option<&'static str>, &'static str); + let cases: [(Page, &str); 4] = [ + ( + |cursor| match cursor { + None => ("2026-10-25T00:00:00Z", Some("c1"), "a,b\r\n1,2\r\n"), + Some(_) => ("2026-10-25T00:00:01Z", None, "a,b\r\n3,4\r\n"), + }, + "settled horizon changed", + ), + ( + |_| ("2026-10-25T00:00:00Z", Some("c1"), "a,b\r\n1,2\r\n"), + "repeated export cursor", + ), + ( + |_| ("2026-10-25T00:00:00Z", Some(""), "a,b\r\n1,2\r\n"), + "empty or repeated export cursor", + ), + ( + |cursor| match cursor { + None => ("2026-10-25T00:00:00Z", Some("c1"), "a,b\r\n1,2\r\n"), + Some(_) => ("2026-10-25T00:00:00Z", None, "a,c\r\n3,4\r\n"), + }, + "different CSV header", + ), + ]; + for (page, refusal) in cases { + let (origin, _, server) = export_console(move |query| { + let (horizon, next, body) = page(query.get("cursor").map(String::as_str)); + export_reply("text/csv", horizon, next, body) + }) + .await; + let directory = tempfile::tempdir().unwrap(); + let output = cli_in( + &origin, + &export_args( + "2026-10-01T00:00:00Z", + "2026-10-25T00:00:00Z", + "csv", + "reviews.csv", + ), + directory.path(), + ) + .await; + assert_eq!(output.status.code(), Some(1)); + assert!( + String::from_utf8_lossy(&output.stderr).contains(refusal), + "{}", + String::from_utf8_lossy(&output.stderr) + ); + assert!(directory_entries(directory.path()).is_empty()); + server.abort(); + } +} + +#[tokio::test] +async fn export_all_stops_at_the_page_cap() { + let (origin, queries, server) = export_console(|query| { + let next = format!("{}x", query.get("cursor").map_or("", String::as_str)); + export_reply( + "application/x-ndjson", + "2026-10-25T00:00:00Z", + Some(&next), + "{}\n", + ) + }) + .await; + let directory = tempfile::tempdir().unwrap(); + let output = cli_in( + &origin, + &export_args( + "2026-10-01T00:00:00Z", + "2026-10-02T00:00:00Z", + "ndjson", + "reviews.ndjson", + ), + directory.path(), + ) + .await; + assert_eq!(output.status.code(), Some(1)); + assert!(String::from_utf8_lossy(&output.stderr).contains("exceeded 1000 pages")); + assert_eq!(queries.lock().unwrap().len(), 1000); + assert!(directory_entries(directory.path()).is_empty()); + server.abort(); +} + +#[tokio::test] +async fn export_all_needs_output_conflicts_with_cursor_and_checks_the_range_first() { + let (origin, queries, server) = export_console(|_| StatusCode::NOT_FOUND.into_response()).await; + let range = [ + "export", + "reviews", + "--from", + "2026-10-01T00:00:00Z", + "--to", + "2026-10-25T00:00:00Z", + ]; + for (extra, refusal) in [ + (&["--all"][..], "--output "), + (&["--output", "reviews.ndjson"][..], "--all"), + ( + &["--all", "--output", "reviews.ndjson", "--cursor", "c1"][..], + "cannot be used with", + ), + ] { + let args: Vec<&str> = range.iter().chain(extra).copied().collect(); + let output = cli(&origin, &args).await; + assert_eq!(output.status.code(), Some(1)); + let stderr = String::from_utf8_lossy(&output.stderr); + assert!(stderr.contains(refusal), "{stderr}"); + } + for (from, to, refusal) in [ + ( + "2026-10-25T00:00:00Z", + "2026-10-25T00:00:00Z", + "--to must be later than --from", + ), + ( + "2026-10-01", + "2026-10-25T00:00:00Z", + "--from is not an RFC3339 timestamp", + ), + ] { + let output = cli(&origin, &export_args(from, to, "csv", "reviews.csv")).await; + assert_eq!(output.status.code(), Some(1)); + assert!(String::from_utf8_lossy(&output.stderr).contains(refusal)); + } + assert!(queries.lock().unwrap().is_empty()); + server.abort(); +}