use std::{borrow::ToOwned, process};

use crate::workdir::Workdir;

macro_rules! slice_tests {
    ($name:ident, $start:expr_2021, $end:expr_2021, $expected:expr_2021) => {
        mod $name {
            use super::test_slice;

            #[test]
            fn headers_no_index() {
                let name = concat!(stringify!($name), "headers_no_index");
                test_slice(name, $start, $end, $expected, true, false, false, false);
            }

            #[test]
            fn no_headers_no_index() {
                let name = concat!(stringify!($name), "no_headers_no_index");
                test_slice(name, $start, $end, $expected, false, false, false, false);
            }

            #[test]
            fn no_headers_no_index_json() {
                let name = concat!(stringify!($name), "no_headers_no_index_json");
                test_slice(name, $start, $end, $expected, false, false, false, true);
            }

            #[test]
            fn headers_index() {
                let name = concat!(stringify!($name), "headers_index");
                test_slice(name, $start, $end, $expected, true, true, false, false);
            }

            #[test]
            fn no_headers_index() {
                let name = concat!(stringify!($name), "no_headers_index");
                test_slice(name, $start, $end, $expected, false, true, false, false);
            }

            #[test]
            fn headers_index_json() {
                let name = concat!(stringify!($name), "headers_index_json");
                test_slice(name, $start, $end, $expected, true, true, false, true);
            }

            #[test]
            fn no_headers_index_json() {
                let name = concat!(stringify!($name), "no_headers_index_json");
                test_slice(name, $start, $end, $expected, false, true, false, true);
            }

            #[test]
            fn headers_no_index_len() {
                let name = concat!(stringify!($name), "headers_no_index_len");
                test_slice(name, $start, $end, $expected, true, false, true, false);
            }

            #[test]
            fn no_headers_no_index_len() {
                let name = concat!(stringify!($name), "no_headers_no_index_len");
                test_slice(name, $start, $end, $expected, false, false, true, false);
            }

            #[test]
            fn headers_no_index_len_json() {
                let name = concat!(stringify!($name), "headers_no_index_len_json");
                test_slice(name, $start, $end, $expected, true, false, true, true);
            }

            #[test]
            fn no_headers_no_index_len_json() {
                let name = concat!(stringify!($name), "no_headers_no_index_len_json");
                test_slice(name, $start, $end, $expected, false, false, true, true);
            }

            #[test]
            fn headers_index_len() {
                let name = concat!(stringify!($name), "headers_index_len");
                test_slice(name, $start, $end, $expected, true, true, true, false);
            }

            #[test]
            fn no_headers_index_len() {
                let name = concat!(stringify!($name), "no_headers_index_len");
                test_slice(name, $start, $end, $expected, false, true, true, false);
            }

            #[test]
            fn headers_index_len_json() {
                let name = concat!(stringify!($name), "headers_index_len_json");
                test_slice(name, $start, $end, $expected, true, true, true, true);
            }

            #[test]
            fn no_headers_index_len_json() {
                let name = concat!(stringify!($name), "no_headers_index_len_json");
                test_slice(name, $start, $end, $expected, false, true, true, true);
            }
        }
    };
}

fn setup(name: &str, headers: bool, use_index: bool) -> (Workdir, process::Command) {
    let wrk = Workdir::new(name);
    let mut data = vec![svec!["a"], svec!["b"], svec!["c"], svec!["d"], svec!["e"]];
    if headers {
        data.insert(0, svec!["header"]);
    }
    if use_index {
        wrk.create_indexed("in.csv", data);
    } else {
        wrk.create("in.csv", data);
    }

    let mut cmd = wrk.command("slice");
    cmd.arg("in.csv");

    (wrk, cmd)
}

fn test_slice(
    name: &str,
    start: Option<isize>,
    end: Option<usize>,
    expected: &[&str],
    headers: bool,
    use_index: bool,
    as_len: bool,
    json_output: bool,
) {
    let (wrk, mut cmd) = setup(name, headers, use_index);
    if let Some(start) = start {
        cmd.arg("--start").arg(start.to_string());
    }
    if let Some(end) = end {
        if as_len {
            let start = start.unwrap_or(0);
            if start < 0 {
                cmd.arg("--len").arg(end.to_string());
            } else {
                cmd.arg("--len")
                    .arg((end - start.unsigned_abs()).to_string());
            }
        } else {
            cmd.arg("--end").arg(end.to_string());
        }
    }
    if !headers {
        cmd.arg("--no-headers");
    }
    if json_output {
        let output_file = wrk.path("output.json").to_string_lossy().to_string();

        cmd.arg("--json").args(["--output", &output_file]);

        wrk.assert_success(&mut cmd);

        let gots = wrk.read_to_string(&output_file).unwrap();
        let gotj: serde_json::Value = serde_json::from_str(&gots).unwrap();
        let got = gotj.to_string();

        let expected_vec = expected
            .iter()
            .map(|&s| {
                if headers {
                    format!("{{\"header\":\"{}\"}}", s)
                } else {
                    format!("{{\"0\":\"{}\"}}", s)
                }
            })
            .collect::<Vec<String>>();
        let expected = format!("[{}]", expected_vec.join(","));

        assert_eq!(got, expected);
    } else {
        let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
        let mut expected = expected
            .iter()
            .map(|&s| vec![s.to_owned()])
            .collect::<Vec<Vec<String>>>();
        if headers {
            expected.insert(0, svec!["header"]);
        }
        assert_eq!(got, expected);
    }
}

fn test_index(name: &str, idx: isize, expected: &str, headers: bool, use_index: bool) {
    let (wrk, mut cmd) = setup(name, headers, use_index);
    cmd.arg("--index").arg(idx.to_string());
    if !headers {
        cmd.arg("--no-headers");
    }

    let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
    let mut expected = vec![vec![expected.to_owned()]];
    if headers {
        expected.insert(0, svec!["header"]);
    }
    assert_eq!(got, expected);
}

