paperboy 0.5.5

//! Filesystem/data **producers** — the sources a `FOR … IN <producer>` loop
//! iterates. Everything that yields multiple items is a producer:
//! - `[ … ]`                literal (handled inline in [`super::run`]),
//! - `FILES "dir" [MATCH …]`  file paths (glob; `**` recurses),
//! - `FOLDERS "dir" [MATCH "glob"] [WITH r="glob"[?], …]`  subfolders (glob
//!   filters folder names; `**` recurses), one file per role,
//! - `TUPLES FROM "file"`   one tuple per CSV/TSV/JSON row,
//! - `ZIP(a, b, …)`         positional N-tuples (equal length required).
//!
//! Each yields an ordered list of [`ProducerItem`]s. An item carries its
//! *positional* `values` (matched against the loop's destructuring pattern and
//! forming the row key) and any *named* fields (FOLDERS roles, CSV headers) that
//! bind directly by name — so metadata columns are never lost to arity.
//!
//! These helpers are pure and front-end agnostic; [`super::run::Exec`]
//! orchestrates them (resolving `LIST` names, substituting `{{var}}`s in paths,
//! applying the `# root:` base directory).

use super::flow::RoleBinding;
use std::path::{Path, PathBuf};

/// One item produced by a loop source: its positional `values` (for pattern
/// destructuring + row key) and any `named` fields that bind by name.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct ProducerItem {
    pub values: Vec<String>,
    pub named: Vec<(String, String)>,
}

impl ProducerItem {
    /// A single positional value (the common arity-1 case: `FILES`, `FOLDERS`).
    pub fn scalar(v: impl Into<String>) -> Self {
        ProducerItem {
            values: vec![v.into()],
            named: Vec::new(),
        }
    }
}

/// Resolve a possibly-relative path against `root` (the report's directory or
/// the `# root:` override). Absolute paths are returned unchanged.
pub fn resolve_path(root: Option<&Path>, p: &str) -> PathBuf {
    let path = Path::new(p);
    if path.is_absolute() {
        path.to_path_buf()
    } else if let Some(root) = root {
        root.join(path)
    } else {
        path.to_path_buf()
    }
}

/// Match a glob (`*` = any run, `?` = one char, `{a,b,…}` = any one of the
/// listed alternatives, otherwise literal) against a single filename (no path
/// separators). Case-sensitive, matching the shell.
///
/// Brace alternation earns its keep on the one thing file corpora always do:
/// spell the same role several ways. A liveness clip is a `.webm`, a `.mov` or
/// an `.mp4`; the face crop beside it is named `crop_face` or `facefromdoc`.
/// Without `{…}` each spelling needs a loop of its own, and the script stops
/// describing the corpus and starts enumerating it.
///
/// Alternatives may nest and may themselves contain `*`/`?`; an unmatched `{`
/// is treated as a literal brace, so a filename that really does contain one
/// still matches itself.
pub fn glob_match(pattern: &str, name: &str) -> bool {
    fn m(p: &[u8], n: &[u8]) -> bool {
        match p.first() {
            None => n.is_empty(),
            Some(b'*') => {
                // Collapse consecutive `*` and try every split.
                m(&p[1..], n) || (!n.is_empty() && m(p, &n[1..]))
            }
            Some(b'?') => !n.is_empty() && m(&p[1..], &n[1..]),
            Some(b'{') => match brace_alternatives(p) {
                // Each alternative is spliced in front of the pattern's tail
                // and tried in turn: `a{b,c}d` is `abd` or `acd`.
                Some((alts, rest)) => alts.iter().any(|alt| {
                    let mut expanded = alt.to_vec();
                    expanded.extend_from_slice(rest);
                    m(&expanded, n)
                }),
                // No closing brace: an ordinary character after all.
                None => !n.is_empty() && n[0] == b'{' && m(&p[1..], &n[1..]),
            },
            Some(&c) => !n.is_empty() && n[0] == c && m(&p[1..], &n[1..]),
        }
    }
    m(pattern.as_bytes(), name.as_bytes())
}

