#![allow(
clippy::panic,
clippy::unwrap_used,
clippy::expect_used,
clippy::indexing_slicing,
clippy::as_conversions,
clippy::cast_possible_truncation,
clippy::cast_precision_loss,
clippy::same_item_push
)]
use std::process::Command;
fn binary_path() -> std::path::PathBuf {
let mut path = std::env::current_exe()
.expect("cannot determine test executable path")
.parent()
.expect("no parent directory")
.parent()
.expect("no grandparent directory")
.to_path_buf();
path.push(if cfg!(windows) { "amn.exe" } else { "amn" });
assert!(
path.exists(),
"amn binary not found at {}. Run `cargo build --features cli,npz,pth,gguf,bnb` \
before `cargo test`.",
path.display()
);
path
}
fn build_safetensors_bf16(tensors: &[(&str, &[usize], &[u8])]) -> Vec<u8> {
let views: Vec<(&str, safetensors::tensor::TensorView<'_>)> = tensors
.iter()
.map(|(name, shape, data)| {
let view = safetensors::tensor::TensorView::new(
safetensors::Dtype::BF16,
shape.to_vec(),
data,
)
.unwrap();
(*name, view)
})
.collect();
safetensors::tensor::serialize(views, None).unwrap()
}
#[cfg(feature = "npz")]
fn build_npz_f32(tensors: &[(&str, &[usize], &[u8])]) -> Vec<u8> {
use std::io::Write;
let mut zip = zip::ZipWriter::new(std::io::Cursor::new(Vec::<u8>::new()));
let options: zip::write::SimpleFileOptions =
zip::write::SimpleFileOptions::default().compression_method(zip::CompressionMethod::Stored);
for (name, shape, data) in tensors {
let entry = format!("{name}.npy");
zip.start_file(&entry, options).unwrap();
let shape_str: Vec<String> = shape.iter().map(usize::to_string).collect();
let shape_tuple = if shape.len() == 1 {
format!("({},)", shape_str[0])
} else {
format!("({})", shape_str.join(", "))
};
let dict = format!("{{'descr': '<f4', 'fortran_order': False, 'shape': {shape_tuple}, }}");
let mut header = dict.into_bytes();
let header_total = 10 + header.len() + 1;
let pad = (64 - header_total % 64) % 64;
for _ in 0..pad {
header.push(b' ');
}
header.push(b'\n');
let header_len_u16 = u16::try_from(header.len()).unwrap();
let mut entry_bytes: Vec<u8> = Vec::with_capacity(10 + header.len() + data.len());
entry_bytes.extend_from_slice(&[0x93, b'N', b'U', b'M', b'P', b'Y', 1, 0]);
entry_bytes.extend_from_slice(&header_len_u16.to_le_bytes());
entry_bytes.extend_from_slice(&header);
entry_bytes.extend_from_slice(data);
zip.write_all(&entry_bytes).unwrap();
}
zip.finish().unwrap().into_inner()
}
fn write_temp(bytes: &[u8], ext: &str) -> (tempfile::TempDir, std::path::PathBuf) {
let dir = tempfile::tempdir().expect("create temp dir");
let path = dir.path().join(format!("fixture.{ext}"));
std::fs::write(&path, bytes).unwrap();
(dir, path)
}
#[cfg(feature = "bnb")]
fn bf16_bytes_from_f32_iter<I: IntoIterator<Item = f32>>(values: I) -> Vec<u8> {
let mut out = Vec::new();
for v in values {
let bits = (v.to_bits() >> 16) as u16;
out.extend_from_slice(&bits.to_le_bytes());
}
out
}
#[test]
#[cfg(feature = "npz")]
fn convert_npz_to_safetensors_smokes() {
let f32_data: Vec<u8> = (0..4u32)
.flat_map(|i| f32::from(i as u16).to_le_bytes())
.collect();
let other: Vec<u8> = (0..6u32)
.flat_map(|i| (f32::from(i as u16) * 0.5).to_le_bytes())
.collect();
let npz_bytes = build_npz_f32(&[("w", &[2, 2], &f32_data), ("b", &[6], &other)]);
let (_dir, in_path) = write_temp(&npz_bytes, "npz");
let out_path = in_path.with_extension("safetensors");
let output = Command::new(binary_path())
.args([
"convert",
in_path.to_str().unwrap(),
"--to",
"safetensors",
"-o",
out_path.to_str().unwrap(),
])
.output()
.expect("run amn convert");
let stderr = String::from_utf8_lossy(&output.stderr);
let stdout = String::from_utf8_lossy(&output.stdout);
assert!(
output.status.success(),
"amn convert failed\nstdout: {stdout}\nstderr: {stderr}"
);
assert!(out_path.exists(), "output file missing");
let bytes = std::fs::read(&out_path).unwrap();
let parsed = safetensors::SafeTensors::deserialize(&bytes).unwrap();
let mut names: Vec<&str> = parsed.names();
names.sort_unstable();
assert_eq!(names, vec!["b", "w"]);
}
#[test]
#[cfg(feature = "gguf")]
fn convert_safetensors_bf16_to_gguf_smokes() {
let bf16: Vec<u8> = (0..8u32)
.flat_map(|i| {
let v = i as f32;
let bits = (v.to_bits() >> 16) as u16;
bits.to_le_bytes()
})
.collect();
let st_bytes = build_safetensors_bf16(&[("w", &[2, 4], &bf16)]);
let (_dir, in_path) = write_temp(&st_bytes, "safetensors");
let out_path = in_path.with_file_name("out.gguf");
let output = Command::new(binary_path())
.args([
"convert",
in_path.to_str().unwrap(),
"--to",
"gguf",
"-o",
out_path.to_str().unwrap(),
])
.output()
.expect("run amn convert");
let stderr = String::from_utf8_lossy(&output.stderr);
let stdout = String::from_utf8_lossy(&output.stdout);
assert!(
output.status.success(),
"amn convert failed\nstdout: {stdout}\nstderr: {stderr}"
);
let parsed = anamnesis::parse_gguf(&out_path).unwrap();
let info = parsed.inspect();
assert_eq!(info.tensor_count, 1);
let collected: Vec<_> = parsed.tensors().collect();
assert_eq!(collected[0].name, "w");
assert_eq!(collected[0].dtype, anamnesis::GgufType::BF16);
assert_eq!(collected[0].shape, &[4, 2]);
assert_eq!(collected[0].data.as_ref(), bf16.as_slice());
}
#[test]
#[cfg(feature = "bnb")]
fn convert_safetensors_bf16_to_bnb_nf4_smokes() {
let bf16 = bf16_bytes_from_f32_iter((0..64).map(|i| (i as f32 - 31.5) / 32.0));
let st_bytes = build_safetensors_bf16(&[("linear", &[64, 1], &bf16)]);
let (_dir, in_path) = write_temp(&st_bytes, "safetensors");
let out_path = in_path.with_file_name("out-bnb-nf4.safetensors");
let output = Command::new(binary_path())
.args([
"convert",
in_path.to_str().unwrap(),
"--to",
"bnb-nf4",
"-o",
out_path.to_str().unwrap(),
])
.output()
.expect("run amn convert");
let stderr = String::from_utf8_lossy(&output.stderr);
let stdout = String::from_utf8_lossy(&output.stdout);
assert!(
output.status.success(),
"amn convert failed\nstdout: {stdout}\nstderr: {stderr}"
);
let model = anamnesis::parse(&out_path).unwrap();
assert_eq!(model.header.scheme, anamnesis::QuantScheme::Bnb4);
let names: Vec<&str> = model
.header
.tensors
.iter()
.map(|t| t.name.as_str())
.collect();
assert!(names.contains(&"linear.weight"));
assert!(names.contains(&"linear.weight.absmax"));
assert!(names.contains(&"linear.weight.quant_map"));
assert!(names.contains(&"linear.weight.quant_state.bitsandbytes__nf4"));
}
#[test]
#[cfg(all(feature = "npz", feature = "bnb"))]
fn convert_npz_to_bnb_nf4_smokes() {
let f32_data: Vec<u8> = (0..64)
.flat_map(|i| ((i as f32 - 31.5) / 32.0).to_le_bytes())
.collect();
let npz_bytes = build_npz_f32(&[("linear", &[64, 1], &f32_data)]);
let (_dir, in_path) = write_temp(&npz_bytes, "npz");
let out_path = in_path.with_file_name("out-npz-bnb-nf4.safetensors");
let output = Command::new(binary_path())
.args([
"convert",
in_path.to_str().unwrap(),
"--to",
"bnb-nf4",
"-o",
out_path.to_str().unwrap(),
])
.output()
.expect("run amn convert");
let stderr = String::from_utf8_lossy(&output.stderr);
let stdout = String::from_utf8_lossy(&output.stdout);
assert!(
output.status.success(),
"amn convert npz -> bnb-nf4 failed\nstdout: {stdout}\nstderr: {stderr}"
);
let model = anamnesis::parse(&out_path).unwrap();
assert_eq!(model.header.scheme, anamnesis::QuantScheme::Bnb4);
let names: Vec<&str> = model
.header
.tensors
.iter()
.map(|t| t.name.as_str())
.collect();
assert!(names.contains(&"linear.weight"));
assert!(names.contains(&"linear.weight.absmax"));
}
#[test]
#[cfg(all(feature = "pth", feature = "bnb"))]
fn convert_pth_to_bnb_nf4_passes_ineligible_tensors_through() {
let in_path = std::path::Path::new("tests/fixtures/pth_reference/algzoo_rnn_small.pth");
assert!(in_path.exists(), "committed .pth fixture missing");
let dir = tempfile::tempdir().expect("tempdir");
let out_path = dir.path().join("out-pth-bnb-nf4.safetensors");
let output = Command::new(binary_path())
.args([
"convert",
in_path.to_str().unwrap(),
"--to",
"bnb-nf4",
"-o",
out_path.to_str().unwrap(),
])
.output()
.expect("run amn convert");
let stderr = String::from_utf8_lossy(&output.stderr);
let stdout = String::from_utf8_lossy(&output.stdout);
assert!(
output.status.success(),
"amn convert pth -> bnb-nf4 failed\nstdout: {stdout}\nstderr: {stderr}"
);
let model = anamnesis::parse(&out_path).unwrap();
let names: Vec<&str> = model
.header
.tensors
.iter()
.map(|t| t.name.as_str())
.collect();
assert_eq!(names.len(), 3, "all three tensors survive: {names:?}");
assert!(names.contains(&"rnn.weight_ih_l0"));
assert!(names.contains(&"rnn.weight_hh_l0"));
assert!(names.contains(&"linear.weight"));
assert!(
!names.iter().any(|n| n.ends_with(".absmax")),
"no tensor is NF4-eligible, so none should gain companions: {names:?}"
);
assert!(
model
.header
.tensors
.iter()
.all(|t| t.dtype == anamnesis::Dtype::BF16),
"passthrough tensors should be BF16"
);
}
#[test]
#[cfg(all(feature = "gguf", feature = "bnb"))]
fn convert_gguf_to_bnb_nf4_smokes() {
use std::collections::HashMap;
let f32_bytes: Vec<u8> = (0..64)
.flat_map(|i| ((i as f32 - 31.5) / 32.0).to_le_bytes())
.collect();
let shape = [1usize, 64];
let tensors = [anamnesis::GgufWriteTensor {
name: "linear",
shape: &shape,
dtype: anamnesis::GgufType::F32,
data: &f32_bytes,
}];
let dir = tempfile::tempdir().expect("tempdir");
let in_path = dir.path().join("in.gguf");
anamnesis::write_gguf(&in_path, &tensors, &HashMap::new()).expect("write gguf fixture");
let out_path = dir.path().join("out-gguf-bnb-nf4.safetensors");
let output = Command::new(binary_path())
.args([
"convert",
in_path.to_str().unwrap(),
"--to",
"bnb-nf4",
"-o",
out_path.to_str().unwrap(),
])
.output()
.expect("run amn convert");
let stderr = String::from_utf8_lossy(&output.stderr);
let stdout = String::from_utf8_lossy(&output.stdout);
assert!(
output.status.success(),
"amn convert gguf -> bnb-nf4 failed\nstdout: {stdout}\nstderr: {stderr}"
);
let model = anamnesis::parse(&out_path).unwrap();
assert_eq!(model.header.scheme, anamnesis::QuantScheme::Bnb4);
let names: Vec<&str> = model
.header
.tensors
.iter()
.map(|t| t.name.as_str())
.collect();
assert!(names.contains(&"linear.weight"));
assert!(names.contains(&"linear.weight.absmax"));
}
#[test]
#[cfg(feature = "gguf")]
fn convert_gguf_to_gguf_preserves_metadata() {
use anamnesis::GgufMetadataValue;
use std::collections::HashMap;
let f32_bytes: Vec<u8> = (0..8).flat_map(|i| (i as f32).to_le_bytes()).collect();
let shape = [8usize];
let tensors = [anamnesis::GgufWriteTensor {
name: "w",
shape: &shape,
dtype: anamnesis::GgufType::F32,
data: &f32_bytes,
}];
let mut meta: HashMap<String, GgufMetadataValue> = HashMap::new();
meta.insert(
"general.architecture".to_owned(),
GgufMetadataValue::String("llama".to_owned()),
);
meta.insert("llama.block_count".to_owned(), GgufMetadataValue::U32(32));
let dir = tempfile::tempdir().expect("tempdir");
let in_path = dir.path().join("in.gguf");
anamnesis::write_gguf(&in_path, &tensors, &meta).expect("write gguf fixture");
let out_path = dir.path().join("out.gguf");
let output = Command::new(binary_path())
.args([
"convert",
in_path.to_str().unwrap(),
"--to",
"gguf",
"-o",
out_path.to_str().unwrap(),
])
.output()
.expect("run amn convert");
let stderr = String::from_utf8_lossy(&output.stderr);
assert!(
output.status.success(),
"amn convert gguf -> gguf failed\nstderr: {stderr}"
);
let reparsed = anamnesis::parse_gguf(&out_path).expect("re-parse converted gguf");
let out_meta = reparsed.metadata();
assert_eq!(
out_meta.get("general.architecture"),
Some(&GgufMetadataValue::String("llama".to_owned())),
"source architecture KV must survive the round trip: {out_meta:?}"
);
assert_eq!(
out_meta.get("llama.block_count"),
Some(&GgufMetadataValue::U32(32)),
"typed numeric KV must survive with its type intact: {out_meta:?}"
);
}
#[test]
#[cfg(feature = "gguf")]
fn convert_quantized_safetensors_to_gguf_auto_chains() {
let in_path = std::path::Path::new("tests/fixtures/safetensors_reference/fp8.safetensors");
assert!(in_path.exists(), "committed FP8 fixture missing");
let dir = tempfile::tempdir().expect("tempdir");
let out_path = dir.path().join("out.gguf");
let output = Command::new(binary_path())
.args([
"convert",
in_path.to_str().unwrap(),
"--to",
"gguf",
"-o",
out_path.to_str().unwrap(),
])
.output()
.expect("run amn convert");
let stdout = String::from_utf8_lossy(&output.stdout);
let stderr = String::from_utf8_lossy(&output.stderr);
assert!(
output.status.success(),
"quantised safetensors -> gguf should auto-chain\nstdout: {stdout}\nstderr: {stderr}"
);
assert!(
stdout.contains("dequantized to BF16"),
"expected the chain to report a dequant step, got: {stdout}"
);
let reparsed = anamnesis::parse_gguf(&out_path).expect("re-parse converted gguf");
assert!(
reparsed.tensors().all(|t| !t.dtype.is_quantized()),
"output GGUF must be scalar (dequantised)"
);
}
#[test]
#[cfg(feature = "gguf")]
fn convert_gguf_metadata_flags_merge_over_source_kv() {
use anamnesis::GgufMetadataValue;
use std::collections::HashMap;
let f32_bytes: Vec<u8> = (0..4).flat_map(|i| (i as f32).to_le_bytes()).collect();
let shape = [4usize];
let tensors = [anamnesis::GgufWriteTensor {
name: "w",
shape: &shape,
dtype: anamnesis::GgufType::F32,
data: &f32_bytes,
}];
let mut source_meta: HashMap<String, GgufMetadataValue> = HashMap::new();
source_meta.insert(
"general.architecture".to_owned(),
GgufMetadataValue::String("source-arch".to_owned()),
);
source_meta.insert(
"general.name".to_owned(),
GgufMetadataValue::String("keepme".to_owned()),
);
let dir = tempfile::tempdir().expect("tempdir");
let in_path = dir.path().join("in.gguf");
anamnesis::write_gguf(&in_path, &tensors, &source_meta).expect("write gguf fixture");
let meta_path = dir.path().join("meta.json");
std::fs::write(
&meta_path,
r#"{
"general.architecture": "from-file",
"llama.block_count": {"type": "u32", "value": 32},
"tokenizer.ggml.token_type": {"type": "array", "item_type": "i32", "value": [1, 2]}
}"#,
)
.expect("write metadata json");
let out_path = dir.path().join("out.gguf");
let output = Command::new(binary_path())
.args([
"convert",
in_path.to_str().unwrap(),
"--to",
"gguf",
"-o",
out_path.to_str().unwrap(),
"--gguf-metadata",
meta_path.to_str().unwrap(),
"--gguf-kv",
"general.architecture=from-flag",
])
.output()
.expect("run amn convert");
let stderr = String::from_utf8_lossy(&output.stderr);
assert!(
output.status.success(),
"amn convert with metadata flags failed\nstderr: {stderr}"
);
let reparsed = anamnesis::parse_gguf(&out_path).expect("re-parse converted gguf");
let meta = reparsed.metadata();
assert_eq!(
meta.get("general.architecture"),
Some(&GgufMetadataValue::String("from-flag".to_owned())),
"flag should win the three-way merge: {meta:?}"
);
assert_eq!(
meta.get("general.name"),
Some(&GgufMetadataValue::String("keepme".to_owned())),
"untouched source KV must survive: {meta:?}"
);
assert_eq!(
meta.get("llama.block_count"),
Some(&GgufMetadataValue::U32(32))
);
assert_eq!(
meta.get("tokenizer.ggml.token_type"),
Some(&GgufMetadataValue::Array(Box::new(
anamnesis::GgufMetadataArray::I32(vec![1, 2])
))),
"the explicit item_type must survive as I32, not the inferred U32"
);
}
#[test]
#[cfg(feature = "gguf")]
fn convert_rejects_malformed_gguf_kv() {
let bf16: Vec<u8> = vec![0u8; 2];
let st_bytes = build_safetensors_bf16(&[("w", &[1], &bf16)]);
let (_dir, in_path) = write_temp(&st_bytes, "safetensors");
let output = Command::new(binary_path())
.args([
"convert",
in_path.to_str().unwrap(),
"--to",
"gguf",
"--gguf-kv",
"no-equals-sign",
])
.output()
.expect("run amn convert");
assert!(
!output.status.success(),
"should reject malformed --gguf-kv"
);
let stderr = String::from_utf8_lossy(&output.stderr);
assert!(
stderr.contains("key=value"),
"expected a `key=value` hint, got stderr: {stderr}"
);
}
#[test]
fn convert_unknown_target_errors_cleanly() {
let bf16: Vec<u8> = vec![0u8; 2];
let st_bytes = build_safetensors_bf16(&[("w", &[1], &bf16)]);
let (_dir, in_path) = write_temp(&st_bytes, "safetensors");
let output = Command::new(binary_path())
.args(["convert", in_path.to_str().unwrap(), "--to", "frobnicate"])
.output()
.expect("run amn convert");
assert!(!output.status.success(), "should reject unknown target");
let stderr = String::from_utf8_lossy(&output.stderr);
assert!(
stderr.contains("supported convert targets"),
"expected target-list error, got stderr: {stderr}"
);
}