slice_tests!(slice_simple, Some(0), Some(1), &["a"]);
slice_tests!(slice_simple_2, Some(1), Some(3), &["b", "c"]);
slice_tests!(slice_no_start, None, Some(1), &["a"]);
slice_tests!(slice_no_end, Some(3), None, &["d", "e"]);
slice_tests!(slice_all, None, None, &["a", "b", "c", "d", "e"]);
slice_tests!(slice_negative_start, Some(-2), None, &["d", "e"]);

#[test]
fn slice_negative_with_len() {
    test_slice(
        "slice_negative_start_headers_index_len",
        Some(-4),
        Some(2),
        &["b", "c"],
        true,
        true,
        true,
        false,
    );
    test_slice(
        "slice_negative_start_no_headers_index_len",
        Some(-4),
        Some(2),
        &["b", "c"],
        false,
        true,
        true,
        false,
    );
    test_slice(
        "slice_negative_start_headers_no_index_len",
        Some(-4),
        Some(2),
        &["b", "c"],
        true,
        false,
        true,
        false,
    );
}

#[test]
fn slice_negative_with_len_json() {
    test_slice(
        "slice_negative_start_headers_index_len_json",
        Some(-4),
        Some(2),
        &["b", "c"],
        true,
        true,
        true,
        true,
    );
    test_slice(
        "slice_negative_start_no_headers_index_len_json",
        Some(-4),
        Some(2),
        &["b", "c"],
        false,
        true,
        true,
        true,
    );
    test_slice(
        "slice_negative_start_headers_no_index_len_json",
        Some(-4),
        Some(2),
        &["b", "c"],
        true,
        false,
        true,
        true,
    );
}