/// Split `p`, which starts at a `{`, into its top-level comma-separated
/// alternatives and the pattern that follows the matching `}`.
///
/// Returns `None` when the brace is never closed, which is the caller's cue to
/// treat it as a literal. Nested braces are counted rather than parsed, so an
/// inner group's commas stay with the alternative that contains them and the
/// recursion in [`glob_match`] handles the group itself.
fn brace_alternatives(p: &[u8]) -> Option<(Vec<&[u8]>, &[u8])> {
    let mut depth = 0usize;
    let mut start = 1usize;
    let mut alts = Vec::new();
    for (i, &c) in p.iter().enumerate() {
        match c {
            b'{' => depth += 1,
            b',' if depth == 1 => {
                alts.push(&p[start..i]);
                start = i + 1;
            }
            b'}' => {
                depth -= 1;
                if depth == 0 {
                    alts.push(&p[start..i]);
                    return Some((alts, &p[i + 1..]));
                }
            }
            _ => {}
        }
    }
    None
}

/// List files under `dir` for a `FILES "dir" [MATCH glob]` producer. When the
/// glob contains `**` the walk recurses into subdirectories; the filename is
/// matched against the glob's last path segment. Results are sorted for
/// deterministic, reproducible ordering.
pub fn list_files(dir: &Path, glob: Option<&str>) -> Result<Vec<PathBuf>, String> {
    if !dir.is_dir() {
        return Err(format!("directory not found: {}", dir.display()));
    }
    let recursive = glob.is_some_and(|g| g.contains("**"));
    let file_pat: Option<String> = glob.map(|g| g.rsplit('/').next().unwrap_or(g).to_string());
    let mut out = Vec::new();
    collect_files(dir, recursive, file_pat.as_deref(), &mut out)?;
    out.sort();
    Ok(out)
}

fn collect_files(
    dir: &Path,
    recursive: bool,
    file_pat: Option<&str>,
    out: &mut Vec<PathBuf>,
) -> Result<(), String> {
    let entries = std::fs::read_dir(dir).map_err(|e| format!("{}: {e}", dir.display()))?;
    for entry in entries {
        let entry = entry.map_err(|e| e.to_string())?;
        let path = entry.path();
        // `file_type()` describes the entry itself and never follows a symlink,
        // so a real directory recurses while a directory *symlink* does not —
        // that's what stops a self-referential link (`loop -> ..`) from
        // recursing forever into a stack overflow.
        let file_type = entry.file_type().map_err(|e| e.to_string())?;
        if file_type.is_dir() {
            if recursive {
                collect_files(&path, recursive, file_pat, out)?;
            }
            continue;
        }
        if file_type.is_symlink() && path.is_dir() {
            // A symlink pointing at a directory: skip it (don't descend, and
            // don't mistake it for a file) so cycles can't blow the stack.
            continue;
        }
        let name = entry.file_name();
        let name = name.to_string_lossy();
        let matches = match file_pat {
            Some(pat) => glob_match(pat, &name),
            None => true,
        };
        if matches {
            out.push(path);
        }
    }
    Ok(())
}

/// List subfolders of `dir` (sorted) for a `FOLDERS "dir" [MATCH glob]`
/// producer. The glob filters folder *names* exactly as `FILES … MATCH` filters
/// file names, and likewise recurses into subdirectories when it contains `**`
/// — so `FOLDERS "." MATCH "**"` walks a whole input tree the way a nested
/// per-case layout (`<type>/<batch>/<case>/`) requires.
pub fn list_folders(dir: &Path, glob: Option<&str>) -> Result<Vec<PathBuf>, String> {
    if !dir.is_dir() {
        return Err(format!("directory not found: {}", dir.display()));
    }
    let recursive = glob.is_some_and(|g| g.contains("**"));
    // Only the glob's last segment names the folder; a leading `**/` (or a bare
    // `**`) is the recursion marker, not part of the name pattern.
    let name_pat: Option<String> = glob
        .map(|g| g.rsplit('/').next().unwrap_or(g).to_string())
        .filter(|p| p != "**");
    let mut out = Vec::new();
    collect_folders(dir, recursive, name_pat.as_deref(), &mut out)?;
    out.sort();
    Ok(out)
}

