use std::collections::{BTreeMap, HashMap};
use std::fs;
use std::path::{Path, PathBuf};
use std::time::Instant;
use crate::convert::{self, Registry};
use crate::detect::{self, Detection, FormatTable, UNKNOWN_FORMAT};
use crate::hash::{self, CanonicalArtifact, DedupIndex};
use crate::manifest::{self, ArtifactKind, ManifestSchema, ManifestWriter, Record, Status};
use crate::mirror;
use crate::report::{
DivisionStatus, ExplainEntry, FormatCount, RunReport, ScanReport, SizeOutlier, StatusCounts,
StatusReport,
};
use crate::segments;
use crate::walk::{self, EntryKind, WalkOptions};
use crate::{Error, Result};
pub const SYMLINK_FORMAT: &str = "symlink";
pub const SPECIAL_FORMAT: &str = "special";
pub struct Rules {
pub table: FormatTable,
pub registry: Registry,
}
impl Rules {
pub fn builtin() -> Result<Self> {
let table = FormatTable::builtin()?;
let registry = Registry::builtin()?;
if table.version() != registry.version() {
return Err(Error::Rules {
name: "rules".to_string(),
message: format!(
"formats.toml version {:?} does not match converters.toml version {:?}",
table.version(),
registry.version()
),
});
}
Ok(Rules { table, registry })
}
pub fn version(&self) -> &str {
self.table.version()
}
}
pub struct RunOptions<'a> {
pub root: &'a Path,
pub mirror_root: &'a Path,
pub manifest_dir: &'a Path,
pub division: &'a str,
pub walk: WalkOptions,
}
pub fn scan(root: &Path, rules: &Rules, options: &WalkOptions) -> Result<ScanReport> {
let entries = walk::walk_division(root, options)?;
let mut inventory: Vec<(String, u64, String)> = Vec::new();
for entry in &entries {
let absolute = root.join(&entry.path);
let (size, format) = match entry.kind {
EntryKind::File => {
let size = absolute
.metadata()
.map_err(|e| Error::io("stat", &absolute, e))?
.len();
let detection = detect::detect_file(&absolute, &rules.table)?;
(size, detection.detected)
}
EntryKind::Symlink => {
let size = fs::symlink_metadata(&absolute)
.map_err(|e| Error::io("stat", &absolute, e))?
.len();
(size, SYMLINK_FORMAT.to_string())
}
EntryKind::Other => (0, SPECIAL_FORMAT.to_string()),
};
inventory.push((entry.path.to_string_lossy().into_owned(), size, format));
}
let mut by_format: BTreeMap<String, (u64, u64)> = BTreeMap::new();
let mut total_bytes = 0u64;
for (_, size, format) in &inventory {
let entry = by_format.entry(format.clone()).or_insert((0, 0));
entry.0 += 1;
entry.1 += size;
total_bytes += size;
}
let mean = if inventory.is_empty() {
0.0
} else {
total_bytes as f64 / inventory.len() as f64
};
let mut outliers: Vec<SizeOutlier> = inventory
.iter()
.filter(|(_, size, _)| *size >= 1_048_576 && *size as f64 >= 10.0 * mean)
.map(|(path, size, format)| SizeOutlier {
path: path.clone(),
bytes: *size,
format: format.clone(),
})
.collect();
outliers.sort_by(|a, b| b.bytes.cmp(&a.bytes).then(a.path.cmp(&b.path)));
Ok(ScanReport {
root: root.display().to_string(),
rules_version: rules.version().to_string(),
files: inventory.len() as u64,
bytes: total_bytes,
formats: by_format
.into_iter()
.map(|(format, (files, bytes))| FormatCount {
format,
files,
bytes,
})
.collect(),
outliers,
})
}
pub(crate) fn validate_division(name: &str) -> Result<()> {
let usable = !name.is_empty()
&& name
.chars()
.all(|c| c.is_ascii_alphanumeric() || c == '-' || c == '_');
if usable {
Ok(())
} else {
Err(Error::Division {
name: name.to_string(),
})
}
}
fn resolve_for_check(path: &Path) -> Result<PathBuf> {
use std::path::Component;
let absolute = std::path::absolute(path).map_err(|e| Error::io("absolutize", path, e))?;
let mut resolved = PathBuf::new();
for component in absolute.components() {
match component {
Component::Prefix(_) | Component::RootDir => resolved.push(component),
Component::CurDir => {}
Component::ParentDir => {
resolved.pop();
}
Component::Normal(name) => {
resolved.push(name);
if let Ok(canonical) = resolved.canonicalize() {
resolved = canonical;
}
}
}
}
Ok(resolved)
}
fn check_layout(root: &Path, mirror_root: &Path, manifest_dir: &Path) -> Result<()> {
let root_resolved = resolve_for_check(root)?;
for (name, dir) in [
("mirror root", mirror_root),
("manifest directory", manifest_dir),
] {
let dir_resolved = resolve_for_check(dir)?;
if dir_resolved.starts_with(&root_resolved) {
return Err(Error::Layout {
message: format!(
"{name} {} lies inside the division root {}",
dir.display(),
root.display()
),
});
}
if root_resolved.starts_with(&dir_resolved) {
return Err(Error::Layout {
message: format!(
"division root {} lies inside the {name} {}",
root.display(),
dir.display()
),
});
}
}
fs::create_dir_all(mirror_root).map_err(|e| Error::io("create_dir", mirror_root, e))?;
fs::create_dir_all(manifest_dir).map_err(|e| Error::io("create_dir", manifest_dir, e))?;
Ok(())
}
fn new_record(
source_path: &str,
source_hash: &str,
source_size: u64,
detection: &Detection,
rules_version: &str,
status: Status,
) -> Record {
Record {
schema: ManifestSchema,
source_path: source_path.to_string(),
source_hash: source_hash.to_string(),
source_size,
declared_format: detection.declared.clone(),
detected_format: detection.detected.clone(),
format_mismatch: detection.mismatch,
status,
text_path: None,
text_hash: None,
artifact_kind: None,
converter_id: None,
converter_version: None,
tool_version: None,
rules_version: rules_version.to_string(),
media: None,
parent_source: None,
dedup_of: None,
warnings: Vec::new(),
error: None,
duration_ms: None,
}
}
fn elapsed_ms(started: Instant) -> u64 {
u64::try_from(started.elapsed().as_millis()).unwrap_or(u64::MAX)
}
fn terminal_by_path(records: Vec<Record>) -> HashMap<String, Record> {
let mut terminal = HashMap::new();
for record in records {
terminal.insert(record.source_path.clone(), record);
}
terminal
}
fn seed_dedup(terminal: &HashMap<String, Record>, rules: &Rules, dedup: &mut DedupIndex) {
let mut candidates: Vec<&Record> = terminal.values().collect();
candidates.sort_by(|a, b| a.source_path.cmp(&b.source_path));
for record in candidates {
let canonical_status = record.status == Status::Converted
|| (record.status == Status::SkippedUnchanged && record.dedup_of.is_none());
if !canonical_status || record.rules_version != rules.version() {
continue;
}
let (Some(text_path), Some(text_hash), Some(converter_id), Some(converter_version)) = (
&record.text_path,
&record.text_hash,
&record.converter_id,
&record.converter_version,
) else {
continue;
};
let current = rules.registry.converter_for(&record.detected_format);
let versions_match =
current.is_some_and(|c| c.id() == converter_id && c.version() == converter_version);
if !versions_match {
continue;
}
dedup.insert(
&record.source_hash,
CanonicalArtifact {
source_path: record.source_path.clone(),
text_path: text_path.clone(),
text_hash: text_hash.clone(),
converter_id: converter_id.clone(),
converter_version: converter_version.clone(),
artifact_kind: record.artifact_kind,
},
);
}
}
fn record_outcome(
writer: &mut ManifestWriter,
terminal: &mut HashMap<String, Record>,
counts: &mut StatusCounts,
record: Record,
) -> Result<()> {
writer.append(&record)?;
match record.status {
Status::Converted => counts.converted += 1,
Status::Failed => counts.failed += 1,
Status::Unsupported => counts.unsupported += 1,
Status::SkippedUnchanged => counts.skipped_unchanged += 1,
Status::Dedup => counts.dedup += 1,
}
terminal.insert(record.source_path.clone(), record);
Ok(())
}
fn remove_stale_artifact(text_absolute: &Path, segments_absolute: &Path, record: &mut Record) {
for stale in [text_absolute, segments_absolute] {
match fs::remove_file(stale) {
Ok(()) => {}
Err(e)
if e.kind() == std::io::ErrorKind::NotFound
|| e.kind() == std::io::ErrorKind::NotADirectory => {}
Err(e) => record
.warnings
.push(format!("stale_artifact_not_removed: {e}")),
}
}
}
fn envelope_intact(artifact: &Path, text_hash: &str, segments_path: &Path) -> bool {
let Ok(text) = fs::read_to_string(artifact) else {
return false;
};
if hash::hash_bytes(text.as_bytes()) != text_hash {
return false;
}
let Ok(sidecar) = fs::read_to_string(segments_path) else {
return false;
};
segments::parse_jsonl(&sidecar).is_ok_and(|segs| segments::validate(&segs, &text).is_ok())
}
fn source_front(
absolute: &Path,
table: &FormatTable,
) -> Result<(String, u64, Detection, Vec<String>)> {
let source_hash = hash::hash_file(absolute)?;
let source_size = absolute
.metadata()
.map_err(|e| Error::io("stat", absolute, e))?
.len();
let (detection, warnings) = detect::detect_file_with_warnings(absolute, table)?;
Ok((source_hash, source_size, detection, warnings))
}
pub fn run(rules: &Rules, options: &RunOptions) -> Result<RunReport> {
validate_division(options.division)?;
check_layout(options.root, options.mirror_root, options.manifest_dir)?;
let entries = walk::walk_division(options.root, &options.walk)?;
let shard_path = options
.manifest_dir
.join(format!("{}.jsonl", options.division));
let (prior, mut warnings) = if shard_path.exists() {
let shard = manifest::read_shard(&shard_path)?;
(shard.records, shard.warnings)
} else {
(Vec::new(), Vec::new())
};
let mut terminal = terminal_by_path(prior);
let mut dedup = DedupIndex::default();
seed_dedup(&terminal, rules, &mut dedup);
let mut writer = ManifestWriter::open(&shard_path)?;
let mut counts = StatusCounts::default();
let mut special_entries = 0u64;
for entry in &entries {
let started = Instant::now();
let relative = &entry.path;
let absolute = options.root.join(relative);
let text_absolute = mirror::mirror_path(options.mirror_root, options.division, relative);
let segments_absolute =
segments::segments_path(options.mirror_root, options.division, relative);
if entry.kind == EntryKind::Other {
special_entries += 1;
continue;
}
let Some(source_path) = relative.to_str().map(str::to_string) else {
let detection = Detection {
declared: detect::declared_format(relative, &rules.table),
detected: UNKNOWN_FORMAT.to_string(),
mismatch: false,
};
let mut record = new_record(
&relative.to_string_lossy(),
"",
0,
&detection,
rules.version(),
Status::Failed,
);
record.error = Some("non_utf8_path: source path is not valid UTF-8".to_string());
record.duration_ms = Some(elapsed_ms(started));
remove_stale_artifact(&text_absolute, &segments_absolute, &mut record);
record_outcome(&mut writer, &mut terminal, &mut counts, record)?;
continue;
};
if entry.kind == EntryKind::Symlink {
let detection = Detection {
declared: None,
detected: SYMLINK_FORMAT.to_string(),
mismatch: false,
};
let mut record = match fs::read_link(&absolute) {
Ok(target) => {
let target_bytes = target.as_os_str().as_encoded_bytes();
let mut record = new_record(
&source_path,
&hash::hash_bytes(target_bytes),
target_bytes.len() as u64,
&detection,
rules.version(),
Status::Unsupported,
);
record.error = Some(convert::NO_CONVERTER_REASON.to_string());
record
}
Err(e) => {
let mut record = new_record(
&source_path,
"",
0,
&detection,
rules.version(),
Status::Failed,
);
record.error = Some(format!("source_io_error: readlink: {e}"));
record
}
};
record.duration_ms = Some(elapsed_ms(started));
remove_stale_artifact(&text_absolute, &segments_absolute, &mut record);
record_outcome(&mut writer, &mut terminal, &mut counts, record)?;
continue;
}
let front = match source_front(&absolute, &rules.table) {
Ok(front) => front,
Err(e) => {
let detection = Detection {
declared: detect::declared_format(relative, &rules.table),
detected: UNKNOWN_FORMAT.to_string(),
mismatch: false,
};
let mut record = new_record(
&source_path,
"",
0,
&detection,
rules.version(),
Status::Failed,
);
record.error = Some(format!("source_io_error: {e}"));
record.duration_ms = Some(elapsed_ms(started));
remove_stale_artifact(&text_absolute, &segments_absolute, &mut record);
record_outcome(&mut writer, &mut terminal, &mut counts, record)?;
continue;
}
};
let (source_hash, source_size, detection, detect_warnings) = front;
let converter = rules.registry.converter_for(&detection.detected);
let current_version = converter.map(|c| c.version());
if let Some(previous) = terminal.get(&source_path) {
let key_matches = previous.source_hash == source_hash
&& previous.rules_version == rules.version()
&& previous.converter_version.as_deref() == current_version;
let has_text = matches!(
previous.status,
Status::Converted | Status::Dedup | Status::SkippedUnchanged
);
if key_matches && has_text {
let intact = match (&previous.text_path, &previous.text_hash) {
(Some(text_path), Some(text_hash)) => {
mirror::resolve_recorded_path(options.mirror_root, text_path).is_ok_and(
|artifact| envelope_intact(&artifact, text_hash, &segments_absolute),
)
}
_ => false,
};
if intact {
let mut record = previous.clone();
record.status = Status::SkippedUnchanged;
record.error = None;
record.duration_ms = Some(elapsed_ms(started));
record_outcome(&mut writer, &mut terminal, &mut counts, record)?;
continue;
}
}
}
let text_relative = mirror::recorded_text_path(options.division, &source_path);
if let Some(canonical) = dedup.get(&source_hash)
&& canonical.source_path != source_path
&& converter.is_some_and(|c| {
c.id() == canonical.converter_id && c.version() == canonical.converter_version
})
{
let canonical = canonical.clone();
let canonical_segments = segments::segments_path(
options.mirror_root,
options.division,
Path::new(&canonical.source_path),
);
if let Ok(canonical_text) =
mirror::resolve_recorded_path(options.mirror_root, &canonical.text_path)
&& let (Ok(text), Ok(segment_lines)) = (
fs::read_to_string(&canonical_text),
fs::read_to_string(&canonical_segments),
)
&& hash::hash_bytes(text.as_bytes()) == canonical.text_hash
&& segments::parse_jsonl(&segment_lines)
.is_ok_and(|segs| segments::validate(&segs, &text).is_ok())
{
let mut record = new_record(
&source_path,
&source_hash,
source_size,
&detection,
rules.version(),
Status::Dedup,
);
record.warnings.clone_from(&detect_warnings);
match mirror::write_atomic(&text_absolute, &text)
.and_then(|()| mirror::write_atomic(&segments_absolute, &segment_lines))
{
Ok(()) => {
record.text_path = Some(text_relative);
record.text_hash = Some(hash::hash_bytes(text.as_bytes()));
record.artifact_kind = canonical.artifact_kind;
record.converter_id = Some(canonical.converter_id.clone());
record.converter_version = Some(canonical.converter_version.clone());
record.dedup_of = Some(canonical.source_path.clone());
}
Err(e) => {
record.status = Status::Failed;
record.error = Some(format!("mirror_write_error: {e}"));
remove_stale_artifact(&text_absolute, &segments_absolute, &mut record);
}
}
record.duration_ms = Some(elapsed_ms(started));
record_outcome(&mut writer, &mut terminal, &mut counts, record)?;
continue;
}
}
let Some(converter) = converter else {
let mut record = new_record(
&source_path,
&source_hash,
source_size,
&detection,
rules.version(),
Status::Unsupported,
);
record.warnings.clone_from(&detect_warnings);
record.error = rules
.registry
.unsupported_reason(&detection.detected)
.map(str::to_string);
record.duration_ms = Some(elapsed_ms(started));
remove_stale_artifact(&text_absolute, &segments_absolute, &mut record);
record_outcome(&mut writer, &mut terminal, &mut counts, record)?;
continue;
};
let mut record = new_record(
&source_path,
&source_hash,
source_size,
&detection,
rules.version(),
Status::Converted,
);
record.warnings.clone_from(&detect_warnings);
let outcome = if source_size > convert::MAX_SOURCE_BYTES {
Err(format!(
"resource_limit: source is {source_size} bytes over the {} byte ceiling",
convert::MAX_SOURCE_BYTES
))
} else {
fs::read(&absolute)
.map_err(|e| format!("source_io_error: read: {e}"))
.and_then(|bytes| {
if hash::hash_bytes(&bytes) != source_hash {
return Err(
"source_changed_during_run: bytes changed between hash and conversion"
.to_string(),
);
}
let guarded = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
converter.convert(&bytes, &detection.detected)
}));
match guarded {
Ok(result) => result.map_err(|e| e.to_string()),
Err(_) => Err("converter_panic: conversion panicked".to_string()),
}
})
.and_then(|outcome| {
if outcome.text.len() > convert::MAX_OUTPUT_BYTES {
return Err(format!(
"resource_limit: output is {} bytes over the {} byte ceiling",
outcome.text.len(),
convert::MAX_OUTPUT_BYTES
));
}
segments::validate(&outcome.segments, &outcome.text)
.map_err(|detail| format!("invalid_segments: {detail}"))?;
Ok(outcome)
})
};
match outcome {
Ok(outcome) => match segments::to_jsonl(&outcome.segments)
.map_err(|e| e.to_string())
.and_then(|lines| {
mirror::write_atomic(&text_absolute, &outcome.text)
.and_then(|()| mirror::write_atomic(&segments_absolute, &lines))
.map_err(|e| e.to_string())
}) {
Ok(()) => {
let text_hash = hash::hash_bytes(outcome.text.as_bytes());
record.text_path = Some(text_relative.clone());
record.text_hash = Some(text_hash.clone());
record.artifact_kind = Some(ArtifactKind::Text);
record.converter_id = Some(outcome.converter_id.clone());
record.converter_version = Some(outcome.converter_version.clone());
record.warnings.extend(outcome.warnings);
dedup.replace(
&source_hash,
CanonicalArtifact {
source_path: source_path.clone(),
text_path: text_relative,
text_hash,
converter_id: outcome.converter_id,
converter_version: outcome.converter_version,
artifact_kind: record.artifact_kind,
},
);
}
Err(detail) => {
record.status = Status::Failed;
record.converter_id = Some(converter.id().to_string());
record.converter_version = Some(converter.version().to_string());
record.error = Some(format!("mirror_write_error: {detail}"));
remove_stale_artifact(&text_absolute, &segments_absolute, &mut record);
}
},
Err(reason) => {
record.status = Status::Failed;
record.converter_id = Some(converter.id().to_string());
record.converter_version = Some(converter.version().to_string());
record.error = Some(reason);
remove_stale_artifact(&text_absolute, &segments_absolute, &mut record);
}
}
record.duration_ms = Some(elapsed_ms(started));
record_outcome(&mut writer, &mut terminal, &mut counts, record)?;
}
warnings.sort();
Ok(RunReport {
division: options.division.to_string(),
root: options.root.display().to_string(),
sources: entries.len() as u64 - special_entries,
special_entries,
counts,
warnings,
})
}
fn shard_paths(manifest_dir: &Path) -> Result<Vec<PathBuf>> {
let entries = fs::read_dir(manifest_dir).map_err(|e| Error::io("read_dir", manifest_dir, e))?;
let mut shards = Vec::new();
for entry in entries {
let entry = entry.map_err(|e| Error::io("read_dir", manifest_dir, e))?;
let path = entry.path();
if path.extension().and_then(|e| e.to_str()) == Some("jsonl") {
shards.push(path);
}
}
shards.sort();
Ok(shards)
}
fn division_name(shard: &Path) -> String {
shard
.file_stem()
.map(|stem| stem.to_string_lossy().into_owned())
.unwrap_or_default()
}
pub fn status(manifest_dir: &Path) -> Result<StatusReport> {
let mut divisions = Vec::new();
for shard_path in shard_paths(manifest_dir)? {
let shard = manifest::read_shard(&shard_path)?;
let terminal = terminal_by_path(shard.records);
let mut counts = StatusCounts::default();
let mut with_text = 0u64;
for record in terminal.values() {
match record.status {
Status::Converted => counts.converted += 1,
Status::Failed => counts.failed += 1,
Status::Unsupported => counts.unsupported += 1,
Status::SkippedUnchanged => counts.skipped_unchanged += 1,
Status::Dedup => counts.dedup += 1,
}
if record.text_path.is_some() {
with_text += 1;
}
}
let sources = terminal.len() as u64;
let coverage = if sources == 0 {
0.0
} else {
with_text as f64 / sources as f64
};
divisions.push(DivisionStatus {
division: division_name(&shard_path),
sources,
with_text,
coverage,
counts,
warnings: shard.warnings,
});
}
Ok(StatusReport { divisions })
}
pub fn explain(manifest_dir: &Path, source_path: &str) -> Result<Vec<ExplainEntry>> {
let mut entries = Vec::new();
for shard_path in shard_paths(manifest_dir)? {
let division = division_name(&shard_path);
for record in manifest::read_shard(&shard_path)?.records {
if record.source_path == source_path {
entries.push(ExplainEntry {
division: division.clone(),
record,
});
}
}
}
Ok(entries)
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn builtin_rules_agree_on_a_version() {
let rules = Rules::builtin().unwrap();
assert_eq!(rules.version(), "4");
}
use crate::convert::{ConvertError, Converter, Outcome, PlainTextPassthrough};
use crate::segments::Segment;
struct PanicConverter;
impl Converter for PanicConverter {
fn id(&self) -> &'static str {
"panic-test"
}
fn version(&self) -> &'static str {
"1.0.0"
}
fn convert(
&self,
_source: &[u8],
_detected_format: &str,
) -> std::result::Result<Outcome, ConvertError> {
panic!("deliberate test panic");
}
}
struct BadSegmentsConverter;
impl Converter for BadSegmentsConverter {
fn id(&self) -> &'static str {
"bad-segments-test"
}
fn version(&self) -> &'static str {
"1.0.0"
}
fn convert(
&self,
_source: &[u8],
detected_format: &str,
) -> std::result::Result<Outcome, ConvertError> {
Ok(Outcome {
converter_id: self.id().to_string(),
converter_version: self.version().to_string(),
detected_format: detected_format.to_string(),
text: "ok".to_string(),
warnings: Vec::new(),
segments: vec![Segment::span(0, 999, "document")],
})
}
}
fn injected_rules() -> Rules {
let table = FormatTable::parse(
r#"
version = "9"
[[formats]]
id = "text"
name = "Plain text"
extensions = ["txt"]
[[formats]]
id = "panicfmt"
name = "Panic trigger"
extensions = ["panicfmt"]
[[formats]]
id = "badseg"
name = "Bad segments trigger"
extensions = ["badseg"]
"#,
"formats.toml",
)
.unwrap();
let registry = Registry::for_tests(
"9",
vec![
(Box::new(PlainTextPassthrough), vec!["text"]),
(Box::new(PanicConverter), vec!["panicfmt"]),
(Box::new(BadSegmentsConverter), vec!["badseg"]),
],
);
Rules { table, registry }
}
#[test]
fn a_panicking_converter_fails_one_source_and_the_division_survives() {
let dir = tempfile::tempdir().unwrap();
let root = dir.path().join("src");
fs::create_dir_all(&root).unwrap();
fs::write(root.join("boom.panicfmt"), "trigger\n").unwrap();
fs::write(root.join("fine.txt"), "safe content\n").unwrap();
let rules = injected_rules();
let report = run(
&rules,
&RunOptions {
root: &root,
mirror_root: &dir.path().join("mirror"),
manifest_dir: &dir.path().join("manifest"),
division: "unit",
walk: WalkOptions::default(),
},
)
.unwrap();
assert_eq!(report.counts.failed, 1);
assert_eq!(report.counts.converted, 1);
let records = manifest::read_shard(&dir.path().join("manifest/unit.jsonl"))
.unwrap()
.records;
let failed = records.iter().find(|r| r.status == Status::Failed).unwrap();
assert_eq!(failed.source_path, "boom.panicfmt");
assert!(
failed
.error
.as_deref()
.unwrap()
.starts_with("converter_panic")
);
assert!(!dir.path().join("mirror/unit/boom.panicfmt.txt").exists());
}
#[test]
fn invalid_segments_from_a_converter_fail_before_any_write() {
let dir = tempfile::tempdir().unwrap();
let root = dir.path().join("src");
fs::create_dir_all(&root).unwrap();
fs::write(root.join("bad.badseg"), "trigger\n").unwrap();
let rules = injected_rules();
let report = run(
&rules,
&RunOptions {
root: &root,
mirror_root: &dir.path().join("mirror"),
manifest_dir: &dir.path().join("manifest"),
division: "unit",
walk: WalkOptions::default(),
},
)
.unwrap();
assert_eq!(report.counts.failed, 1);
let records = manifest::read_shard(&dir.path().join("manifest/unit.jsonl"))
.unwrap()
.records;
assert!(
records[0]
.error
.as_deref()
.unwrap()
.starts_with("invalid_segments")
);
assert!(!dir.path().join("mirror/unit/bad.badseg.txt").exists());
assert!(
!dir.path()
.join("mirror/unit/bad.badseg.segments.jsonl")
.exists()
);
}
#[test]
fn division_names_are_validated() {
assert!(validate_division("finance-eu_1").is_ok());
assert!(validate_division("").is_err());
assert!(validate_division("a/b").is_err());
assert!(validate_division("dot.dot").is_err());
}
#[test]
fn layout_rejects_outputs_inside_the_root() {
let dir = tempfile::tempdir().unwrap();
let root = dir.path().join("src");
fs::create_dir_all(&root).unwrap();
let outside = dir.path().join("out");
let err = check_layout(&root, &root.join("mirror"), &outside).unwrap_err();
assert!(err.to_string().contains("mirror root"));
let err = check_layout(&root, &outside.join("mirror"), &root.join("manifest")).unwrap_err();
assert!(err.to_string().contains("manifest directory"));
check_layout(&root, &outside.join("mirror"), &outside.join("manifest")).unwrap();
}
#[cfg(unix)]
#[test]
fn non_utf8_paths_are_rejected_without_an_artifact() {
use std::os::unix::ffi::OsStrExt;
let raw = std::ffi::OsStr::from_bytes(b"bad-\xff-name.txt");
assert!(Path::new(raw).to_str().is_none());
}
#[test]
fn identical_sources_dedup_to_one_conversion() {
let dir = tempfile::tempdir().unwrap();
let root = dir.path().join("src");
fs::create_dir_all(root.join("sub")).unwrap();
fs::write(root.join("first.txt"), "same bytes\n").unwrap();
fs::write(root.join("sub/second.txt"), "same bytes\n").unwrap();
let rules = Rules::builtin().unwrap();
let report = run(
&rules,
&RunOptions {
root: &root,
mirror_root: &dir.path().join("mirror"),
manifest_dir: &dir.path().join("manifest"),
division: "unit",
walk: WalkOptions::default(),
},
)
.unwrap();
assert_eq!(report.counts.converted, 1);
assert_eq!(report.counts.dedup, 1);
let records = manifest::read_shard(&dir.path().join("manifest/unit.jsonl"))
.unwrap()
.records;
let canonical = records
.iter()
.find(|r| r.source_path == "first.txt")
.unwrap();
let duplicate = records
.iter()
.find(|r| r.source_path == "sub/second.txt")
.unwrap();
assert_eq!(canonical.status, Status::Converted);
assert_eq!(duplicate.status, Status::Dedup);
assert_eq!(duplicate.dedup_of.as_deref(), Some("first.txt"));
assert_eq!(duplicate.text_hash, canonical.text_hash);
let mirrored =
fs::read_to_string(dir.path().join("mirror/unit/sub/second.txt.txt")).unwrap();
assert_eq!(mirrored, "same bytes\n");
}
}