#[test]
fn slice_json_escapes_keys() {
    // regression: JSON object keys were interpolated raw, so a header
    // containing a double quote or backslash emitted invalid JSON.
    let wrk = Workdir::new("slice_json_escapes_keys");
    wrk.create(
        "data.csv",
        vec![
            svec![r#"he said "hi""#, r"back\slash"],
            svec!["a", "b"],
            svec!["c", "d"],
        ],
    );
    let mut cmd = wrk.command("slice");
    cmd.arg("data.csv").args(["--start", "0"]).arg("--json");

    let got: String = wrk.stdout(&mut cmd);
    let parsed: serde_json::Value = serde_json::from_str(&got)
        .unwrap_or_else(|e| panic!("output is not valid JSON: {e}\ngot: {got}"));
    let obj = &parsed[0];
    assert_eq!(
        obj[r#"he said "hi""#],
        serde_json::Value::String("a".to_string())
    );
    assert_eq!(
        obj[r"back\slash"],
        serde_json::Value::String("b".to_string())
    );
    wrk.assert_success(&mut cmd);
}

#[test]
fn slice_index() {
    test_index("slice_index", 1, "b", true, false);
}
#[test]
fn slice_index_no_headers() {
    test_index("slice_index_no_headers", 1, "b", false, false);
}
#[test]
fn slice_index_withindex() {
    test_index("slice_index_withindex", 1, "b", true, true);
}
#[test]
fn slice_index_no_headers_withindex() {
    test_index("slice_index_no_headers_withindex", 1, "b", false, true);
}

#[test]
fn slice_neg_index() {
    test_index("slice_neg_index", -1, "e", true, false);
}
#[test]
fn slice_neg_index_no_headers() {
    test_index("slice_neg_index_no_headers", -1, "e", false, false);
}
#[test]
fn slice_neg_index_withindex() {
    test_index("slice_neg_index_withindex", -2, "d", true, true);
}
#[test]
fn slice_neg_index_no_headers_withindex() {
    test_index("slice_neg_index_no_headers_withindex", -2, "d", false, true);
}

fn test_slice_invert(
    name: &str,
    start: Option<isize>,
    end: Option<usize>,
    expected: &[&str],
    headers: bool,
    use_index: bool,
    as_len: bool,
    json_output: bool,
) {
    let (wrk, mut cmd) = setup(name, headers, use_index);
    if let Some(start) = start {
        cmd.arg("--start").arg(start.to_string());
    }
    if let Some(end) = end {
        if as_len {
            let start = start.unwrap_or(0);
            if start < 0 {
                cmd.arg("--len").arg(end.to_string());
            } else {
                cmd.arg("--len")
                    .arg((end - start.unsigned_abs()).to_string());
            }
        } else {
            cmd.arg("--end").arg(end.to_string());
        }
    }
    if !headers {
        cmd.arg("--no-headers");
    }
    cmd.arg("--invert");

    if json_output {
        let output_file = wrk.path("output.json").to_string_lossy().to_string();

        cmd.arg("--json").args(["--output", &output_file]);

        wrk.assert_success(&mut cmd);

        let gots = wrk.read_to_string(&output_file).unwrap();
        let gotj: serde_json::Value = serde_json::from_str(&gots).unwrap();
        let got = gotj.to_string();

        let expected_vec = expected
            .iter()
            .map(|&s| {
                if headers {
                    format!("{{\"header\":\"{}\"}}", s)
                } else {
                    format!("{{\"0\":\"{}\"}}", s)
                }
            })
            .collect::<Vec<String>>();
        let expected = format!("[{}]", expected_vec.join(","));

        assert_eq!(got, expected);
    } else {
        let got: Vec<Vec<String>> = wrk.read_stdout(&mut cmd);
        let mut expected = expected
            .iter()
            .map(|&s| vec![s.to_owned()])
            .collect::<Vec<Vec<String>>>();
        if headers {
            expected.insert(0, svec!["header"]);
        }
        assert_eq!(got, expected);
    }
}

#[test]
fn slice_invert_simple() {
    test_slice_invert(
        "slice_invert_simple",
        Some(0),
        Some(1),
        &["b", "c", "d", "e"],
        true,
        false,
        false,
        false,
    );
}

#[test]
fn slice_invert_middle() {
    test_slice_invert(
        "slice_invert_middle",
        Some(1),
        Some(3),
        &["a", "d", "e"],
        true,
        false,
        false,
        false,
    );
}

#[test]
fn slice_invert_with_index() {
    test_slice_invert(
        "slice_invert_with_index",
        Some(1),
        Some(3),
        &["a", "d", "e"],
        true,
        true,
        false,
        false,
    );
}

#[test]
fn slice_invert_json() {
    test_slice_invert(
        "slice_invert_json",
        Some(1),
        Some(3),
        &["a", "d", "e"],
        true,
        false,
        false,
        true,
    );
}

#[test]
fn slice_invert_negative() {
    test_slice_invert(
        "slice_invert_negative",
        Some(-2),
        None,
        &["a", "b", "c"],
        true,
        false,
        false,
        false,
    );
}

#[test]
fn slice_invert_with_len() {
    test_slice_invert(
        "slice_invert_with_len",
        Some(1),
        Some(2),
        &["a", "c", "d", "e"],
        true,
        false,
        true,
        false,
    );
}

// negative --start with no --end, indexed: previously errored at seek(usize::MAX)
#[test]
fn slice_invert_negative_with_index() {
    test_slice_invert(
        "slice_invert_negative_with_index",
        Some(-2),
        None,
        &["a", "b", "c"],
        true,
        true,
        false,
        false,
    );
}

// --start equals row count, indexed: previously underflowed total_rows-end and
// errored at seek(usize::MAX). Slice 5..end on a 5-row file is empty, so invert
// is the entire file.
#[test]
fn slice_invert_start_at_eof_with_index() {
    test_slice_invert(
        "slice_invert_start_at_eof_with_index",
        Some(5),
        None,
        &["a", "b", "c", "d", "e"],
        true,
        true,
        false,
        false,
    );
}

// --start past EOF, indexed: previously underflowed / errored.
#[test]
fn slice_invert_start_past_eof_with_index() {
    test_slice_invert(
        "slice_invert_start_past_eof_with_index",
        Some(100),
        None,
        &["a", "b", "c", "d", "e"],
        true,
        true,
        false,
        false,
    );
}

// --end past EOF, --invert, indexed: previously panicked on
// `total_rows - end` subtract overflow in the "after end" loop. Slice covers
// the whole file, so invert is empty.
#[test]
fn slice_invert_end_past_eof_with_index() {
    test_slice_invert(
        "slice_invert_end_past_eof_with_index",
        Some(0),
        Some(1000),
        &[],
        true,
        true,
        false,
        false,
    );
}

// JSON variant of the above (no-index): exercises the JSON record collector.
#[test]
fn slice_invert_end_past_eof_json() {
    test_slice_invert(
        "slice_invert_end_past_eof_json",
        Some(0),
        Some(1000),
        &[],
        true,
        false,
        false,
        true,
    );
}

// --end past EOF on the non-invert indexed path: should return all rows
// up to EOF rather than erroring at seek().
#[test]
fn slice_end_past_eof_with_index() {
    test_slice(
        "slice_end_past_eof_with_index",
        Some(0),
        Some(1000),
        &["a", "b", "c", "d", "e"],
        true,
        true,
        false,
        false,
    );
}

// |negative --start| larger than row count must clamp to 0
// (saturating_sub), not wrap to a position past the end of the file.
#[test]
fn slice_negative_start_clamps_to_zero() {
    test_slice(
        "slice_negative_start_clamps_to_zero",
        Some(-100),
        None,
        &["a", "b", "c", "d", "e"],
        true,
        false,
        false,
        false,
    );
    test_slice(
        "slice_negative_start_clamps_to_zero_with_index",
        Some(-100),
        None,
        &["a", "b", "c", "d", "e"],
        true,
        true,
        false,
        false,
    );
}

// Empty non-invert JSON slice must still emit `[]` so downstream parsers
// see a well-formed document, both with and without an index.
#[test]
fn slice_empty_json_no_index() {
    test_slice(
        "slice_empty_json_no_index",
        Some(5),
        Some(5),
        &[],
        true,
        false,
        false,
        true,
    );
}

#[test]
fn slice_empty_json_with_index() {
    test_slice(
        "slice_empty_json_with_index",
        Some(5),
        Some(5),
        &[],
        true,
        true,
        false,
        true,
    );
}

// Empty JSON slice with --no-headers + --json on the indexed path: confirms
// the empty-slice short-circuit forwards flag_no_headers to write_json
// the same way the non-empty path does.
#[test]
fn slice_empty_json_no_headers_with_index() {
    test_slice(
        "slice_empty_json_no_headers_with_index",
        Some(5),
        Some(5),
        &[],
        false,
        true,
        false,
        true,
    );
}

// Same clamping behavior for negative --index.
#[test]
fn slice_neg_index_clamps_to_zero() {
    test_index("slice_neg_index_clamps_to_zero", -100, "a", true, false);
    test_index(
        "slice_neg_index_clamps_to_zero_with_index",
        -100,
        "a",
        true,
        true,
    );
}

#[test]
#[cfg(feature = "polars")]
fn slice_from_parquet() {
    let wrk = Workdir::new("slice_from_parquet");
    let test_file = wrk.load_test_file("NYC311-5.parquet");
    let mut cmd = wrk.command("slice");
    cmd.arg(test_file).arg("--index").arg("2").arg("--json");

    wrk.assert_success(&mut cmd);

    let got: String = wrk.stdout(&mut cmd);
    let expected = r#"[{"Unique Key":"20520945","Created Date":"05/27/2011 12:00:00 AM","Closed Date":null,"Agency":"HPD","Agency Name":"Department of Housing Preservation and Development","Complaint Type":"PAINT - PLASTER","Descriptor":"WALLS","Location Type":"RESIDENTIAL BUILDING","Incident Zip":"11225","Incident Address":"1700 BEDFORD AVENUE","Street Name":"BEDFORD AVENUE","Cross Street 1":"MONTGOMERY STREET","Cross Street 2":"SULLIVAN PLACE","Intersection Street 1":null,"Intersection Street 2":null,"Address Type":"ADDRESS","City":"BROOKLYN","Landmark":null,"Facility Type":"N/A","Status":"Open","Due Date":null,"Resolution Description":"The following complaint conditions are still open.HPD may attempt to contact you to verify the correction of the condition or may conduct an inspection.","Resolution Action Updated Date":"06/15/2011 12:00:00 AM","Community Board":"09 BROOKLYN","BBL":"3013020001","Borough":"BROOKLYN","X Coordinate (State Plane)":"996197","Y Coordinate (State Plane)":"181752","Open Data Channel Type":"UNKNOWN","Park Facility Name":"Unspecified","Park Borough":"BROOKLYN","Vehicle Type":null,"Taxi Company Borough":null,"Taxi Pick Up Location":null,"Bridge Highway Name":null,"Bridge Highway Direction":null,"Road Ramp":null,"Bridge Highway Segment":null,"Latitude":null,"Longitude":null,"Location":null}]"#;
    assert_eq!(got, expected);
}

#[cfg(feature = "polars")]
#[test]
fn slice_from_json() {
    let wrk = Workdir::new("slice_from_json");
    let test_file = wrk.load_test_file("NYC311-5.json");
    let mut cmd = wrk.command("slice");
    cmd.arg(test_file).arg("--index").arg("2").arg("--json");

    wrk.assert_success(&mut cmd);

    let got: String = wrk.stdout(&mut cmd);
    let expected = r#""Unique Key":"20520945""#;
    assert!(got.contains(expected));
}

#[cfg(feature = "polars")]
#[test]
fn slice_from_jsonl() {
    let wrk = Workdir::new("slice_from_jsonl");
    let test_file = wrk.load_test_file("NYC311-5.jsonl");
    let mut cmd = wrk.command("slice");
    cmd.arg(test_file).arg("--index").arg("2").arg("--json");

    // Assert the full converted row, not a substring. The literal "Unique Key":"20520945"
    // is embedded in the raw JSONL bytes, so a substring check passes even when conversion
    // fails and the original bytes are read as-is. A full-object match can only pass on
    // genuine conversion (see issue #3988). Compare as parsed JSON because polars does not
    // preserve the source column order when converting JSONL; serde_json's (preserve_order)
    // Map is IndexMap-backed, whose PartialEq is order-independent.
    let got: serde_json::Value =
        serde_json::from_str(&wrk.stdout_on_success::<String>(&mut cmd)).unwrap();
    let expected: serde_json::Value = serde_json::from_str(
        r#"[{"Unique Key":"20520945","Created Date":"05/27/2011 12:00:00 AM","Closed Date":null,"Agency":"HPD","Agency Name":"Department of Housing Preservation and Development","Complaint Type":"PAINT - PLASTER","Descriptor":"WALLS","Location Type":"RESIDENTIAL BUILDING","Incident Zip":"11225","Incident Address":"1700 BEDFORD AVENUE","Street Name":"BEDFORD AVENUE","Cross Street 1":"MONTGOMERY STREET","Cross Street 2":"SULLIVAN PLACE","Intersection Street 1":null,"Intersection Street 2":null,"Address Type":"ADDRESS","City":"BROOKLYN","Landmark":null,"Facility Type":"N/A","Status":"Open","Due Date":null,"Resolution Description":"The following complaint conditions are still open.HPD may attempt to contact you to verify the correction of the condition or may conduct an inspection.","Resolution Action Updated Date":"06/15/2011 12:00:00 AM","Community Board":"09 BROOKLYN","BBL":"3013020001","Borough":"BROOKLYN","X Coordinate (State Plane)":"996197","Y Coordinate (State Plane)":"181752","Open Data Channel Type":"UNKNOWN","Park Facility Name":"Unspecified","Park Borough":"BROOKLYN","Vehicle Type":null,"Taxi Company Borough":null,"Taxi Pick Up Location":null,"Bridge Highway Name":null,"Bridge Highway Direction":null,"Road Ramp":null,"Bridge Highway Segment":null,"Latitude":null,"Longitude":null,"Location":null}]"#,
    )
    .unwrap();
    assert_eq!(got, expected);
}

#[cfg(feature = "polars")]
#[test]
fn slice_from_avro() {
    let wrk = Workdir::new("slice_from_avro");
    let test_file = wrk.load_test_file("NYC311-5.avro");
    let mut cmd = wrk.command("slice");
    cmd.arg(test_file).arg("--index").arg("2").arg("--json");

    // Assert the full converted row, not a substring. String values are stored verbatim
    // in the Avro binary, so a substring check passes even when conversion fails and the
    // raw bytes are read as-is. An exact match can only pass on genuine conversion
    // (see issue #3988).
    let got: String = wrk.stdout_on_success(&mut cmd);
    let expected = r#"[{"Unique Key":"20520945","Created Date":"05/27/2011 12:00:00 AM","Closed Date":null,"Agency":"HPD","Agency Name":"Department of Housing Preservation and Development","Complaint Type":"PAINT - PLASTER","Descriptor":"WALLS","Location Type":"RESIDENTIAL BUILDING","Incident Zip":"11225","Incident Address":"1700 BEDFORD AVENUE","Street Name":"BEDFORD AVENUE","Cross Street 1":"MONTGOMERY STREET","Cross Street 2":"SULLIVAN PLACE","Intersection Street 1":null,"Intersection Street 2":null,"Address Type":"ADDRESS","City":"BROOKLYN","Landmark":null,"Facility Type":"N/A","Status":"Open","Due Date":null,"Resolution Description":"The following complaint conditions are still open.HPD may attempt to contact you to verify the correction of the condition or may conduct an inspection.","Resolution Action Updated Date":"06/15/2011 12:00:00 AM","Community Board":"09 BROOKLYN","BBL":"3013020001","Borough":"BROOKLYN","X Coordinate (State Plane)":"996197","Y Coordinate (State Plane)":"181752","Open Data Channel Type":"UNKNOWN","Park Facility Name":"Unspecified","Park Borough":"BROOKLYN","Vehicle Type":null,"Taxi Company Borough":null,"Taxi Pick Up Location":null,"Bridge Highway Name":null,"Bridge Highway Direction":null,"Road Ramp":null,"Bridge Highway Segment":null,"Latitude":null,"Longitude":null,"Location":null}]"#;
    assert_eq!(got, expected);
}

#[cfg(feature = "polars")]
#[test]
fn slice_from_arrow() {
    let wrk = Workdir::new("slice_from_arrow");
    let test_file = wrk.load_test_file("NYC311-5.arrow");
    let mut cmd = wrk.command("slice");
    cmd.arg(test_file).arg("--index").arg("2").arg("--json");

    wrk.assert_success(&mut cmd);

    let got: String = wrk.stdout(&mut cmd);
    let expected = r#""Unique Key":"20520945""#;
    assert!(got.contains(expected));
}

#[cfg(feature = "polars")]
#[test]
fn slice_from_csvgz() {
    let wrk = Workdir::new("slice_from_csvgz");
    let test_file = wrk.load_test_file("NYC311-5.csv.gz");
    let mut cmd = wrk.command("slice");
    cmd.arg(test_file).arg("--index").arg("2").arg("--json");

    wrk.assert_success(&mut cmd);

    let got: String = wrk.stdout(&mut cmd);
    let expected = "20520945";
    assert!(got.contains(expected));
}

#[cfg(feature = "polars")]
#[test]
fn slice_from_csvzst() {
    let wrk = Workdir::new("slice_from_csvzst");
    let test_file = wrk.load_test_file("NYC311-5.csv.zst");
    let mut cmd = wrk.command("slice");
    cmd.arg(test_file).arg("--index").arg("2").arg("--json");

    wrk.assert_success(&mut cmd);

    let got: String = wrk.stdout(&mut cmd);
    let expected = "20520945";
    assert!(got.contains(expected));
}

#[cfg(feature = "polars")]
#[test]
fn slice_from_csvzlib() {
    let wrk = Workdir::new("slice_from_csvzlib");
    let test_file = wrk.load_test_file("NYC311-5.csv.zlib");
    let mut cmd = wrk.command("slice");
    cmd.arg(test_file).arg("--index").arg("2").arg("--json");

    wrk.assert_success(&mut cmd);

    let got: String = wrk.stdout(&mut cmd);
    let expected = "20520945";
    assert!(got.contains(expected));
}

#[cfg(feature = "polars")]
#[test]
fn slice_from_tsvgz() {
    let wrk = Workdir::new("slice_from_tsvgz");
    let test_file = wrk.load_test_file("NYC311-5.tsv.gz");
    let mut cmd = wrk.command("slice");
    cmd.arg(test_file).arg("--index").arg("2").arg("--json");

    wrk.assert_success(&mut cmd);

    let got: String = wrk.stdout(&mut cmd);
    let expected = "20520945";
    assert!(got.contains(expected));
}

#[cfg(feature = "polars")]
#[test]
fn slice_from_tsvzst() {
    let wrk = Workdir::new("slice_from_tsvzst");
    let test_file = wrk.load_test_file("NYC311-5.tsv.zst");
    let mut cmd = wrk.command("slice");
    cmd.arg(test_file).arg("--index").arg("2").arg("--json");

    wrk.assert_success(&mut cmd);

    let got: String = wrk.stdout(&mut cmd);
    let expected = "20520945";
    assert!(got.contains(expected));
}

#[cfg(feature = "polars")]
#[test]
fn slice_from_tsvzlib() {
    let wrk = Workdir::new("slice_from_tsvzlib");
    let test_file = wrk.load_test_file("NYC311-5.tsv.zlib");
    let mut cmd = wrk.command("slice");
    cmd.arg(test_file).arg("--index").arg("2").arg("--json");

    wrk.assert_success(&mut cmd);

    let got: String = wrk.stdout(&mut cmd);
    let expected = "20520945";
    assert!(got.contains(expected));
}

#[cfg(feature = "polars")]
#[test]
fn slice_from_csvzip() {
    // single-entry zip (boston311-100.csv.zip contains just boston311-100.csv)
    let wrk = Workdir::new("slice_from_csvzip");
    let test_file = wrk.load_test_file("boston311-100.csv.zip");
    let mut cmd = wrk.command("slice");
    cmd.arg(test_file).arg("--index").arg("0").arg("--json");

    wrk.assert_success(&mut cmd);

    let got: String = wrk.stdout(&mut cmd);
    let expected = "101004143000";
    assert!(got.contains(expected));
}

#[cfg(feature = "polars")]
#[test]
fn slice_from_zip_multientry() {
    // testzip.zip has multiple entries (positions.csv, NYC311-5.ssv, buses.csv,
    // NYC311-5.tsv) plus __MACOSX system files. `slice` extracts the zip via
    // `process_input`, which (like the reader-level `select_zip_entry` path)
    // skips the system files and returns tabular entries first, so the first
    // entry is the FIRST CSV/TSV/TAB/SSV entry in archive order: positions.csv
    // (headers: position,title). Issue #3988.
    let wrk = Workdir::new("slice_from_zip_multientry");
    let test_file = wrk.load_test_file("testzip.zip");
    let mut cmd = wrk.command("slice");
    cmd.arg(test_file).arg("--index").arg("0").arg("--json");

    wrk.assert_success(&mut cmd);

    let got: String = wrk.stdout(&mut cmd);
    assert!(got.contains("position"));
    assert!(got.contains("title"));
}

#[test]
fn slice_from_tsv() {
    let wrk = Workdir::new("slice_from_tsv");
    let test_file = wrk.load_test_file("NYC311-5.tsv");
    let mut cmd = wrk.command("slice");
    cmd.arg(test_file).arg("--index").arg("2").arg("--json");

    wrk.assert_success(&mut cmd);

    let got: String = wrk.stdout(&mut cmd);
    let expected = "20520945";
    assert!(got.contains(expected));
}

// do not run this test on Windows as it doesn't have gzip
#[cfg(all(feature = "polars", not(target_os = "windows")))]
#[test]
fn slice_float_precision() {
    let wrk = Workdir::new("slice_float_precision");

    // Create a test file with floating-point values
    let data = vec![
        svec!["id", "value1", "value2"],
        svec![
            "1",
            "3.1415926535897932384626433",
            "2.7182818284590452353602874"
        ],
        svec![
            "2",
            "1.4142135623730950488016887",
            "1.7320508075688772935274463"
        ],
        svec![
            "3",
            "0.5772156649015328606065120",
            "0.2617993877991494365385536"
        ],
    ];
    wrk.create("float_data.csv", data);
    let test_csv = wrk.path("float_data.csv");

    // Create a schema file for float_data.csv
    // set value to a Decimal with 20 decimal places
    let schema_data = r#"{
        "fields": {
            "id": "String",
            "value1": { "Decimal" : [25,20] },
            "value2": "Float64"
        },"metadata": null
    }"#;
    wrk.create_from_string("float_data.csv.pschema.json", schema_data);

    // Create a parquet file using sqlp command
    let mut cmd = wrk.command("sqlp");
    cmd.arg(&test_csv)
        .arg("select * from float_data")
        .arg("--cache-schema")
        .args(["--format", "parquet"])
        .args(["--output", "float_data.parquet"]);
    wrk.assert_success(&mut cmd);
    let parquet_file = wrk.path("float_data.parquet");
    assert!(parquet_file.exists());

    // gzip float_data.csv, so we go through the polars special format
    // processing workflow that checks for default precision
    let mut cmd = std::process::Command::new("gzip");
    cmd.arg(&test_csv);
    wrk.assert_success(&mut cmd);
    // Check if the gzipped file exists
    let gzipped_file = wrk.path("float_data.csv.gz");
    assert!(gzipped_file.exists());
    // Copy the schema file to match the gzipped filename
    std::fs::copy(
        wrk.path("float_data.csv.pschema.json"),
        wrk.path("float_data.csv.gz.pschema.json"),
    )
    .unwrap();

    // Test with default precision
    let mut cmd = wrk.command("slice");
    cmd.arg(&gzipped_file).arg("--json");
    wrk.assert_success(&mut cmd);
    let got_default: String = wrk.stdout(&mut cmd);

    // Test with custom precision (2 decimal places)
    let mut cmd = wrk.command("slice");
    cmd.arg(&gzipped_file).arg("--json");
    cmd.env("QSV_POLARS_FLOAT_PRECISION", "2");
    wrk.assert_success(&mut cmd);
    let got_precision_2: String = wrk.stdout(&mut cmd);

    // Test with custom precision (5 decimal places)
    let mut cmd = wrk.command("slice");
    cmd.arg(&gzipped_file).arg("--json");
    cmd.env("QSV_POLARS_FLOAT_PRECISION", "5");
    wrk.assert_success(&mut cmd);
    let got_precision_5: String = wrk.stdout(&mut cmd);

    // Test with custom precision (20 decimal places)
    let mut cmd = wrk.command("slice");
    cmd.arg(&parquet_file).arg("--json");
    cmd.env("QSV_POLARS_FLOAT_PRECISION", "20");
    wrk.assert_success(&mut cmd);
    let got_precision_20: String = wrk.stdout(&mut cmd);

    // Verify that the precision affects the output
    // The default precision should be different from precision 2
    assert_ne!(got_default, got_precision_2);

    // Precision 2 should have fewer decimal places than precision 5
    assert!(got_precision_2.len() < got_precision_5.len());

    // Precision 5 should have fewer decimal places than precision 20
    assert!(got_precision_5.len() < got_precision_20.len());

    // Verify that precision 2 has exactly 2 decimal places
    assert!(got_precision_2.contains("3.14"));
    assert!(got_precision_2.contains("2.72"));

    // since we use a polar schema declaring value1 as decimal,
    // QSV_POLARS_FLOAT_PRECISION does not apply
    assert!(got_precision_2.contains("3.14159265358979323846"));
    // it does apply to the value2 column though since its a Float64
    assert!(!got_precision_2.contains("2.718"));

    // Verify that precision 5 has exactly 5 decimal places
    assert!(got_precision_5.contains("3.14159"));
    assert!(got_precision_5.contains("2.71828"));
    assert!(got_precision_5.contains("3.14159265358979323846"));
    assert!(!got_precision_5.contains("2.718281"));

    // Verify that precision 20 has exactly 20 decimal places
    assert!(got_precision_20.contains("3.14159265358979323846"));
    assert!(got_precision_20.contains("1.41421356237309504880"));
    assert!(got_precision_20.contains("2.71828182845904509080"));
    assert!(!got_precision_20.contains("3.141592653589793238462"));
}

#[cfg(feature = "polars")]
#[test]
fn slice_from_json_with_pschema() {
    let wrk = Workdir::new("slice_from_json_with_pschema");

    // Create a test JSON file
    let json_data = r#"[
        {"id": "1", "name": "John", "age": 30, "active": true},
        {"id": "2", "name": "Jane", "age": 25, "active": false},
        {"id": "3", "name": "Bob", "age": 35, "active": true}
    ]"#;
    wrk.create_from_string("test.json", json_data);

    // Create a pschema.json file with a modified schema
    // Change the type of 'age' to String to test schema application
    let schema_data = r#"{
        "fields": {
            "id": "String",
            "name": "String",
            "age": "String",
            "active": "Boolean"
        },"metadata": null
    }"#;
    wrk.create_from_string("test.json.pschema.json", schema_data);

    // Run slice command
    let mut cmd = wrk.command("slice");
    cmd.arg("test.json").arg("--index").arg("1").arg("--json");

    wrk.assert_success(&mut cmd);

    // Verify the output - age should be a string
    let got: String = wrk.stdout(&mut cmd);
    let expected = r#""age":"25""#;
    assert!(got.contains(expected));
}

