use crate::workdir::Workdir;

#[test]
fn sample_seed() {
    let wrk = Workdir::new("sample_seed");
    wrk.create(
        "in.csv",
        vec![
            svec!["R", "S"],
            svec!["1", "b"],
            svec!["2", "a"],
            svec!["3", "d"],
            svec!["4", "c"],
            svec!["5", "f"],
            svec!["6", "e"],
            svec!["7", "i"],
            svec!["8", "h"],
        ],
    );

    let mut cmd = wrk.command("sample");
    cmd.args(["--seed", "42"]).arg("5").arg("in.csv");

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let expected = vec![
        svec!["R", "S"],
        svec!["6", "e"],
        svec!["8", "h"],
        svec!["3", "d"],
        svec!["7", "i"],
        svec!["5", "f"],
    ];
    assert_eq!(got, expected);
}

#[test]
fn sample_seed_delimiter() {
    let wrk = Workdir::new("sample_seed_delimiter");
    wrk.create_with_delim(
        "in.csv",
        vec![
            svec!["R", "S"],
            svec!["1", "b"],
            svec!["2", "a"],
            svec!["3", "d"],
            svec!["4", "c"],
            svec!["5", "f"],
            svec!["6", "e"],
            svec!["7", "i"],
            svec!["8", "h"],
        ],
        b'|',
    );

    let mut cmd = wrk.command("sample");
    cmd.args(["--seed", "42"]).arg("5").arg("in.csv");

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let expected = vec![
        svec!["R|S"],
        svec!["6|e"],
        svec!["8|h"],
        svec!["3|d"],
        svec!["7|i"],
        svec!["5|f"],
    ];
    assert_eq!(got, expected);
}

#[test]
fn sample_seed_faster() {
    let wrk = Workdir::new("sample_seed_faster");
    wrk.create(
        "in.csv",
        vec![
            svec!["R", "S"],
            svec!["1", "b"],
            svec!["2", "a"],
            svec!["3", "d"],
            svec!["4", "c"],
            svec!["5", "f"],
            svec!["6", "e"],
            svec!["7", "i"],
            svec!["8", "h"],
        ],
    );

    let mut cmd = wrk.command("sample");
    cmd.args(["--rng", "faster"])
        .args(["--seed", "42"])
        .arg("5")
        .arg("in.csv");

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let expected = vec![
        svec!["R", "S"],
        svec!["8", "h"],
        svec!["2", "a"],
        svec!["7", "i"],
        svec!["4", "c"],
        svec!["5", "f"],
    ];
    assert_eq!(got, expected);
}

#[test]
fn sample_seed_secure() {
    let wrk = Workdir::new("sample_seed_secure");
    wrk.create(
        "in.csv",
        vec![
            svec!["R", "S"],
            svec!["1", "b"],
            svec!["2", "a"],
            svec!["3", "d"],
            svec!["4", "c"],
            svec!["5", "f"],
            svec!["6", "e"],
            svec!["7", "i"],
            svec!["8", "h"],
        ],
    );

    let mut cmd = wrk.command("sample");
    cmd.args(["--rng", "cryptosecure"])
        .args(["--seed", "42"])
        .arg("5")
        .arg("in.csv");

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let expected = vec![
        svec!["R", "S"],
        svec!["1", "b"],
        svec!["6", "e"],
        svec!["3", "d"],
        svec!["4", "c"],
        svec!["8", "h"],
    ];
    assert_eq!(got, expected);
}

#[test]
fn sample_seed_url() {
    let wrk = Workdir::new("sample_seed_url");

    let mut cmd = wrk.command("sample");
    cmd.args(["--seed", "42"])
        .arg("5")
        .arg("https://github.com/dathere/qsv/raw/master/resources/test/aliases.csv");

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let expected = vec![
        ["position", "title"],
        ["Q107058011", "ambassador to Mauritania"],
        [
            "Q100797227",
            "Minister of Family and Social Services of the Government of the Balearic Islands",
        ],
        [
            "Q106968387",
            "Minister of Research and Universities of the Government of Catalonia",
        ],
        ["Q106918017", "conseller d'Obres Públiques i Urbanisme"],
        [
            "Q106162142",
            "Conseiller aux Infrastructures, au Territoire et à l'Environnement de la Généralité \
             valencienne",
        ],
    ];
    assert_eq!(got, expected);
}

#[test]
fn sample_percentage_seed_no_index_percentage() {
    let wrk = Workdir::new("sample_percentage_seed_no_index_percentage");
    wrk.create(
        "in.csv",
        vec![
            svec!["R", "S"],
            svec!["1", "b"],
            svec!["2", "a"],
            svec!["3", "d"],
            svec!["4", "c"],
            svec!["5", "f"],
            svec!["6", "e"],
            svec!["7", "i"],
            svec!["8", "h"],
            svec!["8", "h"],
        ],
    );

    let mut cmd = wrk.command("sample");
    cmd.args(["--seed", "42"]).arg("0.6").arg("in.csv");

    // no error since percentage sampling no longer requires an index
    // though note the results are different even with the same seed and
    // sample size. This is because we use sample_reservoir method, not
    // sample_random_access method
    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let expected = vec![
        svec!["R", "S"],
        svec!["6", "e"],
        svec!["8", "h"],
        svec!["3", "d"],
        svec!["7", "i"],
        svec!["8", "h"],
    ];
    assert_eq!(got, expected);
}

#[test]
fn sample_percentage_seed_indexed() {
    let wrk = Workdir::new("sample_indexed");
    wrk.create_indexed(
        "in.csv",
        vec![
            svec!["R", "S"],
            svec!["1", "b"],
            svec!["2", "a"],
            svec!["3", "d"],
            svec!["4", "c"],
            svec!["5", "f"],
            svec!["6", "e"],
            svec!["7", "i"],
            svec!["8", "h"],
            svec!["8", "h"],
        ],
    );

    let mut cmd = wrk.command("sample");
    cmd.args(["--seed", "42"]).arg("0.4").arg("in.csv");

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let expected = vec![
        svec!["R", "S"],
        svec!["4", "c"],
        svec!["5", "f"],
        svec!["6", "e"],
    ];
    assert_eq!(got, expected);
}

#[test]
fn sample_percentage_seed_indexed_faster() {
    let wrk = Workdir::new("sample_indexed_faster");
    wrk.create_indexed(
        "in.csv",
        vec![
            svec!["R", "S"],
            svec!["1", "b"],
            svec!["2", "a"],
            svec!["3", "d"],
            svec!["4", "c"],
            svec!["5", "f"],
            svec!["6", "e"],
            svec!["7", "i"],
            svec!["8", "h"],
            svec!["8", "h"],
        ],
    );

    let mut cmd = wrk.command("sample");
    cmd.args(["--rng", "faster"])
        .args(["--seed", "42"])
        .arg("0.4")
        .arg("in.csv");

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let expected = vec![
        svec!["R", "S"],
        svec!["5", "f"],
        svec!["8", "h"],
        svec!["8", "h"],
    ];
    assert_eq!(got, expected);
}

#[test]
fn sample_percentage_seed_indexed_secure() {
    let wrk = Workdir::new("sample_indexed_secure");
    wrk.create_indexed(
        "in.csv",
        vec![
            svec!["R", "S"],
            svec!["1", "b"],
            svec!["2", "a"],
            svec!["3", "d"],
            svec!["4", "c"],
            svec!["5", "f"],
            svec!["6", "e"],
            svec!["7", "i"],
            svec!["8", "h"],
            svec!["8", "h"],
        ],
    );

    let mut cmd = wrk.command("sample");
    cmd.args(["--rng", "cryptosecure"])
        .args(["--seed", "42"])
        .arg("0.4")
        .arg("in.csv");

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let expected = vec![
        svec!["R", "S"],
        svec!["1", "b"],
        svec!["3", "d"],
        svec!["8", "h"],
    ];
    assert_eq!(got, expected);
}

#[test]
fn sample_indexed_random_access() {
    let wrk = Workdir::new("sample_indexed_random_access");
    wrk.create_indexed(
        "in.csv",
        vec![
            svec!["R", "S"],
            svec!["1", "b"],
            svec!["2", "a"],
            svec!["3", "d"],
            svec!["4", "c"],
            svec!["5", "f"],
            svec!["6", "e"],
            svec!["7", "i"],
            svec!["8", "h"],
            svec!["9", "i"],
            svec!["10", "j"],
            svec!["11", "k"],
            svec!["12", "l"],
            svec!["13", "m"],
            svec!["14", "n"],
            svec!["15", "o"],
            svec!["16", "p"],
            svec!["17", "q"],
            svec!["18", "r"],
            svec!["19", "s"],
            svec!["20", "t"],
            svec!["21", "u"],
            svec!["22", "v"],
            svec!["23", "w"],
            svec!["24", "x"],
            svec!["25", "y"],
            svec!["26", "z"],
        ],
    );

    let mut cmd = wrk.command("sample");
    cmd.args(["--seed", "42"]).arg("4").arg("in.csv");

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let expected = vec![
        svec!["R", "S"],
        svec!["7", "i"],
        svec!["19", "s"],
        svec!["22", "v"],
        svec!["24", "x"],
    ];
    assert_eq!(got, expected);
}

#[test]
fn sample_indexed_random_access_faster() {
    let wrk = Workdir::new("sample_indexed_random_access_faster");
    wrk.create_indexed(
        "in.csv",
        vec![
            svec!["R", "S"],
            svec!["1", "b"],
            svec!["2", "a"],
            svec!["3", "d"],
            svec!["4", "c"],
            svec!["5", "f"],
            svec!["6", "e"],
            svec!["7", "i"],
            svec!["8", "h"],
            svec!["9", "i"],
            svec!["10", "j"],
            svec!["11", "k"],
            svec!["12", "l"],
            svec!["13", "m"],
            svec!["14", "n"],
            svec!["15", "o"],
            svec!["16", "p"],
            svec!["17", "q"],
            svec!["18", "r"],
            svec!["19", "s"],
            svec!["20", "t"],
            svec!["21", "u"],
            svec!["22", "v"],
            svec!["23", "w"],
            svec!["24", "x"],
            svec!["25", "y"],
            svec!["26", "z"],
        ],
    );

    let mut cmd = wrk.command("sample");
    cmd.args(["--rng", "faster"])
        .args(["--seed", "42"])
        .arg("4")
        .arg("in.csv");

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let expected = vec![
        svec!["R", "S"],
        svec!["11", "k"],
        svec!["15", "o"],
        svec!["21", "u"],
        svec!["22", "v"],
    ];
    assert_eq!(got, expected);
}

