108 lines
4.4 KiB
Rust
108 lines
4.4 KiB
Rust
//! Deterministic parent/index publication tests using an owned protocol fixture.
|
|
//! Genuine parser execution is covered separately in documents.rs.
|
|
use anyhow::Result;
|
|
use searchi_core::{
|
|
extraction_protocol::{DocumentFormat, RequestHeader},
|
|
IndexOptions, SearchIndex,
|
|
};
|
|
use std::{
|
|
fs,
|
|
sync::atomic::{AtomicBool, Ordering},
|
|
time::{Duration, Instant},
|
|
};
|
|
|
|
#[test]
|
|
fn mid_worker_same_stat_changes_and_cancellation_roll_back_index() -> Result<()> {
|
|
for cancel in [false, true] {
|
|
let directory = tempfile::tempdir()?;
|
|
let root = directory.path().join("files");
|
|
fs::create_dir(&root)?;
|
|
let source = root.join("contract.pdf");
|
|
fs::write(root.join("notes.txt"), "preserved evidence")?;
|
|
let mut index = SearchIndex::open(":memory:")?;
|
|
index.index_folder(&root)?;
|
|
fs::write(&source, b"original")?;
|
|
let initial = fs::metadata(&source)?;
|
|
let digest = RequestHeader::from_bytes(DocumentFormat::Pdf, b"original")?.input_sha256;
|
|
let marker = directory.path().join("admitted");
|
|
let release = directory.path().join("release");
|
|
let fixture_source = directory.path().join("worker.rs");
|
|
let executable = directory
|
|
.path()
|
|
.join(format!("fixture{}", std::env::consts::EXE_SUFFIX));
|
|
let response = serde_json::json!({
|
|
"version":1,"input_sha256":digest,
|
|
"outcome":{"status":"extracted","passages":[{"text":"stale document text","source":{"kind":"page","number":1}}]}
|
|
}).to_string();
|
|
fs::write(
|
|
&fixture_source,
|
|
format!(
|
|
r#"
|
|
use std::io::{{Read,Write}};
|
|
fn main() {{
|
|
let mut bytes=Vec::new(); std::io::stdin().read_to_end(&mut bytes).unwrap();
|
|
std::fs::write({marker:?}, b"admitted").unwrap();
|
|
while !std::path::Path::new({release:?}).exists() {{ std::thread::sleep(std::time::Duration::from_millis(5)); }}
|
|
let response={response:?}.as_bytes(); let mut out=std::io::stdout();
|
|
out.write_all(&(response.len() as u32).to_le_bytes()).unwrap(); out.write_all(response).unwrap();
|
|
}}
|
|
"#
|
|
),
|
|
)?;
|
|
let status = std::process::Command::new("rustc")
|
|
.arg(&fixture_source)
|
|
.arg("-o")
|
|
.arg(&executable)
|
|
.status()?;
|
|
anyhow::ensure!(status.success(), "Compile owned protocol fixture");
|
|
index.set_document_worker(executable);
|
|
let cancelled = AtomicBool::new(false);
|
|
std::thread::scope(|scope| {
|
|
let changer = scope.spawn(|| -> Result<()> {
|
|
let deadline = Instant::now() + Duration::from_secs(10);
|
|
while !marker.exists() && Instant::now() < deadline {
|
|
std::thread::sleep(Duration::from_millis(5));
|
|
}
|
|
if !marker.exists() {
|
|
cancelled.store(true, Ordering::Release);
|
|
anyhow::bail!("Worker did not admit fixture input");
|
|
}
|
|
if cancel {
|
|
cancelled.store(true, Ordering::Release);
|
|
} else {
|
|
fs::write(&source, b"modified")?;
|
|
fs::File::options()
|
|
.write(true)
|
|
.open(&source)?
|
|
.set_times(fs::FileTimes::new().set_modified(initial.modified()?))?;
|
|
assert_eq!(fs::metadata(&source)?.len(), initial.len());
|
|
assert_eq!(fs::metadata(&source)?.modified()?, initial.modified()?);
|
|
fs::write(&release, b"release")?;
|
|
}
|
|
Ok(())
|
|
});
|
|
let result =
|
|
index.index_folder_cancellable(&root, &IndexOptions::default(), &cancelled);
|
|
assert!(
|
|
result.is_err(),
|
|
"Mutation or cancellation must roll back publication"
|
|
);
|
|
let message = format!("{:#}", result.unwrap_err());
|
|
assert!(
|
|
message.contains(if cancel {
|
|
"cancelled"
|
|
} else {
|
|
"changed during extraction"
|
|
}),
|
|
"{message}"
|
|
);
|
|
changer.join().unwrap().unwrap();
|
|
});
|
|
assert_eq!(index.stats()?.files, 1);
|
|
assert_eq!(index.search("evidence", 10)?.len(), 1);
|
|
assert!(index.search("stale", 10)?.is_empty());
|
|
assert!(!index.contains_path(&source.to_string_lossy())?);
|
|
}
|
|
Ok(())
|
|
}
|