#[cfg(feature = "polars")]
#[test]
fn slice_from_jsonl_with_pschema() {
    let wrk = Workdir::new("slice_from_jsonl_with_pschema");

    // Create a test JSONL file
    let jsonl_data = r#"{"id": "1", "name": "John", "age": 30, "active": true}
{"id": "2", "name": "Jane", "age": 25, "active": false}
{"id": "3", "name": "Bob", "age": 35, "active": true}"#;
    wrk.create_from_string("test.jsonl", jsonl_data);

    // Create a pschema.json file with a modified schema
    // Change the type of 'age' to String to test schema application
    let schema_data = r#"{
        "fields": {
            "id": "String",
            "name": "String",
            "age": "String",
            "active": "Boolean"
        },"metadata": null
    }"#;
    wrk.create_from_string("test.jsonl.pschema.json", schema_data);

    // Run slice command
    let mut cmd = wrk.command("slice");
    cmd.arg("test.jsonl").arg("--index").arg("1").arg("--json");

    wrk.assert_success(&mut cmd);

    // Verify the output - age should be a string
    let got: String = wrk.stdout(&mut cmd);
    let expected = r#""age":"25""#;
    assert!(got.contains(expected));
}

#[cfg(feature = "polars")]
#[test]
fn slice_from_csvgz_with_pschema() {
    let wrk = Workdir::new("slice_from_csvgz_with_pschema");

    // Create a test CSV file
    let csv_data = "id,name,age,active\n1,John,30,true\n2,Jane,25,false\n3,Bob,35,true";
    wrk.create_from_string("test.csv", csv_data);

    // Create a pschema.json file with a modified schema
    // Change the type of 'age' to String to test schema application
    let schema_data = r#"{
        "fields": {
            "id": "String",
            "name": "String",
            "age": "String",
            "active": "Boolean"
        },"metadata": null
    }"#;
    wrk.create_from_string("test.csv.pschema.json", schema_data);

    // Compress the CSV file with gzip
    let mut cmd = std::process::Command::new("gzip");
    cmd.arg(wrk.path("test.csv"));
    wrk.assert_success(&mut cmd);

    // Run slice command
    let mut cmd = wrk.command("slice");
    cmd.arg("test.csv.gz").arg("--index").arg("1").arg("--json");

    wrk.assert_success(&mut cmd);

    // Verify the output - age should be a string
    let got: String = wrk.stdout(&mut cmd);
    let expected = r#""age":"25""#;
    assert!(got.contains(expected));
}