#[test]
fn sample_indexed_random_access_secure() {
    let wrk = Workdir::new("sample_indexed_random_access_secure");
    wrk.create_indexed(
        "in.csv",
        vec![
            svec!["R", "S"],
            svec!["1", "b"],
            svec!["2", "a"],
            svec!["3", "d"],
            svec!["4", "c"],
            svec!["5", "f"],
            svec!["6", "e"],
            svec!["7", "i"],
            svec!["8", "h"],
            svec!["9", "i"],
            svec!["10", "j"],
            svec!["11", "k"],
            svec!["12", "l"],
            svec!["13", "m"],
            svec!["14", "n"],
            svec!["15", "o"],
            svec!["16", "p"],
            svec!["17", "q"],
            svec!["18", "r"],
            svec!["19", "s"],
            svec!["20", "t"],
            svec!["21", "u"],
            svec!["22", "v"],
            svec!["23", "w"],
            svec!["24", "x"],
            svec!["25", "y"],
            svec!["26", "z"],
        ],
    );

    let mut cmd = wrk.command("sample");
    cmd.args(["--rng", "cryptosecure"])
        .args(["--seed", "42"])
        .arg("4")
        .arg("in.csv");

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let expected = vec![
        svec!["R", "S"],
        svec!["1", "b"],
        svec!["3", "d"],
        svec!["7", "i"],
        svec!["10", "j"],
    ];
    assert_eq!(got, expected);
}

#[test]
fn sample_percentage_negative_sample_size_error() {
    let wrk = Workdir::new("sample_negative");
    wrk.create_indexed(
        "in.csv",
        vec![
            svec!["R", "S"],
            svec!["1", "b"],
            svec!["2", "a"],
            svec!["3", "d"],
            svec!["4", "c"],
            svec!["5", "f"],
            svec!["6", "e"],
            svec!["7", "i"],
            svec!["8", "h"],
            svec!["8", "h"],
        ],
    );

    let mut cmd = wrk.command("sample");
    cmd.args(["--seed", "42"]).arg("-5").arg("in.csv");

    wrk.assert_err(&mut cmd);
}

#[test]
fn sample_bernoulli_seed() {
    let wrk = Workdir::new("sample_bernoulli_seed");
    wrk.create(
        "in.csv",
        vec![
            svec!["R", "S"],
            svec!["1", "b"],
            svec!["2", "a"],
            svec!["3", "d"],
            svec!["4", "c"],
            svec!["5", "f"],
            svec!["6", "e"],
            svec!["7", "i"],
            svec!["8", "h"],
        ],
    );

    let mut cmd = wrk.command("sample");
    cmd.args(["--bernoulli"])
        .args(["--seed", "42"])
        .arg("0.5")
        .arg("in.csv");

    wrk.assert_success(&mut cmd);

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let expected = vec![
        svec!["R", "S"],
        svec!["4", "c"],
        svec!["5", "f"],
        svec!["6", "e"],
    ];
    assert_eq!(got, expected);
}

#[test]
fn sample_bernoulli_seed_faster() {
    let wrk = Workdir::new("sample_bernoulli_seed_faster");
    wrk.create(
        "in.csv",
        vec![
            svec!["R", "S"],
            svec!["1", "b"],
            svec!["2", "a"],
            svec!["3", "d"],
            svec!["4", "c"],
            svec!["5", "f"],
            svec!["6", "e"],
            svec!["7", "i"],
            svec!["8", "h"],
        ],
    );

    let mut cmd = wrk.command("sample");
    cmd.args(["--bernoulli"])
        .args(["--rng", "faster"])
        .args(["--seed", "76"])
        .arg("0.45")
        .arg("in.csv");

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let expected = vec![
        svec!["R", "S"],
        svec!["1", "b"],
        svec!["2", "a"],
        svec!["4", "c"],
        svec!["6", "e"],
    ];
    assert_eq!(got, expected);
}

#[test]
fn sample_bernoulli_seed_secure() {
    let wrk = Workdir::new("sample_bernoulli_seed_secure");
    wrk.create(
        "in.csv",
        vec![
            svec!["R", "S"],
            svec!["1", "b"],
            svec!["2", "a"],
            svec!["3", "d"],
            svec!["4", "c"],
            svec!["5", "f"],
            svec!["6", "e"],
            svec!["7", "i"],
            svec!["8", "h"],
        ],
    );

    let mut cmd = wrk.command("sample");
    cmd.args(["--bernoulli"])
        .args(["--rng", "cryptosecure"])
        .args(["--seed", "42"])
        .arg("0.5")
        .arg("in.csv");

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let expected = vec![svec!["R", "S"], svec!["3", "d"]];
    assert_eq!(got, expected);
}

#[test]
fn sample_bernoulli_invalid_probability() {
    let wrk = Workdir::new("sample_bernoulli_invalid");
    wrk.create(
        "in.csv",
        vec![svec!["R", "S"], svec!["1", "b"], svec!["2", "a"]],
    );

    // Test probability > 1.0
    let mut cmd = wrk.command("sample");
    cmd.args(["--bernoulli"]).arg("1.5").arg("in.csv");
    wrk.assert_err(&mut cmd);

    // Test probability <= 0.0
    let mut cmd = wrk.command("sample");
    cmd.args(["--bernoulli"]).arg("0.0").arg("in.csv");
    wrk.assert_err(&mut cmd);

    let mut cmd = wrk.command("sample");
    cmd.args(["--bernoulli"]).arg("-0.5").arg("in.csv");
    wrk.assert_err(&mut cmd);
}

#[test]
fn sample_systematic() {
    let wrk = Workdir::new("sample_systematic");
    wrk.create(
        "in.csv",
        vec![
            svec!["R", "S"],
            svec!["1", "b"],
            svec!["2", "a"],
            svec!["3", "d"],
            svec!["4", "c"],
            svec!["5", "f"],
            svec!["6", "e"],
            svec!["7", "i"],
            svec!["8", "h"],
            svec!["9", "g"],
            svec!["10", "j"],
        ],
    );

    let mut cmd = wrk.command("sample");
    cmd.args(["--systematic", "first"]).arg("3").arg("in.csv");

    wrk.assert_success(&mut cmd);

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let expected = vec![
        svec!["R", "S"],
        svec!["1", "b"],
        svec!["4", "c"],
        svec!["7", "i"],
        svec!["10", "j"],
    ];
    assert_eq!(got, expected);
}

#[test]
fn sample_stratified() {
    let wrk = Workdir::new("sample_stratified");
    wrk.create(
        "in.csv",
        vec![
            svec!["Group", "Value"],
            svec!["A", "1"],
            svec!["A", "2"],
            svec!["A", "3"],
            svec!["B", "4"],
            svec!["B", "5"],
            svec!["B", "6"],
            svec!["C", "7"],
            svec!["C", "8"],
            svec!["C", "9"],
            svec!["C", "10"],
            svec!["C", "11"],
            svec!["D", "12"],
        ],
    );

    let mut cmd = wrk.command("sample");
    cmd.args(["--stratified", "Group"])
        .args(["--seed", "42"])
        .arg("2")
        .arg("in.csv");

    wrk.assert_success(&mut cmd);

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let expected = vec![
        svec!["Group", "Value"],
        svec!["A", "3"],
        svec!["A", "2"],
        svec!["B", "4"],
        svec!["B", "6"],
        svec!["C", "9"],
        svec!["C", "8"],
        svec!["D", "12"],
    ];
    assert_eq!(got, expected);
}

#[test]
fn sample_stratified_large_sample_size() {
    let wrk = Workdir::new("sample_stratified_large_sample_size");
    wrk.create(
        "in.csv",
        vec![
            svec!["Group", "Value"],
            svec!["A", "1"],
            svec!["A", "2"],
            svec!["A", "3"],
            svec!["B", "4"],
            svec!["B", "5"],
            svec!["B", "6"],
            svec!["C", "7"],
            svec!["C", "8"],
            svec!["C", "9"],
            svec!["C", "10"],
            svec!["C", "11"],
            svec!["D", "12"],
        ],
    );

    let mut cmd = wrk.command("sample");
    cmd.args(["--stratified", "Group"])
        .args(["--seed", "42"])
        .arg("100")
        .arg("in.csv");

    wrk.assert_success(&mut cmd);

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let expected = vec![
        svec!["Group", "Value"],
        svec!["A", "1"],
        svec!["A", "2"],
        svec!["A", "3"],
        svec!["B", "4"],
        svec!["B", "5"],
        svec!["B", "6"],
        svec!["C", "7"],
        svec!["C", "8"],
        svec!["C", "9"],
        svec!["C", "10"],
        svec!["C", "11"],
        svec!["D", "12"],
    ];
    assert_eq!(got, expected);
}

#[test]
fn sample_weighted() {
    let wrk = Workdir::new("sample_weighted");
    wrk.create(
        "in.csv",
        vec![
            svec!["ID", "Weight"],
            svec!["1", "10"],
            svec!["2", "20"],
            svec!["3", "30"],
            svec!["4", "40"],
            svec!["5", "50"],
            svec!["6", "60"],
            svec!["7", "70"],
            svec!["8", "80"],
            svec!["9", "90"],
            svec!["10", "100"],
        ],
    );

    let mut cmd = wrk.command("sample");
    cmd.args(["--weighted", "ID"])
        .args(["--seed", "42"])
        .arg("4")
        .arg("in.csv");

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let expected = vec![
        svec!["ID", "Weight"],
        svec!["5", "50"],
        svec!["6", "60"],
        svec!["9", "90"],
        svec!["10", "100"],
    ];
    assert_eq!(got, expected);
}

