tests/collectors_mainframe.rs
//! Offline INJECT/ASSERT contracts for the **mainframe** collector (feature
//! `mainframe`): a COBOL copybook + a hand-crafted EBCDIC-037 fixed-width record
//! blob → assert the copybook→schema AND the decoded Arrow VALUES (esp. COMP-3 →
//! correct `Decimal128`). The copybook IS the schema, so this is a fully
//! deterministic inject/assert target. Emits functional-status rows under
//! `--features testmatrix`.
#![cfg(feature = "mainframe")]
use arrow_array::{Array, Decimal128Array, Int32Array, Int64Array, StringArray};
use arrow_schema::DataType;
use korp_collectors::fixedwidth::{Usage, decode_ascii_records, parse_copybook, parse_pic};
use korp_collectors::mainframe::{
CodePage, MainframeCollector, binary_to_i128, decode_ebcdic_records, ebcdic_to_utf8,
unpack_comp3, unzone,
};
use korp_collectors::{BatchSink, Bounds, Collector, CollectorCfg};
mod common;
use common::emit_for;
// EBCDIC-037 encode, via the crate's own `utf8_to_ebcdic` — the exact inverse of
// the decoder under test, derived from the SAME IBM-037 table. This used to be a
// hand-built partial table (letters/digits/space) living only in this file, free
// to drift from the decoder it was supposed to mirror; the shared encoder is
// consumed instead of twinned.
fn ascii_to_ebcdic037(s: &str) -> Vec<u8> {
korp_collectors::mainframe::utf8_to_ebcdic(s, CodePage::Ibm037)
}
/// The encoder is the decoder's inverse over the whole code page, not just the
/// letters/digits/space the old hand-built table covered. RED-when-broken: a
/// single wrong entry breaks the round-trip.
#[test]
fn ebcdic_encode_decode_round_trips_the_whole_code_page() {
use korp_collectors::mainframe::utf8_to_ebcdic;
// Every byte 0x00..=0xFF decodes to a char that re-encodes to that same byte.
let all: Vec<u8> = (0u16..=255).map(|b| b as u8).collect();
let text = ebcdic_to_utf8(&all, CodePage::Ibm037);
let back = utf8_to_ebcdic(&text, CodePage::Ibm037);
assert_eq!(
back, all,
"EBCDIC-037 encode is the exact inverse of decode"
);
// ...and the fixture alphabet still lands where the old hand table put it.
assert_eq!(utf8_to_ebcdic("A", CodePage::Ibm037), vec![0xC1]);
assert_eq!(utf8_to_ebcdic("J", CodePage::Ibm037), vec![0xD1]);
assert_eq!(utf8_to_ebcdic("S", CodePage::Ibm037), vec![0xE2]);
assert_eq!(utf8_to_ebcdic("0", CodePage::Ibm037), vec![0xF0]);
assert_eq!(utf8_to_ebcdic(" ", CodePage::Ibm037), vec![0x40]);
emit_for(
"korp-collectors/mainframe",
"ebcdic_encode_is_decode_inverse",
true,
"all 256 IBM-037 bytes round-trip encode∘decode",
);
}
// ── #1 PIC parser + copybook → schema (spec IS the schema) ────────────────────
const COPYBOOK: &str = "\
01 CUSTOMER-RECORD.
05 CUST-NAME PIC X(5).
05 CUST-BALANCE PIC S9(5)V99 COMP-3.
05 CUST-ID PIC 9(4) COMP.
";
#[test]
fn copybook_parses_to_schema() {
// PIC parse facts
let bal = parse_pic("S9(5)V99");
assert_eq!((bal.signed, bal.int_digits, bal.frac_digits), (true, 5, 2));
assert_eq!(parse_pic("X(5)").x_len, 5);
let layout = parse_copybook(COPYBOOK).expect("parse copybook");
// COMP-3 S9(5)V99 = ceil((7+1)/2)=4 bytes; X(5)=5; 9(4) COMP=2 → LRECL 11.
assert_eq!(layout.lrecl, 11, "5 + 4 + 2 bytes");
assert_eq!(layout.fields.len(), 3);
assert_eq!(layout.fields[1].usage, Usage::Comp3);
let schema = layout.schema();
let types: Vec<DataType> = schema
.fields()
.iter()
.map(|f| f.data_type().clone())
.collect();
assert_eq!(
types,
vec![DataType::Utf8, DataType::Decimal128(7, 2), DataType::Int32],
"PIC X→Utf8, S9(5)V99 COMP-3→Decimal128(7,2), 9(4) COMP→Int32"
);
let names: Vec<&str> = schema.fields().iter().map(|f| f.name().as_str()).collect();
assert_eq!(names, vec!["CUST-NAME", "CUST-BALANCE", "CUST-ID"]);
emit_for(
"korp-collectors/mainframe",
"copybook_to_schema",
true,
"3 fields, LRECL 11, COMP-3→Decimal128(7,2)",
);
}
// ── #2 decode primitives (hand-computed expected bytes) ───────────────────────
#[test]
fn decode_primitives_by_value() {
// COMP-3: -123.45 = 12 34 5D (unscaled -12345)
assert_eq!(unpack_comp3(&[0x12, 0x34, 0x5D]), Some(-12345));
// COMP-3: +123.45 = 12 34 5C
assert_eq!(unpack_comp3(&[0x12, 0x34, 0x5C]), Some(12345));
// COMP-3: -42.50 at scale 2 = 00 04 25 0D (unscaled -4250)
assert_eq!(unpack_comp3(&[0x00, 0x04, 0x25, 0x0D]), Some(-4250));
// bad nibble
assert_eq!(unpack_comp3(&[0xAB, 0xCD]), None);
// zoned EBCDIC "123" unsigned = F1 F2 F3 → 123; negative sign zone D on last
assert_eq!(unzone(&[0xF1, 0xF2, 0xF3]), Some(123));
assert_eq!(unzone(&[0xF1, 0xF2, 0xD3]), Some(-123));
// binary big-endian: 0x04 0xD2 = 1234
assert_eq!(binary_to_i128(&[0x04, 0xD2], false), 1234);
// signed two's complement: 0xFF 0xFF = -1
assert_eq!(binary_to_i128(&[0xFF, 0xFF], true), -1);
// EBCDIC text round-trip
let bytes = ascii_to_ebcdic037("ANNA");
assert_eq!(ebcdic_to_utf8(&bytes, CodePage::Ibm037), "ANNA");
emit_for(
"korp-collectors/mainframe",
"decode_primitives",
true,
"COMP-3/zoned/binary/EBCDIC by hand-computed bytes",
);
}
// ── #3 full EBCDIC record → Arrow VALUES ──────────────────────────────────────
#[test]
fn ebcdic_record_decodes_to_arrow_values() {
let layout = parse_copybook(COPYBOOK).expect("parse copybook");
// Build one 11-byte EBCDIC record:
// CUST-NAME = "ANNA " (X(5), EBCDIC)
// CUST-BALANCE = -42.50 (S9(5)V99 COMP-3, 4 bytes = 00 04 25 0D)
// CUST-ID = 1234 (9(4) COMP, 2 bytes big-endian = 04 D2)
let mut rec = Vec::new();
rec.extend_from_slice(&ascii_to_ebcdic037("ANNA ")); // 5 bytes
rec.extend_from_slice(&[0x00, 0x04, 0x25, 0x0D]); // -4250 packed
rec.extend_from_slice(&[0x04, 0xD2]); // 1234 binary
assert_eq!(rec.len(), layout.lrecl);
let rb = decode_ebcdic_records(&layout, &rec, CodePage::Ibm037).expect("decode");
assert_eq!(rb.num_rows(), 1);
let name = rb.column(0).as_any().downcast_ref::<StringArray>().unwrap();
let bal = rb
.column(1)
.as_any()
.downcast_ref::<Decimal128Array>()
.unwrap();
let id = rb.column(2).as_any().downcast_ref::<Int32Array>().unwrap();
assert_eq!(name.value(0), "ANNA", "EBCDIC→UTF-8, right-trimmed");
assert_eq!(bal.value(0), -4250, "unscaled COMP-3 mantissa");
assert_eq!(bal.value_as_string(0), "-42.50", "Decimal128(7,2) render");
assert_eq!(id.value(0), 1234, "big-endian COMP binary");
emit_for(
"korp-collectors/mainframe",
"ebcdic_record_to_arrow",
true,
"ANNA / -42.50 / 1234 by value",
);
}
// ── #4 empty input → 0 rows (RAGNARÖK reason) ─────────────────────────────────
#[tokio::test]
async fn empty_input_sets_reason() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("empty.ebc");
std::fs::write(&path, b"").unwrap();
let cfg = CollectorCfg::new(path.to_string_lossy().to_string(), "customers")
.with_params(serde_json::json!({ "copybook": COPYBOOK }));
let sink = BatchSink::memory();
let stats = MainframeCollector::default()
.collect(&cfg, Bounds::default(), &sink)
.await
.unwrap();
assert_eq!(stats.rows, 0);
assert!(
stats.reason.is_some(),
"0-row collect is RED unless reason set (RAGNARÖK)"
);
emit_for(
"korp-collectors/mainframe",
"empty_input_ragnarok",
true,
"0 records → reason set",
);
}
// ── #5 ASCII fixed-width cousin (same layout parser, no EBCDIC/COMP-3) ─────────
const ASCII_COPYBOOK: &str = "\
01 ROW.
05 CITY PIC X(6).
05 POP PIC 9(6).
";
#[test]
fn ascii_fixed_width_reuses_layout() {
let layout = parse_copybook(ASCII_COPYBOOK).expect("parse");
assert_eq!(layout.lrecl, 12, "6 + 6 (all DISPLAY)");
// two 12-byte ASCII records
let blob = b"Solna 082000Malmo 344000";
let rb = decode_ascii_records(&layout, blob).expect("decode ascii");
assert_eq!(rb.num_rows(), 2);
let city = rb.column(0).as_any().downcast_ref::<StringArray>().unwrap();
let pop = rb.column(1).as_any().downcast_ref::<Int64Array>().unwrap();
assert_eq!(city.value(0), "Solna");
assert_eq!(pop.value(0), 82000, "leading-zero ASCII digits → Int64");
assert_eq!(city.value(1), "Malmo");
assert_eq!(pop.value(1), 344000);
emit_for(
"korp-collectors/mainframe",
"ascii_fixed_width",
true,
"2 rows, X→Utf8, 9(6)→Int64",
);
}
// ── #6 REDEFINES overlay (redefining field placed at the redefined offset) ────
const REDEFINES_COPYBOOK: &str = "\
01 REC.
05 RAW-CODE PIC X(4).
05 NUM-CODE REDEFINES RAW-CODE PIC 9(4).
";
#[test]
fn copybook_redefines_overlays_at_same_offset() {
let layout = parse_copybook(REDEFINES_COPYBOOK).expect("parse redefines copybook");
// The overlay is no longer dropped — BOTH alternatives are present.
assert_eq!(layout.fields.len(), 2, "redefining field kept, not dropped");
let names: Vec<&str> = layout.fields.iter().map(|f| f.name.as_str()).collect();
assert_eq!(names, vec!["RAW-CODE", "NUM-CODE"]);
// Same byte offset, and the overlay does NOT extend the record.
assert_eq!(layout.fields[0].offset, 0);
assert_eq!(
layout.fields[1].offset, 0,
"NUM-CODE overlays RAW-CODE at offset 0"
);
assert_eq!(
layout.lrecl, 4,
"REDEFINES does not advance the record cursor"
);
// Distinct decode types over the same bytes: X→Utf8, 9(4)→Int64.
assert_eq!(layout.fields[0].arrow_type, DataType::Utf8);
assert_eq!(layout.fields[1].arrow_type, DataType::Int64);
// Decode one 4-byte record: both columns read the SAME four bytes "1234".
let rb = decode_ascii_records(&layout, b"1234").expect("decode redefines record");
assert_eq!(rb.num_rows(), 1);
let raw = rb.column(0).as_any().downcast_ref::<StringArray>().unwrap();
let num = rb.column(1).as_any().downcast_ref::<Int64Array>().unwrap();
assert_eq!(raw.value(0), "1234", "text view of the overlaid bytes");
assert_eq!(
num.value(0),
1234,
"numeric view decoded from the SAME offset"
);
emit_for(
"korp-collectors/mainframe",
"copybook_redefines_overlay",
true,
"REDEFINES overlays at same offset; both fields present; LRECL unchanged",
);
}
// ── descriptor: mainframe + fixed-width collectors declare FlatFile targets ───
// The source→target descriptor seam is default-only in `collectors_sample.rs`
// (mainframe is feature-gated), so assert the two file-format collectors'
// declarations here, where `--features mainframe` puts them in the registry.
#[test]
fn mainframe_collectors_declare_flatfile_warehouse_descriptors() {
use korp_collectors::fixedwidth::FixedWidthCollector;
use korp_collectors::{SourceEndpoint, TargetKind, registry_descriptors};
// Direct declarations on each collector.
for (st, table) in [
(
MainframeCollector::default().source_target(),
"mainframe_import",
),
(
FixedWidthCollector::default().source_target(),
"fixedwidth_import",
),
] {
assert_eq!(st.source, SourceEndpoint::FlatFile);
assert_eq!(st.target, TargetKind::WarehouseTable);
assert_eq!(st.target_table, table);
assert!(
st.map_scene.is_none(),
"a bulk copybook import feeds no map scene"
);
assert!(
st.label().contains("flat-file"),
"label carries the source tag: {}",
st.label()
);
}
// Both JOIN the registry under `--features mainframe`.
let descs = registry_descriptors();
for name in ["mainframe-copybook", "fixed-width"] {
let (_, st) = descs
.iter()
.find(|(m, _)| m.name == name)
.unwrap_or_else(|| panic!("{name} present in registry under feature mainframe"));
assert_eq!(st.source, SourceEndpoint::FlatFile);
assert_eq!(st.target, TargetKind::WarehouseTable);
emit_for(
&format!("korp-collectors/{name}"),
"declares_source_target",
true,
&st.label(),
);
}
}