#[cfg(feature = "polars")]
#[cfg(not(all(target_arch = "aarch64", target_os = "windows")))]
#[test]
fn slice_from_csvzst_with_pschema() {
    let wrk = Workdir::new("slice_from_csvzst_with_pschema");

    // Create a test CSV file
    let csv_data = "id,name,age,active\n1,John,30,true\n2,Jane,25,false\n3,Bob,35,true";
    wrk.create_from_string("test.csv", csv_data);

    // Create a pschema.json file with a modified schema
    // Change the type of 'age' to String to test schema application
    let schema_data = r#"{
        "fields": {
            "id": "String",
            "name": "String",
            "age": "String",
            "active": "Boolean"
        }
    }"#;
    wrk.create_from_string("test.csv.pschema.json", schema_data);

    // Compress the CSV file with zstd
    let mut cmd = std::process::Command::new("zstd");
    cmd.arg(wrk.path("test.csv"));
    wrk.assert_success(&mut cmd);

    // Run slice command
    let mut cmd = wrk.command("slice");
    cmd.arg("test.csv.zst")
        .arg("--index")
        .arg("1")
        .arg("--json");

    wrk.assert_success(&mut cmd);

    // Verify the output - age should be a string
    let got: String = wrk.stdout(&mut cmd);
    let expected = r#""age":"25""#;
    assert!(got.contains(expected));
}