#[test]
fn sample_cluster() {
    let wrk = Workdir::new("sample_cluster");
    wrk.create(
        "in.csv",
        vec![
            svec!["Household", "Person", "Age"],
            svec!["H1", "P1", "25"],
            svec!["H1", "P2", "30"],
            svec!["H1", "P3", "35"],
            svec!["H2", "P3", "45"],
            svec!["H2", "P4", "50"],
            svec!["H2", "P5", "55"],
            svec!["H3", "P5", "35"],
            svec!["H3", "P6", "40"],
            svec!["H3", "P7", "45"],
            svec!["H4", "P7", "28"],
            svec!["H4", "P8", "32"],
            svec!["H4", "P9", "36"],
            svec!["H4", "P10", "40"],
            svec!["H5", "P11", "44"],
            svec!["H5", "P12", "48"],
        ],
    );

    let mut cmd = wrk.command("sample");
    cmd.args(["--cluster", "Household"])
        .args(["--seed", "42"])
        .arg("2")
        .arg("in.csv");

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let expected = vec![
        svec!["Household", "Person", "Age"],
        svec!["H1", "P1", "25"],
        svec!["H1", "P2", "30"],
        svec!["H1", "P3", "35"],
        svec!["H3", "P5", "35"],
        svec!["H3", "P6", "40"],
        svec!["H3", "P7", "45"],
    ];
    assert_eq!(got, expected);
}

#[test]
fn sample_stratified_invalid_column() {
    let wrk = Workdir::new("sample_stratified_invalid");
    wrk.create(
        "in.csv",
        vec![svec!["Group", "Value"], svec!["A", "1"], svec!["B", "2"]],
    );

    let mut cmd = wrk.command("sample");
    cmd.args(["--stratified", "999"]).arg("1").arg("in.csv");

    wrk.assert_err(&mut cmd);
}

#[test]
fn sample_weighted_negative_weights() {
    let wrk = Workdir::new("sample_weighted_negative");
    wrk.create(
        "in.csv",
        vec![
            svec!["ID", "Weight"],
            svec!["1", "-10"],
            svec!["2", "20"],
            svec!["3", "30"],
            svec!["4", "40"],
            svec!["5", "-50"],
        ],
    );

    let mut cmd = wrk.command("sample");
    cmd.args(["--weighted", "1"]).arg("1").arg("in.csv");

    wrk.assert_err(&mut cmd);
}

#[test]
fn sample_stratified_empty_stratum() {
    let wrk = Workdir::new("sample_stratified_empty");
    wrk.create(
        "in.csv",
        vec![
            svec!["Group", "Value"],
            svec!["A", "1"],
            svec!["", "2"], // empty stratum
            svec!["A", "3"],
            svec!["B", "4"],
            svec!["", "5"], // another empty stratum
        ],
    );

    let mut cmd = wrk.command("sample");
    cmd.args(["--stratified", "Group"])
        .args(["--seed", "42"])
        .arg("2")
        .arg("in.csv");

    wrk.assert_success(&mut cmd);

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let expected = vec![
        svec!["Group", "Value"],
        svec!["", "2"],
        svec!["", "5"],
        svec!["A", "1"],
        svec!["A", "3"],
        svec!["B", "4"],
    ];
    assert_eq!(got, expected);
}

#[test]
fn sample_weighted_zero_weights() {
    let wrk = Workdir::new("sample_weighted_zero");
    wrk.create(
        "in.csv",
        vec![
            svec!["ID", "Weight"],
            svec!["1", "0"],
            svec!["2", "0"],
            svec!["3", "30"],
            svec!["4", "0"],
            svec!["5", "50"],
        ],
    );

    let mut cmd = wrk.command("sample");
    cmd.args(["--weighted", "Weight"])
        .args(["--seed", "42"])
        .arg("2")
        .arg("in.csv");

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let expected = vec![svec!["ID", "Weight"], svec!["3", "30"], svec!["5", "50"]];
    assert_eq!(got, expected);
}

#[test]
fn sample_cluster_single_record() {
    let wrk = Workdir::new("sample_cluster_single");
    wrk.create(
        "in.csv",
        vec![
            svec!["Cluster", "Value"],
            svec!["A", "1"], // single record cluster
            svec!["B", "2"],
            svec!["B", "3"],
            svec!["C", "4"], // single record cluster
        ],
    );

    let mut cmd = wrk.command("sample");
    cmd.args(["--cluster", "Cluster"])
        .args(["--seed", "42"])
        .arg("2")
        .arg("in.csv");

    wrk.assert_success(&mut cmd);

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let expected = vec![
        svec!["Cluster", "Value"],
        svec!["A", "1"],
        svec!["B", "2"],
        svec!["B", "3"],
    ];
    assert_eq!(got, expected);
}

#[test]
fn sample_systematic_with_headers() {
    let wrk = Workdir::new("sample_systematic_headers");
    wrk.create(
        "in.csv",
        vec![
            svec!["Header1", "Header2"], // should be preserved
            svec!["1", "a"],
            svec!["2", "b"],
            svec!["3", "c"],
            svec!["4", "d"],
            svec!["5", "e"],
            svec!["6", "f"],
            svec!["7", "g"],
            svec!["8", "h"],
            svec!["9", "i"],
        ],
    );

    let mut cmd = wrk.command("sample");
    cmd.args(["--systematic", "first"]).arg("3").arg("in.csv");

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let expected = vec![
        svec!["Header1", "Header2"],
        svec!["1", "a"],
        svec!["4", "d"],
        svec!["7", "g"],
    ];
    assert_eq!(got, expected);
}

#[test]
fn sample_systematic_with_headers_random_with_seed() {
    let wrk = Workdir::new("sample_systematic_headers_random_with_seed");
    wrk.create(
        "in.csv",
        vec![
            svec!["Header1", "Header2"], // should be preserved
            svec!["1", "a"],
            svec!["2", "b"],
            svec!["3", "c"],
            svec!["4", "d"],
            svec!["5", "e"],
            svec!["6", "f"],
            svec!["7", "g"],
            svec!["8", "h"],
            svec!["9", "i"],
            svec!["10", "j"],
        ],
    );

    let mut cmd = wrk.command("sample");
    cmd.args(["--systematic", "random", "--seed", "65"])
        .arg("4.5")
        .arg("in.csv");

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let expected = vec![
        svec!["Header1", "Header2"],
        svec!["5", "e"],
        svec!["9", "i"],
    ];
    assert_eq!(got, expected);
}

#[test]
fn sample_systematic_no_headers() {
    let wrk = Workdir::new("sample_systematic_no_headers");
    wrk.create(
        "in.csv",
        vec![
            svec!["1", "a"],
            svec!["2", "b"],
            svec!["3", "c"],
            svec!["4", "d"],
            svec!["5", "e"],
            svec!["6", "f"],
            svec!["7", "g"],
            svec!["8", "h"],
            svec!["9", "i"],
            svec!["10", "j"],
        ],
    );

    let mut cmd = wrk.command("sample");
    cmd.args(["--systematic", "first", "--no-headers"])
        .arg("3.3")
        .arg("in.csv");

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let expected = vec![svec!["1", "a"], svec!["4", "d"], svec!["7", "g"]];
    assert_eq!(got, expected);
}

#[test]
fn sample_multiple_methods_error() {
    let wrk = Workdir::new("sample_multiple_methods");
    wrk.create(
        "in.csv",
        vec![svec!["ID", "Value"], svec!["1", "a"], svec!["2", "b"]],
    );

    // Test combining bernoulli with systematic
    let mut cmd = wrk.command("sample");
    cmd.args(["--bernoulli", "--systematic", "first"])
        .arg("0.5")
        .arg("in.csv");
    wrk.assert_err(&mut cmd);

    // Test combining weighted with stratified
    let mut cmd = wrk.command("sample");
    cmd.args(["--weighted", "ID", "--stratified", "ID"])
        .arg("1")
        .arg("in.csv");
    wrk.assert_err(&mut cmd);
}

#[test]
fn sample_invalid_rng() {
    let wrk = Workdir::new("sample_invalid_rng");
    wrk.create(
        "in.csv",
        vec![svec!["ID", "Value"], svec!["1", "a"], svec!["2", "b"]],
    );

    let mut cmd = wrk.command("sample");
    cmd.args(["--rng", "invalid_rng"]).arg("1").arg("in.csv");
    wrk.assert_err(&mut cmd);
}

#[test]
fn sample_systematic_invalid_interval() {
    let wrk = Workdir::new("sample_systematic_invalid_interval");
    wrk.create(
        "in.csv",
        vec![
            svec!["R", "S"],
            svec!["1", "b"],
            svec!["2", "a"],
            svec!["3", "d"],
        ],
    );

    // Test interval of 0
    let mut cmd = wrk.command("sample");
    cmd.args(["--systematic", "first"]).arg("0").arg("in.csv");
    wrk.assert_err(&mut cmd);

    // Test negative interval
    let mut cmd = wrk.command("sample");
    cmd.args(["--systematic", "first"]).arg("-2").arg("in.csv");
    wrk.assert_err(&mut cmd);
}

#[test]
fn sample_weighted_invalid_weights() {
    let wrk = Workdir::new("sample_weighted_invalid");
    wrk.create(
        "in.csv",
        vec![
            svec!["ID", "Weight"],
            svec!["1", "abc"], // non-numeric weight -> treated as 0
            svec!["2", "20.5"],
            svec!["3", ""], // empty weight -> treated as 0
            svec!["4", "40"],
        ],
    );

    let mut cmd = wrk.command("sample");
    cmd.args(["--weighted", "Weight"])
        .args(["--seed", "42"])
        .arg("2")
        .arg("in.csv");

    wrk.assert_success(&mut cmd);

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    // Records with invalid weights are treated as having zero weight
    // we requested two samples but returning only one as we ran out of records
    let expected = vec![svec!["ID", "Weight"], svec!["4", "40"]];
    assert_eq!(got, expected);
}

#[test]
fn sample_cluster_too_many_clusters_with_stats_cache() {
    let wrk = Workdir::new("sample_cluster_too_many_with_stats_cache");
    wrk.create(
        "in.csv",
        vec![
            svec!["Cluster", "Value"],
            svec!["A", "1"],
            svec!["B", "2"],
            svec!["C", "3"],
        ],
    );

    // create stats cache
    let mut stats_cmd = wrk.command("stats");
    stats_cmd.args(["in.csv", "-E", "--stats-jsonl"]);

    wrk.assert_success(&mut stats_cmd);

    // Request more clusters than exist, this error only happens with a stats cache
    let mut cmd = wrk.command("sample");
    cmd.args(["--cluster", "Cluster"])
        .args(["--seed", "42"])
        .arg("5") // Only 3 clusters exist
        .arg("in.csv");

    wrk.assert_err(&mut cmd);
}

