use super::{gather, stop};
#[test]
fn r1_close_retains_primary_and_every_cleanup_failure() {
let closed = gather(
Err(anyhow::anyhow!("primary_failure")),
vec![
Err(anyhow::anyhow!("kill_failure")),
Err(anyhow::anyhow!("reap_failure")),
],
);
assert_eq!(closed.status, 1);
assert_eq!(
closed.errors,
["primary_failure", "kill_failure", "reap_failure"]
);
let closed = gather(Ok(143), vec![Err(anyhow::anyhow!("cleanup_failure"))]);
assert_eq!(closed.status, 143);
assert_eq!(closed.errors, ["cleanup_failure"]);
let closed = gather(Ok(0), vec![Err(anyhow::anyhow!("cleanup_failure"))]);
assert_eq!(closed.status, 1);
}
#[tokio::test]
async fn r1_close_awaits_cancelled_workers_and_their_drops() -> anyhow::Result<()> {
use std::sync::{
Arc,
atomic::{AtomicBool, Ordering},
};
struct Guard(Arc<AtomicBool>);
impl Drop for Guard {
fn drop(&mut self) {
self.0.store(true, Ordering::Release);
}
}
let dropped = Arc::new(AtomicBool::new(false));
let flag = Arc::clone(&dropped);
let (ready, running) = tokio::sync::oneshot::channel();
let mut task = tokio::spawn(async move {
let guard = Guard(flag);
if ready.send(()).is_err() {
return;
}
std::future::pending::<()>().await;
drop(guard);
});
running.await?;
stop(&mut task, "test_worker").await?;
assert!(
dropped.load(Ordering::Acquire),
"worker was aborted without awaiting its drop"
);
Ok(())
}
#[tokio::test]
async fn r1_close_names_panicked_workers() -> anyhow::Result<()> {
let (ready, running) = tokio::sync::oneshot::channel();
let mut task = tokio::spawn(async move {
assert!(ready.send(()).is_err(), "synthetic worker panic");
});
running.await?;
let error = stop(&mut task, "acp_reader")
.await
.err()
.ok_or_else(|| anyhow::anyhow!("worker panic was lost"))?;
assert_eq!(error.to_string(), "acp_reader_panicked");
Ok(())
}