#[cfg(feature = "polars")]
#[test]
fn slice_from_csvzlib_with_pschema() {
    let wrk = Workdir::new("slice_from_csvzlib_with_pschema");

    // Create a test CSV file
    let csv_data = "id,name,age,active\n1,John,30,true\n2,Jane,25,false\n3,Bob,35,true";
    wrk.create_from_string("test.csv", csv_data);

    // Create a pschema.json file with a modified schema
    // Change the type of 'age' to String to test schema application
    let schema_data = r#"{
        "fields": {
            "id": "String",
            "name": "String",
            "age": "String",
            "active": "Boolean"
        }
    }"#;
    wrk.create_from_string("test.csv.pschema.json", schema_data);

    // Compress the CSV file with zlib
    let mut cmd = std::process::Command::new("python3");
    cmd.arg("-c")
        .arg(
            "import zlib; open('test.csv.zlib', 'wb').write(zlib.compress(open('test.csv', \
             'rb').read()))",
        )
        .current_dir(wrk.path("."));
    wrk.assert_success(&mut cmd);

    // Run slice command
    let mut cmd = wrk.command("slice");
    cmd.arg("test.csv.zlib")
        .arg("--index")
        .arg("1")
        .arg("--json");

    wrk.assert_success(&mut cmd);

    // Verify the output - age should be a string
    let got: String = wrk.stdout(&mut cmd);
    let expected = r#""age":"25""#;
    assert!(got.contains(expected));
}