#[test]
fn sample_stratified_with_delimiter() {
    let wrk = Workdir::new("sample_stratified_delimiter");
    wrk.create_with_delim(
        "in.csv",
        vec![
            svec!["Group", "Value"],
            svec!["A", "1"],
            svec!["A", "2"],
            svec!["B", "3"],
            svec!["B", "4"],
        ],
        b'|',
    );

    let mut cmd = wrk.command("sample");
    cmd.args(["--stratified", "Group"])
        .args(["--seed", "42"])
        .args(["--delimiter", "|"])
        .arg("1")
        .arg("in.csv");

    wrk.assert_success(&mut cmd);

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let expected = vec![svec!["Group|Value"], svec!["A|2"], svec!["B|3"]];
    assert_eq!(got, expected);
}

#[test]
fn sample_weighted_all_zero_weights() {
    let wrk = Workdir::new("sample_weighted_all_zero");
    wrk.create(
        "in.csv",
        vec![
            svec!["ID", "Weight"],
            svec!["1", "0"],
            svec!["2", "0"],
            svec!["3", "0"],
        ],
    );

    let mut cmd = wrk.command("sample");
    cmd.args(["--weighted", "Weight"])
        .args(["--seed", "42"])
        .arg("2")
        .arg("in.csv");

    wrk.assert_err(&mut cmd);
}

#[test]
fn sample_systematic_fractional() {
    let wrk = Workdir::new("sample_systematic_fractional");
    wrk.create(
        "in.csv",
        vec![
            svec!["ID", "Value"],
            svec!["1", "a"],
            svec!["2", "b"],
            svec!["3", "c"],
            svec!["4", "d"],
            svec!["5", "e"],
            svec!["6", "f"],
            svec!["7", "g"],
            svec!["8", "h"],
            svec!["9", "i"],
            svec!["10", "j"],
        ],
    );

    // Test with fractional interval (3.5 means every 3rd record and 50% of population)
    let mut cmd = wrk.command("sample");
    cmd.args(["--systematic", "first"])
        .args(["--seed", "42"])
        .arg("3.5")
        .arg("in.csv");

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let expected = vec![
        svec!["ID", "Value"],
        svec!["1", "a"],
        svec!["4", "d"],
        svec!["7", "g"],
        svec!["10", "j"],
    ];
    assert_eq!(got, expected);
}

#[test]
fn sample_multiple_sampling_methods_error() {
    let wrk = Workdir::new("sample_multiple_methods_error");
    wrk.create(
        "in.csv",
        vec![svec!["ID", "Value"], svec!["1", "a"], svec!["2", "b"]],
    );

    // Test combining cluster with weighted
    let mut cmd = wrk.command("sample");
    cmd.args(["--cluster", "ID", "--weighted", "Value"])
        .arg("1")
        .arg("in.csv");
    wrk.assert_err(&mut cmd);

    // Test combining systematic with stratified
    let mut cmd = wrk.command("sample");
    cmd.args(["--systematic", "first", "--stratified", "ID"])
        .arg("1")
        .arg("in.csv");
    wrk.assert_err(&mut cmd);
}

#[test]
#[ignore = "requires remote NYC311-50k.csv fixture removed to reduce repo size"]
fn sample_remote_bernoulli_streaming_standard_rng() {
    let wrk = Workdir::new("sample_remote_bernoulli_streaming_standard_rng");

    // Use a small test file from the qsv repository that we know supports range requests
    let test_url = "https://raw.githubusercontent.com/dathere/qsv/refs/heads/master/resources/test/NYC311-50k.csv";

    let mut cmd = wrk.command("sample");
    cmd.args(["--bernoulli"])
        .args(["--seed", "42"])
        .arg("0.3") // 30% probability
        .arg(test_url);

    wrk.assert_success(&mut cmd);

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);

    // Verify we got the header
    assert_eq!(
        got[0],
        vec![
            "Unique Key",
            "Created Date",
            "Closed Date",
            "Agency",
            "Agency Name",
            "Complaint Type",
            "Descriptor",
            "Location Type",
            "Incident Zip",
            "Incident Address",
            "Street Name",
            "Cross Street 1",
            "Cross Street 2",
            "Intersection Street 1",
            "Intersection Street 2",
            "Address Type",
            "City",
            "Landmark",
            "Facility Type",
            "Status",
            "Due Date",
            "Resolution Description",
            "Resolution Action Updated Date",
            "Community Board",
            "BBL",
            "Borough",
            "X Coordinate (State Plane)",
            "Y Coordinate (State Plane)",
            "Open Data Channel Type",
            "Park Facility Name",
            "Park Borough",
            "Vehicle Type",
            "Taxi Company Borough",
            "Taxi Pick Up Location",
            "Bridge Highway Name",
            "Bridge Highway Direction",
            "Road Ramp",
            "Bridge Highway Segment",
            "Latitude",
            "Longitude",
            "Location",
        ]
    );

    // Verify we got some records (exact count as we're using a seed)
    assert_eq!(got.len(), 14_941);

    // Verify the structure of sampled records
    for record in got.iter().skip(1) {
        assert_eq!(record.len(), 41); // Each record should have position and title
        assert!(!record[0].is_empty()); // Unique Key should not be empty
        assert!(!record[1].is_empty()); // Created Date should not be empty
    }
}

#[test]
#[ignore = "requires remote NYC311-50k.csv fixture removed to reduce repo size"]
fn sample_remote_bernoulli_streaming_cryptosecure() {
    let wrk = Workdir::new("sample_remote_bernoulli_streaming_cryptosecure");

    // Use a small test file from the qsv repository that we know supports range requests
    let test_url = "https://raw.githubusercontent.com/dathere/qsv/refs/heads/master/resources/test/NYC311-50k.csv";

    let mut cmd = wrk.command("sample");
    cmd.args(["--bernoulli"])
        .args(["--rng", "cryptosecure"])
        .args(["--seed", "42"])
        .arg("0.3") // 30% probability
        .arg(test_url);

    wrk.assert_success(&mut cmd);

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);

    // Verify we got the header
    assert_eq!(
        got[0],
        vec![
            "Unique Key",
            "Created Date",
            "Closed Date",
            "Agency",
            "Agency Name",
            "Complaint Type",
            "Descriptor",
            "Location Type",
            "Incident Zip",
            "Incident Address",
            "Street Name",
            "Cross Street 1",
            "Cross Street 2",
            "Intersection Street 1",
            "Intersection Street 2",
            "Address Type",
            "City",
            "Landmark",
            "Facility Type",
            "Status",
            "Due Date",
            "Resolution Description",
            "Resolution Action Updated Date",
            "Community Board",
            "BBL",
            "Borough",
            "X Coordinate (State Plane)",
            "Y Coordinate (State Plane)",
            "Open Data Channel Type",
            "Park Facility Name",
            "Park Borough",
            "Vehicle Type",
            "Taxi Company Borough",
            "Taxi Pick Up Location",
            "Bridge Highway Name",
            "Bridge Highway Direction",
            "Road Ramp",
            "Bridge Highway Segment",
            "Latitude",
            "Longitude",
            "Location",
        ]
    );

    // Verify we got some records (exact count as we're using a seed)
    assert_eq!(got.len(), 14_823);

    // Verify the structure of sampled records
    for record in got.iter().skip(1) {
        assert_eq!(record.len(), 41); // Each record should have position and title
        assert!(!record[0].is_empty()); // Unique Key should not be empty
        assert!(!record[1].is_empty()); // Created Date should not be empty
    }
}

#[test]
fn sample_timeseries_basic() {
    let wrk = Workdir::new("sample_timeseries_basic");
    wrk.create(
        "in.csv",
        vec![
            svec!["timestamp", "value"],
            svec!["2024-01-01T00:00:00Z", "10"],
            svec!["2024-01-01T01:00:00Z", "20"],
            svec!["2024-01-01T02:00:00Z", "30"],
            svec!["2024-01-01T03:00:00Z", "40"],
            svec!["2024-01-01T04:00:00Z", "50"],
        ],
    );

    let mut cmd = wrk.command("sample");
    cmd.args(["--timeseries", "timestamp"])
        .args(["--ts-interval", "1h"])
        .args(["--ts-start", "first"])
        .arg("1")
        .arg("in.csv");

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let expected = vec![
        svec!["timestamp", "value"],
        svec!["2024-01-01T00:00:00Z", "10"],
        svec!["2024-01-01T01:00:00Z", "20"],
        svec!["2024-01-01T02:00:00Z", "30"],
        svec!["2024-01-01T03:00:00Z", "40"],
        svec!["2024-01-01T04:00:00Z", "50"],
    ];
    assert_eq!(got, expected);
}

#[test]
fn sample_timeseries_daily() {
    let wrk = Workdir::new("sample_timeseries_daily");
    wrk.create(
        "in.csv",
        vec![
            svec!["date", "value"],
            svec!["2024-01-01", "10"],
            svec!["2024-01-02", "20"],
            svec!["2024-01-03", "30"],
            svec!["2024-01-04", "40"],
            svec!["2024-01-05", "50"],
        ],
    );

    let mut cmd = wrk.command("sample");
    cmd.args(["--timeseries", "date"])
        .args(["--ts-interval", "1d"])
        .args(["--ts-start", "first"])
        .arg("1")
        .arg("in.csv");

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let expected = vec![
        svec!["date", "value"],
        svec!["2024-01-01", "10"],
        svec!["2024-01-02", "20"],
        svec!["2024-01-03", "30"],
        svec!["2024-01-04", "40"],
        svec!["2024-01-05", "50"],
    ];
    assert_eq!(got, expected);
}