fn collect_folders(
    dir: &Path,
    recursive: bool,
    name_pat: Option<&str>,
    out: &mut Vec<PathBuf>,
) -> Result<(), String> {
    for entry in std::fs::read_dir(dir).map_err(|e| format!("{}: {e}", dir.display()))? {
        let entry = entry.map_err(|e| e.to_string())?;
        // As in `collect_files`, `file_type()` never follows a symlink, so a
        // directory *symlink* is skipped rather than descended — that is what
        // stops a self-referential link recursing forever.
        let file_type = entry.file_type().map_err(|e| e.to_string())?;
        if !file_type.is_dir() {
            continue;
        }
        let path = entry.path();
        let name = entry.file_name();
        let name = name.to_string_lossy();
        let matches = match name_pat {
            Some(pat) => glob_match(pat, &name),
            None => true,
        };
        if matches {
            out.push(path.clone());
        }
        if recursive {
            collect_folders(&path, recursive, name_pat, out)?;
        }
    }
    Ok(())
}

/// For one `FOLDERS … WITH role="glob"[?], …` subfolder, resolve each role to
/// the file in the folder matching its glob.
///
/// A **required** role must match exactly one file; an **optional** role
/// (`role="glob"?`) may match none, binding the empty string, so a genuinely
/// optional input (a document with no back side) doesn't fail the run. Matching
/// more than one file is always an error — ambiguity is never resolved silently.
///
/// Returns `Ok(None)` when a *required* role matches nothing and `on_missing` is
/// [`Missing::Skip`], meaning "this folder is not one of the ones being looked
/// for". A recursive walk passes `Skip` (it *searches* a tree, so the
/// intermediate folders it necessarily visits are simply not results); a
/// non-recursive walk passes [`Missing::Error`] (it *enumerates* a known set, so
/// a mis-shaped member is a mistake worth failing loudly on).
pub fn folder_roles(
    folder: &Path,
    roles: &[RoleBinding],
    on_missing: Missing,
) -> Result<Option<Vec<(String, String)>>, String> {
    let mut named = Vec::new();
    for role in roles {
        let mut matches = list_files(folder, Some(&role.glob))?;
        match matches.len() {
            1 => named.push((
                role.name.clone(),
                matches.remove(0).to_string_lossy().into_owned(),
            )),
            0 if role.optional => named.push((role.name.clone(), String::new())),
            0 => match on_missing {
                Missing::Skip => return Ok(None),
                Missing::Error => {
                    return Err(format!(
                        "role '{}' matched no file in {} (glob {:?})",
                        role.name,
                        folder.display(),
                        role.glob
                    ));
                }
            },
            n => {
                return Err(format!(
                    "role '{}' matched {n} files in {} (glob {:?}); expected exactly one",
                    role.name,
                    folder.display(),
                    role.glob
                ));
            }
        }
    }
    Ok(Some(named))
}

/// What a *required* role matching no file means for the folder being examined.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Missing {
    /// Fail the run naming the folder and glob (a flat, enumerated folder set).
    Error,
    /// Skip the folder — it isn't one of the ones being searched for (a
    /// recursive walk, whose intermediate folders never carry the role files).
    Skip,
}

/// Read a `TUPLES FROM "file"` manifest into items. `.csv`/`.tsv` treat the
/// first row as headers (each item's `named` fields), `.json` accepts an array
/// of arrays (positional) or an array of objects (named + positional in key
/// order). Every item exposes both positional `values` and, where available,
/// `named` fields.
pub fn read_tuples(path: &Path) -> Result<Vec<ProducerItem>, String> {
    let text = std::fs::read_to_string(path).map_err(|e| format!("{}: {e}", path.display()))?;
    let ext = path
        .extension()
        .and_then(|e| e.to_str())
        .unwrap_or("")
        .to_ascii_lowercase();
    match ext.as_str() {
        "json" => read_tuples_json(&text),
        "tsv" => Ok(read_delimited(&text, '\t')),
        _ => Ok(read_delimited(&text, ',')),
    }
}