#[cfg(feature = "polars")]
#[test]
// A Decimal pschema applied during special-format conversion preserves all decimal places
// (more than f64 can hold). NOTE: this is exercised via the compressed-CSV reader, NOT
// JSONL — polars' JSONL reader does not support a Decimal target type (it refuses to coerce
// JSON values into a Decimal column), so a JSONL+Decimal pschema simply fails to convert.
// This test previously used JSONL and only "passed" because the resulting conversion error
// was swallowed and the raw bytes (which embed the asserted decimals verbatim) were read
// as-is. See issue #3988.
fn slice_from_csvgz_with_decimal_precision() {
    let wrk = Workdir::new("slice_from_csvgz_with_decimal_precision");

    // High-precision decimal values that exceed f64's ~17 significant digits.
    let csv_data = "id,name,value\n1,John,3.1415926535897932384626433\n2,Jane,2.\
                    7182818284590452353602874\n3,Bob,1.4142135623730950488016887\n";
    wrk.create_from_string("test.csv", csv_data);

    // gzip the CSV; Config detects the .csv.gz special format and decompresses via polars.
    let mut cmd = std::process::Command::new("gzip");
    cmd.arg(wrk.path("test.csv"));
    wrk.assert_success(&mut cmd);

    // The pschema MUST be named to match the actual input (.csv.gz), otherwise it is not
    // loaded and polars falls back to f64 inference (which truncates the precision).
    let schema_data = r#"{
        "fields": {
            "id": "String",
            "name": "String",
            "value": { "Decimal": [38, 25] }
        },"metadata": null
    }"#;
    wrk.create_from_string("test.csv.gz.pschema.json", schema_data);

    let mut cmd = wrk.command("slice");
    cmd.arg("test.csv.gz").arg("--json");

    // All decimal places are preserved (an f64 round-trip would truncate these).
    let got: String = wrk.stdout_on_success(&mut cmd);
    assert!(got.contains("3.1415926535897932384626433"));
    assert!(got.contains("2.7182818284590452353602874"));
    assert!(got.contains("1.4142135623730950488016887"));
}