#[test]
fn sample_timeseries_start_last() {
    let wrk = Workdir::new("sample_timeseries_start_last");
    wrk.create(
        "in.csv",
        vec![
            svec!["timestamp", "value"],
            svec!["2024-01-01T00:00:00Z", "10"],
            svec!["2024-01-01T01:00:00Z", "20"],
            svec!["2024-01-01T02:00:00Z", "30"],
        ],
    );

    let mut cmd = wrk.command("sample");
    cmd.args(["--timeseries", "timestamp"])
        .args(["--ts-interval", "1h"])
        .args(["--ts-start", "last"])
        .arg("1")
        .arg("in.csv");

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    // Starting from last (02:00), intervals are calculated backwards but output is still
    // chronological The start_time affects which intervals are selected, but records are output
    // in time order
    let expected = vec![
        svec!["timestamp", "value"],
        svec!["2024-01-01T00:00:00Z", "10"],
        svec!["2024-01-01T01:00:00Z", "20"],
        svec!["2024-01-01T02:00:00Z", "30"],
    ];
    assert_eq!(got, expected);
}

#[test]
fn sample_timeseries_aggregate_mean() {
    let wrk = Workdir::new("sample_timeseries_aggregate_mean");
    wrk.create(
        "in.csv",
        vec![
            svec!["timestamp", "value"],
            svec!["2024-01-01T00:00:00Z", "10"],
            svec!["2024-01-01T00:30:00Z", "20"],
            svec!["2024-01-01T01:00:00Z", "30"],
            svec!["2024-01-01T01:30:00Z", "40"],
        ],
    );

    let mut cmd = wrk.command("sample");
    cmd.args(["--timeseries", "timestamp"])
        .args(["--ts-interval", "1h"])
        .args(["--ts-aggregate", "mean"])
        .args(["--ts-start", "first"])
        .arg("1")
        .arg("in.csv");

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    assert_eq!(got.len(), 3); // Header + 2 aggregated records (one per hour)
    assert_eq!(got[0], svec!["timestamp", "value"]);
    // First hour: mean of 10 and 20 = 15
    let first_value: f64 = got[1][1].parse().unwrap_or(0.0);
    assert!(
        (first_value - 15.0).abs() < 0.01,
        "Expected mean ~15.0, got {}",
        first_value
    );
    // Second hour: mean of 30 and 40 = 35
    let second_value: f64 = got[2][1].parse().unwrap_or(0.0);
    assert!(
        (second_value - 35.0).abs() < 0.01,
        "Expected mean ~35.0, got {}",
        second_value
    );
}

#[test]
fn sample_timeseries_invalid_column() {
    let wrk = Workdir::new("sample_timeseries_invalid_column");
    wrk.create(
        "in.csv",
        vec![
            svec!["timestamp", "value"],
            svec!["2024-01-01T00:00:00Z", "10"],
        ],
    );

    let mut cmd = wrk.command("sample");
    cmd.args(["--timeseries", "nonexistent"])
        .args(["--ts-interval", "1h"])
        .arg("1")
        .arg("in.csv");

    wrk.assert_err(&mut cmd);
}

#[test]
fn sample_timeseries_invalid_interval() {
    let wrk = Workdir::new("sample_timeseries_invalid_interval");
    wrk.create(
        "in.csv",
        vec![
            svec!["timestamp", "value"],
            svec!["2024-01-01T00:00:00Z", "10"],
        ],
    );

    let mut cmd = wrk.command("sample");
    cmd.args(["--timeseries", "timestamp"])
        .args(["--ts-interval", "invalid"])
        .arg("1")
        .arg("in.csv");

    wrk.assert_err(&mut cmd);
}

#[test]
fn sample_timeseries_adaptive_business_hours() {
    let wrk = Workdir::new("sample_timeseries_adaptive_business_hours");
    // Create data with business hours (9am-5pm) and non-business hours
    wrk.create(
        "in.csv",
        vec![
            svec!["timestamp", "value"],
            svec!["2024-01-01T08:00:00Z", "10"], // Before business hours
            svec!["2024-01-01T09:00:00Z", "20"], // Start of business hours (Monday)
            svec!["2024-01-01T12:00:00Z", "30"], // During business hours
            svec!["2024-01-01T17:00:00Z", "40"], // End of business hours
            svec!["2024-01-01T18:00:00Z", "50"], // After business hours
        ],
    );

    let mut cmd = wrk.command("sample");
    cmd.args(["--timeseries", "timestamp"])
        .args(["--ts-interval", "1d"])
        .args(["--ts-adaptive", "business-hours"])
        .arg("1")
        .arg("in.csv");

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    // Should prefer business hours records (9am-5pm on weekdays)
    assert_eq!(got.len(), 2); // Header + 1 record
    assert_eq!(got[0], svec!["timestamp", "value"]);
    // Should prefer a business hours record (9:00 or 12:00)
    assert!(
        got[1][0] == "2024-01-01T09:00:00Z"
            || got[1][0] == "2024-01-01T12:00:00Z"
            || got[1][0] == "2024-01-01T17:00:00Z",
        "Expected business hours timestamp, got {}",
        got[1][0]
    );
}

#[test]
fn sample_timeseries_adaptive_weekends() {
    let wrk = Workdir::new("sample_timeseries_adaptive_weekends");
    // Create data with weekend (Saturday) and weekday records
    wrk.create(
        "in.csv",
        vec![
            svec!["timestamp", "value"],
            svec!["2024-01-05T10:00:00Z", "10"], // Friday (weekday)
            svec!["2024-01-06T10:00:00Z", "20"], // Saturday (weekend)
            svec!["2024-01-07T10:00:00Z", "30"], // Sunday (weekend)
            svec!["2024-01-08T10:00:00Z", "40"], // Monday (weekday)
        ],
    );

    let mut cmd = wrk.command("sample");
    cmd.args(["--timeseries", "timestamp"])
        .args(["--ts-interval", "1d"])
        .args(["--ts-adaptive", "weekends"])
        .arg("1")
        .arg("in.csv");

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    // Should prefer weekend records
    // We have 4 days (Fri, Sat, Sun, Mon), so 4 records + header = 5 total
    assert_eq!(got.len(), 5); // Header + 4 records (one per day)
    assert_eq!(got[0], svec!["timestamp", "value"]);
    // Check that weekend records are preferred when available
    // For Saturday, should get the weekend record
    let sat_record = got.iter().find(|r| r[0] == "2024-01-06T10:00:00Z");
    assert!(sat_record.is_some(), "Should include Saturday record");
    // For Sunday, should get the weekend record
    let sun_record = got.iter().find(|r| r[0] == "2024-01-07T10:00:00Z");
    assert!(sun_record.is_some(), "Should include Sunday record");
}

#[test]
fn sample_timeseries_adaptive_business_days() {
    let wrk = Workdir::new("sample_timeseries_adaptive_business_days");
    // Create data with weekday and weekend records
    wrk.create(
        "in.csv",
        vec![
            svec!["timestamp", "value"],
            svec!["2024-01-05T10:00:00Z", "10"], // Friday (weekday)
            svec!["2024-01-06T10:00:00Z", "20"], // Saturday (weekend)
            svec!["2024-01-08T10:00:00Z", "30"], // Monday (weekday)
        ],
    );

    let mut cmd = wrk.command("sample");
    cmd.args(["--timeseries", "timestamp"])
        .args(["--ts-interval", "1d"])
        .args(["--ts-adaptive", "business-days"])
        .arg("1")
        .arg("in.csv");

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    // Should prefer weekday records
    assert_eq!(got.len(), 4); // Header + 3 records
    assert_eq!(got[0], svec!["timestamp", "value"]);
    // Should prefer weekday records when available
    let friday_record = got.iter().find(|r| r[0] == "2024-01-05T10:00:00Z");
    assert!(friday_record.is_some(), "Should include Friday record");
    let monday_record = got.iter().find(|r| r[0] == "2024-01-08T10:00:00Z");
    assert!(monday_record.is_some(), "Should include Monday record");
}

#[test]
fn sample_timeseries_aggregate_first() {
    let wrk = Workdir::new("sample_timeseries_aggregate_first");
    wrk.create(
        "in.csv",
        vec![
            svec!["timestamp", "value"],
            svec!["2024-01-01T00:00:00Z", "10"],
            svec!["2024-01-01T00:30:00Z", "20"],
            svec!["2024-01-01T01:00:00Z", "30"],
            svec!["2024-01-01T01:30:00Z", "40"],
        ],
    );

    let mut cmd = wrk.command("sample");
    cmd.args(["--timeseries", "timestamp"])
        .args(["--ts-interval", "1h"])
        .args(["--ts-aggregate", "first"])
        .arg("1")
        .arg("in.csv");

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    assert_eq!(got.len(), 3); // Header + 2 aggregated records
    assert_eq!(got[0], svec!["timestamp", "value"]);
    // First hour: first value should be 10
    assert_eq!(got[1][1], "10");
    // Second hour: first value should be 30
    assert_eq!(got[2][1], "30");
}

#[test]
fn sample_timeseries_aggregate_last() {
    let wrk = Workdir::new("sample_timeseries_aggregate_last");
    wrk.create(
        "in.csv",
        vec![
            svec!["timestamp", "value"],
            svec!["2024-01-01T00:00:00Z", "10"],
            svec!["2024-01-01T00:30:00Z", "20"],
            svec!["2024-01-01T01:00:00Z", "30"],
            svec!["2024-01-01T01:30:00Z", "40"],
        ],
    );

    let mut cmd = wrk.command("sample");
    cmd.args(["--timeseries", "timestamp"])
        .args(["--ts-interval", "1h"])
        .args(["--ts-aggregate", "last"])
        .arg("1")
        .arg("in.csv");

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    assert_eq!(got.len(), 3); // Header + 2 aggregated records
    assert_eq!(got[0], svec!["timestamp", "value"]);
    // First hour: last value should be 20
    assert_eq!(got[1][1], "20");
    // Second hour: last value should be 40
    assert_eq!(got[2][1], "40");
}

#[test]
fn sample_timeseries_aggregate_sum() {
    let wrk = Workdir::new("sample_timeseries_aggregate_sum");
    wrk.create(
        "in.csv",
        vec![
            svec!["timestamp", "value"],
            svec!["2024-01-01T00:00:00Z", "10"],
            svec!["2024-01-01T00:30:00Z", "20"],
            svec!["2024-01-01T01:00:00Z", "30"],
            svec!["2024-01-01T01:30:00Z", "40"],
        ],
    );

    let mut cmd = wrk.command("sample");
    cmd.args(["--timeseries", "timestamp"])
        .args(["--ts-interval", "1h"])
        .args(["--ts-aggregate", "sum"])
        .arg("1")
        .arg("in.csv");

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    assert_eq!(got.len(), 3); // Header + 2 aggregated records
    assert_eq!(got[0], svec!["timestamp", "value"]);
    // First hour: sum of 10 and 20 = 30
    let first_value: f64 = got[1][1].parse().unwrap_or(0.0);
    assert!(
        (first_value - 30.0).abs() < 0.01,
        "Expected sum ~30.0, got {}",
        first_value
    );
    // Second hour: sum of 30 and 40 = 70
    let second_value: f64 = got[2][1].parse().unwrap_or(0.0);
    assert!(
        (second_value - 70.0).abs() < 0.01,
        "Expected sum ~70.0, got {}",
        second_value
    );
}