fn read_delimited(text: &str, delim: char) -> Vec<ProducerItem> {
    let mut lines = text.lines().filter(|l| !l.trim().is_empty());
    let headers: Vec<String> = match lines.next() {
        Some(h) => split_delimited(h, delim),
        None => return Vec::new(),
    };
    lines
        .map(|line| {
            let values = split_delimited(line, delim);
            let named = headers
                .iter()
                .cloned()
                .zip(values.iter().cloned())
                .collect();
            ProducerItem { values, named }
        })
        .collect()
}

/// Split one delimited line, honouring double-quoted fields (RFC 4180 style:
/// `""` is an escaped quote inside a quoted field).
fn split_delimited(line: &str, delim: char) -> Vec<String> {
    let mut out = Vec::new();
    let mut cur = String::new();
    let mut in_quote = false;
    let mut chars = line.chars().peekable();
    while let Some(c) = chars.next() {
        match c {
            '"' if in_quote && chars.peek() == Some(&'"') => {
                cur.push('"');
                chars.next();
            }
            '"' => in_quote = !in_quote,
            _ if c == delim && !in_quote => out.push(std::mem::take(&mut cur)),
            _ => cur.push(c),
        }
    }
    out.push(cur);
    out.into_iter().map(|s| s.trim().to_string()).collect()
}

fn read_tuples_json(text: &str) -> Result<Vec<ProducerItem>, String> {
    let value: serde_json::Value =
        serde_json::from_str(text).map_err(|e| format!("invalid JSON manifest: {e}"))?;
    let arr = value
        .as_array()
        .ok_or("JSON manifest must be an array of rows")?;
    let mut items = Vec::new();
    for row in arr {
        match row {
            serde_json::Value::Array(cells) => items.push(ProducerItem {
                values: cells.iter().map(json_cell).collect(),
                named: Vec::new(),
            }),
            serde_json::Value::Object(map) => {
                let named: Vec<(String, String)> =
                    map.iter().map(|(k, v)| (k.clone(), json_cell(v))).collect();
                let values = named.iter().map(|(_, v)| v.clone()).collect();
                items.push(ProducerItem { values, named });
            }
            other => items.push(ProducerItem::scalar(json_cell(other))),
        }
    }
    Ok(items)
}

/// Stringify a JSON cell for a manifest: strings unwrapped, everything else
/// compact JSON.
fn json_cell(v: &serde_json::Value) -> String {
    match v {
        serde_json::Value::String(s) => s.clone(),
        other => other.to_string(),
    }
}

/// Zip already-expanded producer item-lists into positional N-tuples: item `i`
/// concatenates each input's `values` and merges their `named` fields. Equal
/// length is required; a mismatch is reported and the shortest length is used.
pub fn zip_items(lists: Vec<Vec<ProducerItem>>) -> Result<Vec<ProducerItem>, String> {
    if lists.is_empty() {
        return Ok(Vec::new());
    }
    let min = lists.iter().map(Vec::len).min().unwrap_or(0);
    let max = lists.iter().map(Vec::len).max().unwrap_or(0);
    let mut items = Vec::with_capacity(min);
    for i in 0..min {
        let mut values = Vec::new();
        let mut named = Vec::new();
        for list in &lists {
            values.extend(list[i].values.iter().cloned());
            named.extend(list[i].named.iter().cloned());
        }
        items.push(ProducerItem { values, named });
    }
    if min != max {
        return Err(format!(
            "ZIP inputs have unequal lengths ({min}..{max}); zipped to {min}"
        ));
    }
    Ok(items)
}