#[cfg(feature = "polars")]
#[test]
fn slice_from_tsvgz_with_decimal_precision() {
    let wrk = Workdir::new("slice_from_tsvgz_with_decimal_precision");

    // Create a test TSV file with high precision decimal values
    wrk.create_with_delim(
        "test.tsv",
        vec![
            svec!["id", "name", "value"],
            svec!["1", "John", "3.1415926535897932384626433"],
            svec!["2", "Jane", "2.7182818284590452353602874"],
            svec!["3", "Bob", "1.4142135623730950488016887"],
        ],
        b'\t',
    );

    // Create a pschema.json file with Decimal type for the 'value' field
    // Using Decimal with precision 25 and scale 25 to maintain all decimal places
    let schema_data = r#"{
        "fields": {
            "id": "String",
            "name": "String",
            "value": { "Decimal": [25, 25] }
        }
    }"#;
    wrk.create_from_string("test.tsv.pschema.json", schema_data);

    // Compress the TSV file with gzip
    let mut cmd = std::process::Command::new("gzip");
    cmd.arg(wrk.path("test.tsv"));
    wrk.assert_success(&mut cmd);

    // Run slice command
    let mut cmd = wrk.command("slice");
    cmd.env("QSV_POLARS_FLOAT_PRECISION", "25")
        .arg("test.tsv.gz")
        .arg("--json");

    wrk.assert_success(&mut cmd);

    // Verify the output - value should maintain all decimal places
    let got: String = wrk.stdout(&mut cmd);

    assert!(got.contains("3.1415926535897931159979635"));
    assert!(got.contains("2.7182818284590450907955983"));
    assert!(got.contains("1.4142135623730951454746219"));
}

#[cfg(feature = "polars")]
#[test]
// A Decimal pschema preserves more precision than Float64. Exercised via the compressed-CSV
// reader because polars' JSONL reader does not support a Decimal target type — see the note
// on slice_from_csvgz_with_decimal_precision and issue #3988. Previously used JSONL and only
// "passed" because the conversion error was swallowed and raw bytes were read as-is.
fn slice_from_csvgz_with_decimal_precision_vs_float() {
    let wrk = Workdir::new("slice_from_csvgz_with_decimal_precision_vs_float");

    // High-precision decimal values that exceed f64's ~17 significant digits.
    let csv_data = "id,name,value\n1,John,3.1415926535897932384626433\n2,Jane,2.\
                    7182818284590452353602874\n3,Bob,1.4142135623730950488016887\n";
    wrk.create_from_string("test.csv", csv_data);

    // gzip the CSV; Config detects the .csv.gz special format and decompresses via polars.
    let mut cmd = std::process::Command::new("gzip");
    cmd.arg(wrk.path("test.csv"));
    wrk.assert_success(&mut cmd);

    // Decimal schema (named to match the .csv.gz input) preserves all decimal places.
    let decimal_schema = r#"{
        "fields": {
            "id": "String",
            "name": "String",
            "value": { "Decimal": [38, 25] }
        },"metadata": null
    }"#;
    wrk.create_from_string("test.csv.gz.pschema.json", decimal_schema);

    // Run slice command with Decimal schema
    let mut cmd = wrk.command("slice");
    cmd.arg("test.csv.gz").arg("--json");

    // Get the output with Decimal schema
    let decimal_output: String = wrk.stdout_on_success(&mut cmd);

    // Now create a schema with Float64 instead of Decimal
    let float_schema = r#"{
        "fields": {
            "id": "String",
            "name": "String",
            "value": "Float64"
        },"metadata": null
    }"#;
    wrk.create_from_string("test.csv.gz.pschema.json", float_schema);

    // Run slice command with Float64 schema
    let mut cmd = wrk.command("slice");
    cmd.arg("test.csv.gz").arg("--json");

    // Get the output with Float64 schema
    let float_output: String = wrk.stdout_on_success(&mut cmd);

    // Verify that the Decimal schema preserves more precision than Float64
    // The Float64 output should be truncated compared to the Decimal output
    assert!(decimal_output.len() > float_output.len());

    // Check that the Decimal output contains the full precision values
    assert!(decimal_output.contains("3.1415926535897932384626433"));
    assert!(decimal_output.contains("2.7182818284590452353602874"));
    assert!(decimal_output.contains("1.4142135623730950488016887"));

    // The Float64 output should have truncated values
    // We're checking for the absence of the full precision values in the Float64 output
    assert!(!float_output.contains("3.1415926535897932384626433"));
    assert!(!float_output.contains("2.7182818284590452353602874"));
    assert!(!float_output.contains("1.4142135623730950488016887"));
}