#[test]
fn sample_timeseries_aggregate_count() {
    let wrk = Workdir::new("sample_timeseries_aggregate_count");
    wrk.create(
        "in.csv",
        vec![
            svec!["timestamp", "value"],
            svec!["2024-01-01T00:00:00Z", "10"],
            svec!["2024-01-01T00:30:00Z", "20"],
            svec!["2024-01-01T01:00:00Z", "30"],
            svec!["2024-01-01T01:30:00Z", "40"],
        ],
    );

    let mut cmd = wrk.command("sample");
    cmd.args(["--timeseries", "timestamp"])
        .args(["--ts-interval", "1h"])
        .args(["--ts-aggregate", "count"])
        .arg("1")
        .arg("in.csv");

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    assert_eq!(got.len(), 3); // Header + 2 aggregated records
    assert_eq!(got[0], svec!["timestamp", "value"]);
    // First hour: count should be 2
    let first_count: f64 = got[1][1].parse().unwrap_or(0.0);
    assert!(
        (first_count - 2.0).abs() < 0.01,
        "Expected count ~2.0, got {}",
        first_count
    );
    // Second hour: count should be 2
    let second_count: f64 = got[2][1].parse().unwrap_or(0.0);
    assert!(
        (second_count - 2.0).abs() < 0.01,
        "Expected count ~2.0, got {}",
        second_count
    );
}

#[test]
fn sample_timeseries_aggregate_min() {
    let wrk = Workdir::new("sample_timeseries_aggregate_min");
    wrk.create(
        "in.csv",
        vec![
            svec!["timestamp", "value"],
            svec!["2024-01-01T00:00:00Z", "10"],
            svec!["2024-01-01T00:30:00Z", "20"],
            svec!["2024-01-01T01:00:00Z", "30"],
            svec!["2024-01-01T01:30:00Z", "40"],
        ],
    );

    let mut cmd = wrk.command("sample");
    cmd.args(["--timeseries", "timestamp"])
        .args(["--ts-interval", "1h"])
        .args(["--ts-aggregate", "min"])
        .arg("1")
        .arg("in.csv");

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    assert_eq!(got.len(), 3); // Header + 2 aggregated records
    assert_eq!(got[0], svec!["timestamp", "value"]);
    // First hour: min of 10 and 20 = 10
    let first_value: f64 = got[1][1].parse().unwrap_or(0.0);
    assert!(
        (first_value - 10.0).abs() < 0.01,
        "Expected min ~10.0, got {}",
        first_value
    );
    // Second hour: min of 30 and 40 = 30
    let second_value: f64 = got[2][1].parse().unwrap_or(0.0);
    assert!(
        (second_value - 30.0).abs() < 0.01,
        "Expected min ~30.0, got {}",
        second_value
    );
}

#[test]
fn sample_timeseries_aggregate_max() {
    let wrk = Workdir::new("sample_timeseries_aggregate_max");
    wrk.create(
        "in.csv",
        vec![
            svec!["timestamp", "value"],
            svec!["2024-01-01T00:00:00Z", "10"],
            svec!["2024-01-01T00:30:00Z", "20"],
            svec!["2024-01-01T01:00:00Z", "30"],
            svec!["2024-01-01T01:30:00Z", "40"],
        ],
    );

    let mut cmd = wrk.command("sample");
    cmd.args(["--timeseries", "timestamp"])
        .args(["--ts-interval", "1h"])
        .args(["--ts-aggregate", "max"])
        .arg("1")
        .arg("in.csv");

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    assert_eq!(got.len(), 3); // Header + 2 aggregated records
    assert_eq!(got[0], svec!["timestamp", "value"]);
    // First hour: max of 10 and 20 = 20
    let first_value: f64 = got[1][1].parse().unwrap_or(0.0);
    assert!(
        (first_value - 20.0).abs() < 0.01,
        "Expected max ~20.0, got {}",
        first_value
    );
    // Second hour: max of 30 and 40 = 40
    let second_value: f64 = got[2][1].parse().unwrap_or(0.0);
    assert!(
        (second_value - 40.0).abs() < 0.01,
        "Expected max ~40.0, got {}",
        second_value
    );
}

#[test]
fn sample_timeseries_aggregate_median() {
    let wrk = Workdir::new("sample_timeseries_aggregate_median");
    wrk.create(
        "in.csv",
        vec![
            svec!["timestamp", "value"],
            svec!["2024-01-01T00:00:00Z", "10"],
            svec!["2024-01-01T00:30:00Z", "20"],
            svec!["2024-01-01T01:00:00Z", "30"],
            svec!["2024-01-01T01:30:00Z", "40"],
            svec!["2024-01-01T02:00:00Z", "50"], // Third hour with single value
        ],
    );

    let mut cmd = wrk.command("sample");
    cmd.args(["--timeseries", "timestamp"])
        .args(["--ts-interval", "1h"])
        .args(["--ts-aggregate", "median"])
        .arg("1")
        .arg("in.csv");

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    assert_eq!(got.len(), 4); // Header + 3 aggregated records
    assert_eq!(got[0], svec!["timestamp", "value"]);
    // First hour: median of 10 and 20 = 15
    let first_value: f64 = got[1][1].parse().unwrap_or(0.0);
    assert!(
        (first_value - 15.0).abs() < 0.01,
        "Expected median ~15.0, got {}",
        first_value
    );
    // Second hour: median of 30 and 40 = 35
    let second_value: f64 = got[2][1].parse().unwrap_or(0.0);
    assert!(
        (second_value - 35.0).abs() < 0.01,
        "Expected median ~35.0, got {}",
        second_value
    );
    // Third hour: median of single value 50 = 50
    let third_value: f64 = got[3][1].parse().unwrap_or(0.0);
    assert!(
        (third_value - 50.0).abs() < 0.01,
        "Expected median ~50.0, got {}",
        third_value
    );
}

#[test]
fn sample_timeseries_adaptive_both() {
    let wrk = Workdir::new("sample_timeseries_adaptive_both");
    // Create data with business hours, weekends, and other times
    wrk.create(
        "in.csv",
        vec![
            svec!["timestamp", "value"],
            svec!["2024-01-06T08:00:00Z", "10"], // Saturday 8am (weekend but not business hours)
            svec!["2024-01-06T10:00:00Z", "20"], // Saturday 10am (weekend)
            svec!["2024-01-08T08:00:00Z", "30"], /* Monday 8am (business day but before business
                                                  * hours) */
            svec!["2024-01-08T10:00:00Z", "40"], // Monday 10am (business hours + business day)
            svec!["2024-01-08T18:00:00Z", "50"], /* Monday 6pm (business day but after business
                                                  * hours) */
        ],
    );

    let mut cmd = wrk.command("sample");
    cmd.args(["--timeseries", "timestamp"])
        .args(["--ts-interval", "1d"])
        .args(["--ts-adaptive", "both"])
        .arg("1")
        .arg("in.csv");

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    // Should prefer business hours on weekdays OR weekends
    assert_eq!(got.len(), 3); // Header + 2 records (one per day)
    assert_eq!(got[0], svec!["timestamp", "value"]);
    // For Saturday, should prefer weekend record (both 8am and 10am are weekends, so either is
    // valid)
    let sat_record_8am = got.iter().find(|r| r[0] == "2024-01-06T08:00:00Z");
    let sat_record_10am = got.iter().find(|r| r[0] == "2024-01-06T10:00:00Z");
    assert!(
        sat_record_8am.is_some() || sat_record_10am.is_some(),
        "Should prefer weekend record for Saturday (8am or 10am), got: {:?}",
        got.iter().map(|r| &r[0]).collect::<Vec<_>>()
    );
    // For Monday, should prefer business hours record (10:00)
    let mon_record = got.iter().find(|r| r[0] == "2024-01-08T10:00:00Z");
    assert!(
        mon_record.is_some(),
        "Should prefer business hours record for Monday"
    );
}

// =============================================================================
// Streaming Bernoulli sampling tests (regression coverage for the URL path)
//
// These spin up a local actix-web server with hand-built fixture bytes and
// exercise the boundary detection added in PR #3774:
//   * RFC-4180 quoted newlines in the header are not split incorrectly.
//   * --max-size truncation drops any partial trailing record.
//   * Non-2xx HTTP status fails fast.
//   * --delimiter is honored on the streaming path.
// =============================================================================

use std::{net::SocketAddr, sync::mpsc, thread};

use actix_web::{App, HttpResponse, HttpServer, dev::ServerHandle, rt, web};
use serial_test::serial;

// Bind to 127.0.0.1 with an OS-assigned ephemeral port (port 0). Hardcoded
// ports collide on macOS CI runners (TIME_WAIT after teardown, peer
// integration-test binaries holding the port) and produce flaky
// "Address already in use" failures in the sample suite. The actual bound
// SocketAddr is captured via HttpServer::addrs() and threaded back to each
// test through SampleWebServer so URLs can be built against it.
const SAMPLE_TEST_BIND_HOST: &str = "127.0.0.1";

// Header field 0 contains a quoted newline (per RFC 4180). The OLD streaming
// header parser split on the first raw `\n`, which would land INSIDE the
// quote and corrupt every subsequent record.
const QUOTED_NL_HEADER_CSV: &str = "\"first\nline\",second,third\n1,2,3\n4,5,6\n7,8,9\n";

// Tab-delimited fixture for the --delimiter test. With the default-comma
// parser this would parse as a single-column CSV.
const TSV_BODY: &str = "id\tname\tcity\n1\talice\tparis\n2\tbob\tlondon\n3\tcarol\trome\n";