/// Concatenate already-expanded producer item-lists end-to-end into one longer
/// stream (the runtime side of `CONCAT(a, b, …)`). Unlike [`zip_items`], the
/// inputs need not be equal length — they are appended in order — but every
/// item must share the same positional arity (so the loop's destructuring
/// pattern and row key stay well-defined). A mismatch is a hard error. Each
/// item keeps its own `named` fields; the output layer unions differing field
/// sets across inputs (blank where absent).
pub fn concat_items(lists: Vec<Vec<ProducerItem>>) -> Result<Vec<ProducerItem>, String> {
    let mut items: Vec<ProducerItem> = Vec::new();
    let mut arity: Option<usize> = None;
    for list in lists {
        for item in list {
            match arity {
                None => arity = Some(item.values.len()),
                Some(a) if a != item.values.len() => {
                    return Err(format!(
                        "CONCAT inputs have mismatched arity ({a} vs {}); all inputs \
                         must yield the same number of values per item",
                        item.values.len()
                    ));
                }
                Some(_) => {}
            }
            items.push(item);
        }
    }
    Ok(items)
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::fs;

    /// The file names of `paths`, in order — what every listing test asserts
    /// on, rather than the absolute temp paths the producers return.
    fn names(paths: &[PathBuf]) -> Vec<String> {
        paths
            .iter()
            .map(|p| p.file_name().unwrap().to_string_lossy().into_owned())
            .collect()
    }

    fn tmpdir(tag: &str) -> PathBuf {
        let d = std::env::temp_dir().join(format!(
            "paperboy_prod_{tag}_{}",
            std::time::SystemTime::now()
                .duration_since(std::time::UNIX_EPOCH)
                .unwrap()
                .as_nanos()
        ));
        fs::create_dir_all(&d).unwrap();
        d
    }

    #[test]
    fn glob_matches_star_and_question() {
        assert!(glob_match("*.jpg", "a.jpg"));
        assert!(glob_match("*.jpg", ".jpg"));
        assert!(!glob_match("*.jpg", "a.png"));
        assert!(glob_match("img_?.png", "img_3.png"));
        assert!(!glob_match("img_?.png", "img_33.png"));
        assert!(glob_match("*", "anything"));
    }

    #[test]
    fn glob_matches_brace_alternatives() {
        // The two shapes the Engine4 corpora actually use.
        assert!(glob_match("*.{webm,mov,mp4}", "clip.webm"));
        assert!(glob_match("*.{webm,mov,mp4}", "clip.mp4"));
        assert!(!glob_match("*.{webm,mov,mp4}", "clip.jpg"));
        assert!(glob_match("*{crop_face,facefromdoc}*", "01_crop_face.jpg"));
        assert!(glob_match("*{crop_face,facefromdoc}*", "facefromdoc.png"));
        assert!(!glob_match("*{crop_face,facefromdoc}*", "front.jpg"));

        // An empty alternative matches nothing extra, and alternatives may
        // hold their own wildcards or nest.
        assert!(glob_match("a{,b}c", "ac"));
        assert!(glob_match("a{,b}c", "abc"));
        assert!(glob_match("{*.jpg,*.png}", "x.png"));
        assert!(glob_match("f{a{1,2},b}z", "fa2z"));
        assert!(glob_match("f{a{1,2},b}z", "fbz"));
        assert!(!glob_match("f{a{1,2},b}z", "fa3z"));

        // An unclosed brace is a literal character, not a syntax error: a file
        // really named `weird{name` still matches itself.
        assert!(glob_match("weird{name", "weird{name"));
        assert!(!glob_match("weird{name", "weirdname"));
    }

    #[test]
    fn list_files_filters_and_sorts() {
        let d = tmpdir("files");
        for n in ["b.jpg", "a.jpg", "c.png"] {
            fs::write(d.join(n), "x").unwrap();
        }
        let files = list_files(&d, Some("*.jpg")).unwrap();
        let names = names(&files);
        assert_eq!(names, vec!["a.jpg", "b.jpg"]);
        fs::remove_dir_all(&d).ok();
    }

    #[test]
    fn list_files_recurses_on_double_star() {
        let d = tmpdir("rec");
        fs::create_dir_all(d.join("sub")).unwrap();
        fs::write(d.join("top.jpg"), "x").unwrap();
        fs::write(d.join("sub/deep.jpg"), "x").unwrap();
        let files = list_files(&d, Some("**/*.jpg")).unwrap();
        assert_eq!(files.len(), 2);
        fs::remove_dir_all(&d).ok();
    }

    #[cfg(unix)]
    #[test]
    fn list_files_does_not_follow_symlinked_directories_into_a_cycle() {
        let d = tmpdir("cycle");
        fs::write(d.join("top.jpg"), "x").unwrap();
        // A directory symlink pointing back at its parent: following it while
        // recursing would loop forever and overflow the stack.
        std::os::unix::fs::symlink(&d, d.join("loop")).unwrap();
        let files = list_files(&d, Some("**/*.jpg")).unwrap();
        // Terminates, and the real file is still found (the symlink is skipped,
        // not listed as a file).
        assert_eq!(files.len(), 1);
        assert!(files[0].ends_with("top.jpg"));
        fs::remove_dir_all(&d).ok();
    }

    fn req(name: &str, glob: &str) -> RoleBinding {
        RoleBinding::required(name, glob)
    }

    fn opt(name: &str, glob: &str) -> RoleBinding {
        RoleBinding {
            optional: true,
            ..RoleBinding::required(name, glob)
        }
    }

    #[test]
    fn folder_roles_require_exactly_one_match() {
        let d = tmpdir("roles");
        fs::write(d.join("scan_front.jpg"), "x").unwrap();
        fs::write(d.join("scan_back.jpg"), "x").unwrap();
        let named = folder_roles(
            &d,
            &[req("FRONT", "*_front.jpg"), req("BACK", "*_back.jpg")],
            Missing::Error,
        )
        .unwrap()
        .unwrap();
        assert_eq!(named[0].0, "FRONT");
        assert!(named[0].1.ends_with("scan_front.jpg"));
        // A role matching nothing is an error.
        let err = folder_roles(&d, &[req("LABEL", "*.pdf")], Missing::Error).unwrap_err();
        assert!(err.contains("LABEL"));
        fs::remove_dir_all(&d).ok();
    }

    #[test]
    fn optional_role_matching_nothing_binds_empty_rather_than_failing() {
        let d = tmpdir("optrole");
        fs::write(d.join("scan_front.jpg"), "x").unwrap();
        let named = folder_roles(
            &d,
            &[req("FRONT", "*_front.jpg"), opt("BACK", "*_back.jpg")],
            Missing::Error,
        )
        .unwrap()
        .unwrap();
        assert_eq!(named[1].0, "BACK");
        assert_eq!(named[1].1, "");
        fs::remove_dir_all(&d).ok();
    }

    #[test]
    fn optional_role_matching_several_files_is_still_an_error() {
        // Optional means "may be absent", never "pick one of these for me":
        // silently choosing between two candidates would make the report depend
        // on directory order.
        let d = tmpdir("optambig");
        fs::write(d.join("a_back.jpg"), "x").unwrap();
        fs::write(d.join("b_back.jpg"), "x").unwrap();
        let err = folder_roles(&d, &[opt("BACK", "*_back.jpg")], Missing::Error).unwrap_err();
        assert!(err.contains("BACK"), "{err}");
        assert!(err.contains("expected exactly one"), "{err}");
        fs::remove_dir_all(&d).ok();
    }

    #[test]
    fn missing_skip_reports_the_folder_as_no_match_instead_of_failing() {
        let d = tmpdir("skiprole");
        let got = folder_roles(&d, &[req("FRONT", "*_front.jpg")], Missing::Skip).unwrap();
        assert!(got.is_none());
        fs::remove_dir_all(&d).ok();
    }

    #[test]
    fn list_folders_without_a_glob_lists_immediate_children_only() {
        let d = tmpdir("flat");
        fs::create_dir_all(d.join("case_a/inner")).unwrap();
        fs::create_dir_all(d.join("case_b")).unwrap();
        let got = list_folders(&d, None).unwrap();
        let names = names(&got);
        assert_eq!(names, vec!["case_a", "case_b"]);
        fs::remove_dir_all(&d).ok();
    }

    #[test]
    fn list_folders_filters_folder_names_by_glob() {
        let d = tmpdir("fglob");
        fs::create_dir_all(d.join("case_a")).unwrap();
        fs::create_dir_all(d.join("case_b")).unwrap();
        fs::create_dir_all(d.join("scratch")).unwrap();
        let got = list_folders(&d, Some("case_*")).unwrap();
        let names = names(&got);
        assert_eq!(names, vec!["case_a", "case_b"]);
        fs::remove_dir_all(&d).ok();
    }

    #[test]
    fn list_folders_recurses_on_double_star() {
        // The nested `<type>/<batch>/<case>` layout the C# generators walk: a
        // bare `**` filters nothing, so every folder at every depth is yielded.
        let d = tmpdir("frec");
        fs::create_dir_all(d.join("batch_a/june/case_1")).unwrap();
        fs::create_dir_all(d.join("batch_b/case_2")).unwrap();
        let all = list_folders(&d, Some("**")).unwrap();
        assert_eq!(all.len(), 5, "{all:?}");
        // With a name pattern, only the leaf case folders come back — at
        // whatever depth they happen to sit. The order is by full path (so the
        // run is reproducible), not by folder name.
        let cases = list_folders(&d, Some("**/case_*")).unwrap();
        let names = names(&cases);
        assert_eq!(names, vec!["case_1", "case_2"]);
        fs::remove_dir_all(&d).ok();
    }

    #[cfg(unix)]
    #[test]
    fn list_folders_does_not_follow_symlinked_directories_into_a_cycle() {
        let d = tmpdir("fcycle");
        fs::create_dir_all(d.join("case_1")).unwrap();
        std::os::unix::fs::symlink(&d, d.join("loop")).unwrap();
        let got = list_folders(&d, Some("**")).unwrap();
        // Terminates, and the symlink is not itself listed as a folder.
        let names = names(&got);
        assert_eq!(names, vec!["case_1"]);
        fs::remove_dir_all(&d).ok();
    }

    #[test]
    fn read_tuples_csv_exposes_values_and_headers() {
        let d = tmpdir("csv");
        let f = d.join("docs.csv");
        fs::write(&f, "front,back\nf1.jpg,b1.jpg\nf2.jpg,b2.jpg\n").unwrap();
        let items = read_tuples(&f).unwrap();
        assert_eq!(items.len(), 2);
        assert_eq!(items[0].values, vec!["f1.jpg", "b1.jpg"]);
        assert_eq!(
            items[0].named,
            vec![
                ("front".into(), "f1.jpg".into()),
                ("back".into(), "b1.jpg".into())
            ]
        );
        fs::remove_dir_all(&d).ok();
    }

    #[test]
    fn read_tuples_json_array_of_objects() {
        let d = tmpdir("json");
        let f = d.join("docs.json");
        fs::write(&f, r#"[{"front":"f1","back":"b1"}]"#).unwrap();
        let items = read_tuples(&f).unwrap();
        assert_eq!(items[0].named.len(), 2);
        fs::remove_dir_all(&d).ok();
    }

    #[test]
    fn zip_requires_equal_lengths() {
        let a = vec![ProducerItem::scalar("a1"), ProducerItem::scalar("a2")];
        let b = vec![ProducerItem::scalar("b1"), ProducerItem::scalar("b2")];
        let z = zip_items(vec![a, b]).unwrap();
        assert_eq!(z[0].values, vec!["a1", "b1"]);
        let short = vec![ProducerItem::scalar("x")];
        let long = vec![ProducerItem::scalar("y1"), ProducerItem::scalar("y2")];
        assert!(zip_items(vec![short, long]).is_err());
    }

    #[test]
    fn concat_appends_items_in_order() {
        let a = vec![ProducerItem::scalar("a1"), ProducerItem::scalar("a2")];
        let b = vec![ProducerItem::scalar("b1")];
        let c = concat_items(vec![a, b]).unwrap();
        let flat: Vec<&str> = c.iter().map(|i| i.values[0].as_str()).collect();
        assert_eq!(flat, vec!["a1", "a2", "b1"]);
    }

    #[test]
    fn concat_allows_unequal_lengths_and_empty_inputs() {
        let a = vec![ProducerItem::scalar("only")];
        let empty: Vec<ProducerItem> = Vec::new();
        let c = concat_items(vec![empty.clone(), a, empty]).unwrap();
        assert_eq!(c.len(), 1);
        assert_eq!(c[0].values, vec!["only"]);
    }

    #[test]
    fn concat_rejects_mismatched_arity() {
        let ones = vec![ProducerItem::scalar("a")];
        let pairs = vec![ProducerItem {
            values: vec!["x".into(), "y".into()],
            named: Vec::new(),
        }];
        assert!(concat_items(vec![ones, pairs]).is_err());
    }
}