// Builds a CSV just over 1 MiB with fixed-size 100-byte records so the
// --max-size 1 cap (= 1_048_576 bytes) lands deterministically inside a
// record. Header (`id,payload\n`) is 11 bytes; each data record is exactly
// 100 bytes (`NNNNN,` + 93 'X' + `\n`). Total ≈ 1.2 MiB.
fn large_csv_body() -> String {
    let mut body = String::with_capacity(1_200_100);
    body.push_str("id,payload\n");
    let payload: String = "X".repeat(93);
    for i in 1..=12_000u32 {
        body.push_str(&format!("{i:05},{payload}\n"));
    }
    body
}

async fn serve_quoted_nl_header() -> HttpResponse {
    HttpResponse::Ok()
        .content_type("text/csv")
        .body(QUOTED_NL_HEADER_CSV)
}

async fn serve_tsv() -> HttpResponse {
    HttpResponse::Ok()
        .content_type("text/tab-separated-values")
        .body(TSV_BODY)
}

async fn serve_large_csv() -> HttpResponse {
    HttpResponse::Ok()
        .content_type("text/csv")
        .body(large_csv_body())
}

// Channel payload: Ok((handle, addr)) on a successful bind (addr carries the
// OS-assigned ephemeral port), Err(msg) if bind() failed. Sending Result
// instead of just the handle means a failed bind surfaces a real error to
// start() instead of leaving the receiver to time out.
async fn run_sample_webserver(
    tx: mpsc::Sender<Result<(ServerHandle, SocketAddr), String>>,
) -> std::io::Result<()> {
    let server_builder = HttpServer::new(|| {
        App::new()
            .service(web::resource("/quoted_nl_header.csv").to(serve_quoted_nl_header))
            .service(web::resource("/data.tsv").to(serve_tsv))
            .service(web::resource("/large.csv").to(serve_large_csv))
        // anything else -> 404 (handled by actix-web default)
    });

    // Port 0 -> OS picks an unused ephemeral port; addrs() then reports the
    // actual SocketAddr we landed on.
    let bound = match server_builder.bind((SAMPLE_TEST_BIND_HOST, 0)) {
        Ok(b) => b,
        Err(e) => {
            let _ = tx.send(Err(format!("bind failed: {e}")));
            return Err(e);
        },
    };

    let addr = match bound.addrs().into_iter().next() {
        Some(a) => a,
        None => {
            let _ = tx.send(Err("bind succeeded but no address was reported".to_string()));
            return Err(std::io::Error::other(
                "actix HttpServer::addrs() returned empty",
            ));
        },
    };

    let server = bound.run();
    let _ = tx.send(Ok((server.handle(), addr)));
    server.await
}

// RAII guard so the server is torn down even if the test panics in the middle
// (e.g. read_stdout fails). Without this, a panicking test would leave the
// port bound and cascade into a "Address already in use" on the next #[serial]
// test, producing confusing follow-on failures.
struct SampleWebServer {
    handle: Option<ServerHandle>,
    addr:   SocketAddr,
}

impl SampleWebServer {
    fn start() -> Self {
        let (tx, rx) = mpsc::channel();
        thread::spawn(move || {
            let server_future = run_sample_webserver(tx);
            rt::System::new().block_on(server_future)
        });

        // recv_timeout (rather than recv) so a failed bind that the server
        // thread can't surface in time doesn't hang the test forever.
        match rx.recv_timeout(std::time::Duration::from_secs(10)) {
            Ok(Ok((handle, addr))) => Self {
                handle: Some(handle),
                addr,
            },
            Ok(Err(msg)) => panic!("test webserver failed to bind: {msg}"),
            Err(e) => {
                panic!("test webserver did not start within 10s ({e:?})")
            },
        }
    }

    /// Build a URL against the actual bound (ephemeral) port. The leading
    /// slash on the path is optional — both `"foo.csv"` and `"/foo.csv"`
    /// produce the same URL.
    fn url(&self, path: &str) -> String {
        let path = path.strip_prefix('/').unwrap_or(path);
        format!("http://{}/{path}", self.addr)
    }
}

impl Drop for SampleWebServer {
    fn drop(&mut self) {
        if let Some(handle) = self.handle.take() {
            // Non-graceful stop (graceful = false) so an abandoned mid-response
            // connection — e.g. the --max-size test that stops reading early —
            // can't wedge teardown waiting for in-flight requests to drain.
            //
            // Belt-and-suspenders: run the stop on a detached thread and only
            // wait a bounded time for it. On a slow/contended runner (this hung
            // the Windows ARM64 CI to the 6-hour ceiling) a wedged shutdown must
            // never block the test thread forever; if it doesn't finish in time
            // we leak the ephemeral-port server and move on — the next #[serial]
            // test simply binds a fresh OS-assigned port.
            let (done_tx, done_rx) = mpsc::channel();
            thread::spawn(move || {
                rt::System::new().block_on(handle.stop(false));
                let _ = done_tx.send(());
            });
            let _ = done_rx.recv_timeout(std::time::Duration::from_secs(10));
        }
    }
}

// Run the command exactly once, assert it succeeded, and return the captured
// Output so the caller can parse stdout from the same execution. The previous
// `wrk.assert_success(&mut cmd)` + `wrk.read_stdout(&mut cmd)` pattern ran the
// process twice — doubling fixture-server requests and meaning the parsed
// stdout was from a second run, not the one whose status we asserted.
fn run_and_assert_success(cmd: &mut std::process::Command) -> std::process::Output {
    let output = cmd.output().expect("failed to execute sample command");
    assert!(
        output.status.success(),
        "sample command failed (status {}):\n--- stderr ---\n{}",
        output.status,
        String::from_utf8_lossy(&output.stderr)
    );
    output
}

// Parse a captured stdout buffer as CSV (default comma delimiter), preserving
// the same Vec<Vec<String>> shape that wrk.read_stdout would have produced.
fn parse_csv_stdout(stdout: &[u8]) -> Vec<Vec<String>> {
    let mut rdr = csv::ReaderBuilder::new()
        .has_headers(false)
        .flexible(true)
        .from_reader(stdout);
    rdr.records()
        .map(|r| {
            r.expect("sample stdout was not valid CSV")
                .iter()
                .map(std::string::ToString::to_string)
                .collect()
        })
        .collect()
}

#[test]
#[serial]
fn sample_bernoulli_url_quoted_newline_header() {
    // RAII guard: server is torn down even if the test panics below.
    let server = SampleWebServer::start();

    let wrk = Workdir::new("sample_bernoulli_url_quoted_nl");
    let mut cmd = wrk.command("sample");
    cmd.args(["--bernoulli", "--seed", "42"])
        .arg("0.999")
        .arg(server.url("quoted_nl_header.csv"));

    // Run once, assert success, and parse stdout from the same execution
    // — surfaces qsv's stderr on failure and doesn't double-hit the fixture.
    let output = run_and_assert_success(&mut cmd);
    let got = parse_csv_stdout(&output.stdout);

    // Header must come through INTACT — three fields, with the embedded
    // newline preserved in field 0. The old buggy splitter would have
    // produced one or two fields and shifted every following row.
    assert!(
        !got.is_empty(),
        "no rows emitted (header should always be present)"
    );
    let header = &got[0];
    assert_eq!(
        header.len(),
        3,
        "header should have 3 fields, got {header:?}"
    );
    assert_eq!(header[0], "first\nline");
    assert_eq!(header[1], "second");
    assert_eq!(header[2], "third");

    // Every data row must also have 3 fields. (Bernoulli at 0.999 with
    // seed 42 over only three rows may or may not include all of them —
    // we don't assert on count, only on shape.)
    for row in &got[1..] {
        assert_eq!(row.len(), 3, "data row should have 3 fields, got {row:?}");
    }
}

#[test]
#[serial]
fn sample_bernoulli_url_max_size_truncation() {
    let server = SampleWebServer::start();

    let wrk = Workdir::new("sample_bernoulli_url_max_size");
    let mut cmd = wrk.command("sample");
    cmd.args(["--bernoulli", "--seed", "42"])
        .args(["--max-size", "1"])
        .arg("0.5")
        .arg(server.url("large.csv"));

    // Single run — the previous double-run pattern doubled the ~1.2 MiB
    // download AND meant the stdout we parsed was from a different execution
    // than the one whose status we'd asserted.
    let output = run_and_assert_success(&mut cmd);
    let got = parse_csv_stdout(&output.stdout);

    // The cap is 1 MiB = 1_048_576 bytes. Header is 11 bytes; each data
    // record is 100 bytes. So records 1..=10485 fully fit (ending at byte
    // 11 + 10485*100 = 1_048_511, well under the cap), but record 10486
    // would extend to byte 1_048_611 — past the cap. The streaming code
    // must NOT emit that partial 10486 record.
    assert!(!got.is_empty(), "no rows emitted");
    let header = &got[0];
    assert_eq!(header, &vec!["id".to_string(), "payload".to_string()]);

    let max_id: u32 = got[1..]
        .iter()
        .map(|r| {
            r[0].parse::<u32>()
                .unwrap_or_else(|_| panic!("bad id row: {r:?}"))
        })
        .max()
        .expect("at least one data row should pass Bernoulli with seed 42 + p=0.5");

    assert!(
        max_id <= 10485,
        "saw id {max_id} which is past the --max-size 1 MiB cap (last full record is 10485)"
    );

    // Every emitted record must be well-formed: 2 fields, 5-digit id,
    // 93-char payload of 'X'. This catches half-records being flushed at
    // the cap boundary.
    for row in &got[1..] {
        assert_eq!(row.len(), 2, "data row should have 2 fields: {row:?}");
        assert_eq!(row[0].len(), 5, "id should be 5 chars: {row:?}");
        assert_eq!(row[1].len(), 93, "payload should be 93 chars: {row:?}");
        assert!(
            row[1].chars().all(|c| c == 'X'),
            "payload should be all 'X': {row:?}"
        );
    }
}

#[test]
#[serial]
fn sample_bernoulli_url_404_fails_fast() {
    let server = SampleWebServer::start();

    let wrk = Workdir::new("sample_bernoulli_url_404");
    let mut cmd = wrk.command("sample");
    cmd.args(["--bernoulli", "--seed", "42"])
        .arg("0.5")
        .arg(server.url("does_not_exist.csv"));

    // .error_for_status() should turn the 404 into an Err in the streaming
    // path, so qsv exits with non-zero. Without that, the HTML 404 body
    // would be fed straight into the csv parser.
    wrk.assert_err(&mut cmd);
}

#[test]
#[serial]
fn sample_bernoulli_url_custom_delimiter() {
    let server = SampleWebServer::start();

    let wrk = Workdir::new("sample_bernoulli_url_tsv");
    let mut cmd = wrk.command("sample");
    cmd.args(["--bernoulli", "--seed", "42"])
        .args(["--delimiter", "\t"])
        .arg("0.999")
        .arg(server.url("data.tsv"));

    // qsv's writer also honors --delimiter, so output is tab-separated.
    // read_stdout()'s CSV parser would treat that as a single comma-field, so
    // we parse the raw stdout with tab delimiter ourselves. Run once.
    let output = run_and_assert_success(&mut cmd);
    let stdout = String::from_utf8(output.stdout).expect("sample stdout must be valid UTF-8");

    let got: Vec<Vec<&str>> = stdout.lines().map(|l| l.split('\t').collect()).collect();

    // With --delimiter '\t' honored on the streaming path, fields split into
    // 3 columns. Without the fix the streaming parser would treat the whole
    // row as one comma-field and the writer would emit a single-column CSV.
    assert!(!got.is_empty(), "no rows emitted");
    assert_eq!(
        got[0],
        vec!["id", "name", "city"],
        "TSV header should split into 3 fields when --delimiter '\\t' is honored"
    );
    for row in &got[1..] {
        assert_eq!(row.len(), 3, "TSV data row should have 3 fields: {row:?}");
    }
}

// ---------------------------------------------------------------------------
// DataSketches Sampling-family tests: --varopt, --mergeable-reservoir, and
// the --sketch-out / --sketch-in serialization round-trip.
// ---------------------------------------------------------------------------

fn sketch_test_input() -> Vec<Vec<String>> {
    vec![
        svec!["ID", "Weight"],
        svec!["1", "10"],
        svec!["2", "20"],
        svec!["3", "30"],
        svec!["4", "40"],
        svec!["5", "50"],
        svec!["6", "60"],
        svec!["7", "70"],
        svec!["8", "80"],
        svec!["9", "90"],
        svec!["10", "100"],
    ]
}

#[test]
fn sample_varopt_seeded() {
    // Pins the exact rows VarOpt emits for the (fixture, --seed 42, k=4)
    // tuple. Intentionally exact: this is the regression net that catches
    // accidental changes in how the sampler consumes the RNG (e.g.,
    // reordering an `rng.random::<f64>()` call inside `update`). The
    // `sample_varopt_heavy_weight_dominates` test below carries the
    // semantic invariant (heavy-weight retention); this one nails down
    // the implementation. If the algorithm legitimately changes its RNG
    // consumption, regenerate the expected vector and document why.
    let wrk = Workdir::new("sample_varopt_seeded");
    wrk.create("in.csv", sketch_test_input());

    let mut cmd = wrk.command("sample");
    cmd.args(["--varopt", "Weight"])
        .args(["--seed", "42"])
        .arg("4")
        .arg("in.csv");

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    // VarOpt with this fixture and seed retains a heavy-weight-leaning sample.
    let expected = vec![
        svec!["ID", "Weight"],
        svec!["3", "30"],
        svec!["6", "60"],
        svec!["7", "70"],
        svec!["8", "80"],
    ];
    assert_eq!(got, expected);
}

#[test]
fn sample_varopt_heavy_weight_dominates() {
    // VarOpt should overwhelmingly retain a single very-heavy item among
    // light-weight peers.
    let wrk = Workdir::new("sample_varopt_heavy");
    let mut rows: Vec<Vec<String>> = vec![svec!["ID", "Weight"]];
    for i in 0..200 {
        rows.push(vec![i.to_string(), "1.0".to_string()]);
    }
    rows.push(svec!["9999", "1000000000.0"]);
    wrk.create("in.csv", rows);

    let mut cmd = wrk.command("sample");
    cmd.args(["--varopt", "Weight"])
        .args(["--seed", "7"])
        .arg("5")
        .arg("in.csv");

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    // First row is the header.
    let ids: Vec<&str> = got.iter().skip(1).map(|r| r[0].as_str()).collect();
    assert!(
        ids.contains(&"9999"),
        "heavy-weight item must be retained; got {ids:?}"
    );
    assert_eq!(got.len(), 6, "expected header + 5 sampled rows");
}

#[test]
fn sample_mergeable_reservoir_seeded() {
    // Pins the exact rows Algorithm R emits for (fixture, --seed 42, k=4).
    // Same intentional-pin rationale as `sample_varopt_seeded` above: a
    // refactor that changes how `ReservoirItemsSketch::update` draws from
    // the RNG should require regenerating these expected values, not
    // silently producing a different sample.
    let wrk = Workdir::new("sample_mergeable_reservoir");
    wrk.create("in.csv", sketch_test_input());

    let mut cmd = wrk.command("sample");
    cmd.arg("--mergeable-reservoir")
        .args(["--seed", "42"])
        .arg("4")
        .arg("in.csv");

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let expected = vec![
        svec!["ID", "Weight"],
        svec!["9", "90"],
        svec!["2", "20"],
        svec!["5", "50"],
        svec!["8", "80"],
    ];
    assert_eq!(got, expected);
}

#[test]
fn sample_sketch_out_in_round_trip() {
    let wrk = Workdir::new("sample_sketch_round_trip");
    wrk.create("in.csv", sketch_test_input());

    let sketch_path = wrk.path("snapshot.sk");
    let mut cmd_produce = wrk.command("sample");
    cmd_produce
        .arg("--mergeable-reservoir")
        .args(["--sketch-out", sketch_path.to_str().unwrap()])
        .args(["--seed", "42"])
        .arg("4")
        .arg("in.csv");
    let initial: Vec<Vec<String>> = wrk.read_stdout(&mut cmd_produce);

    let mut cmd_consume = wrk.command("sample");
    cmd_consume
        .args(["--sketch-in", sketch_path.to_str().unwrap()])
        .args(["--seed", "1"])
        .arg("99");
    let merged: Vec<Vec<String>> = wrk.read_stdout(&mut cmd_consume);

    // Round-trip from a single-sketch sketch-in should emit the exact same
    // sample (and header) as the producing call.
    assert_eq!(merged, initial);
}

#[test]
fn sample_sketch_in_merge_two_sketches() {
    let wrk = Workdir::new("sample_sketch_merge_two");
    wrk.create("in.csv", sketch_test_input());

    let a = wrk.path("a.sk");
    let b = wrk.path("b.sk");

    let mut cmd_a = wrk.command("sample");
    cmd_a
        .arg("--mergeable-reservoir")
        .args(["--sketch-out", a.to_str().unwrap()])
        .args(["--seed", "1"])
        .arg("3")
        .arg("in.csv");
    let _: Vec<Vec<String>> = wrk.read_stdout(&mut cmd_a);

    let mut cmd_b = wrk.command("sample");
    cmd_b
        .arg("--mergeable-reservoir")
        .args(["--sketch-out", b.to_str().unwrap()])
        .args(["--seed", "99"])
        .arg("3")
        .arg("in.csv");
    let _: Vec<Vec<String>> = wrk.read_stdout(&mut cmd_b);

    let combined = format!("{},{}", a.to_str().unwrap(), b.to_str().unwrap());
    let mut cmd_merge = wrk.command("sample");
    cmd_merge
        .args(["--sketch-in", &combined])
        .args(["--seed", "5"])
        .arg("99");
    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd_merge);

    // Header preserved, sample size capped at the underlying sketches' k=3.
    assert_eq!(got[0], svec!["ID", "Weight"]);
    assert!(
        got.len() >= 2 && got.len() <= 4,
        "merged sample should have 1 header + 1..3 rows; got {} rows",
        got.len()
    );
    // Every retained row must come from the input domain.
    let valid_ids: std::collections::HashSet<&str> =
        ["1", "2", "3", "4", "5", "6", "7", "8", "9", "10"]
            .into_iter()
            .collect();
    for row in &got[1..] {
        assert!(
            valid_ids.contains(row[0].as_str()),
            "unexpected ID in merge output: {row:?}"
        );
    }
}

#[test]
fn sample_sketch_in_family_mismatch_errors() {
    let wrk = Workdir::new("sample_sketch_family_mismatch");
    wrk.create("in.csv", sketch_test_input());

    let reservoir = wrk.path("r.sk");
    let varopt = wrk.path("v.sk");

    let mut c1 = wrk.command("sample");
    c1.arg("--mergeable-reservoir")
        .args(["--sketch-out", reservoir.to_str().unwrap()])
        .arg("3")
        .arg("in.csv");
    let _: Vec<Vec<String>> = wrk.read_stdout(&mut c1);

    let mut c2 = wrk.command("sample");
    c2.args(["--varopt", "Weight"])
        .args(["--sketch-out", varopt.to_str().unwrap()])
        .arg("3")
        .arg("in.csv");
    let _: Vec<Vec<String>> = wrk.read_stdout(&mut c2);

    let combined = format!(
        "{},{}",
        reservoir.to_str().unwrap(),
        varopt.to_str().unwrap()
    );
    let mut c_merge = wrk.command("sample");
    c_merge.args(["--sketch-in", &combined]).arg("99");
    let out = c_merge.output().expect("spawn");
    assert!(
        !out.status.success(),
        "mixing Reservoir + VarOpt sketches should fail"
    );
    let stderr = String::from_utf8_lossy(&out.stderr);
    assert!(
        stderr.contains("family mismatch") || stderr.contains("Sketch family mismatch"),
        "expected family-mismatch error, got stderr: {stderr}"
    );
}

#[test]
fn sample_varopt_and_weighted_are_mutually_exclusive() {
    let wrk = Workdir::new("sample_varopt_and_weighted");
    wrk.create("in.csv", sketch_test_input());

    let mut cmd = wrk.command("sample");
    cmd.args(["--weighted", "Weight"])
        .args(["--varopt", "Weight"])
        .arg("4")
        .arg("in.csv");
    let out = cmd.output().expect("spawn");
    assert!(!out.status.success());
    let stderr = String::from_utf8_lossy(&out.stderr);
    assert!(
        stderr.contains("Only one sampling method"),
        "expected method-collision error, got: {stderr}"
    );
}
