use std::collections::HashMap;
use std::path::{Path, PathBuf};
use std::sync::Mutex;
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use serde::{Deserialize, Serialize};
use crate::command::Command;
use crate::doubles::Invocation;
use crate::error::{Error, Result};
use crate::result::{Outcome, ProcessResult};
use crate::runner::{JobRunner, ProcessRunner};
const CASSETTE_VERSION: u32 = 4;
#[derive(Debug, Serialize, Deserialize)]
struct Cassette {
version: u32,
entries: Vec<Entry>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
struct Entry {
program: String,
args: Vec<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
stdin_digest: Option<u64>,
#[serde(default, skip_serializing_if = "Option::is_none")]
match_digest: Option<u64>,
#[serde(default, skip_serializing_if = "Option::is_none")]
cwd: Option<String>,
#[serde(default, skip_serializing_if = "is_false")]
has_stdin: bool,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
env_names: Vec<String>,
stdout: String,
stderr: String,
code: Option<i32>,
#[serde(default, skip_serializing_if = "is_false")]
timed_out: bool,
#[serde(default, skip_serializing_if = "Option::is_none")]
signal: Option<i32>,
#[serde(default, skip_serializing_if = "is_false")]
truncated: bool,
#[serde(default, skip_serializing_if = "is_zero_usize")]
total_lines: usize,
#[serde(default, skip_serializing_if = "is_zero_usize")]
total_bytes: usize,
#[serde(default, skip_serializing_if = "is_zero_u64")]
duration_ms: u64,
#[serde(default, skip_serializing_if = "Option::is_none")]
error: Option<CassetteError>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(tag = "kind")]
enum CassetteError {
Spawn {
os_kind: String,
message: String,
},
NotFound {
searched: Option<String>,
},
Stdin {
os_kind: String,
message: String,
},
OutputTooLarge {
max_lines: Option<usize>,
max_bytes: Option<usize>,
total_lines: usize,
total_bytes: usize,
},
Unsupported {
operation: String,
},
Io {
os_kind: String,
message: String,
},
Other {
message: String,
},
}
impl CassetteError {
fn from_error(err: &Error) -> Option<Self> {
Some(match err {
Error::Cancelled { .. } => return None,
Error::Spawn { source, .. } => CassetteError::Spawn {
os_kind: io_kind_name(source.kind()).to_owned(),
message: source.to_string(),
},
Error::NotFound { searched, .. } => CassetteError::NotFound {
searched: searched.clone(),
},
Error::Stdin { source, .. } => CassetteError::Stdin {
os_kind: io_kind_name(source.kind()).to_owned(),
message: source.to_string(),
},
Error::OutputTooLarge {
max_lines,
max_bytes,
total_lines,
total_bytes,
..
} => CassetteError::OutputTooLarge {
max_lines: *max_lines,
max_bytes: *max_bytes,
total_lines: *total_lines,
total_bytes: *total_bytes,
},
Error::Unsupported { operation } => CassetteError::Unsupported {
operation: operation.clone(),
},
Error::Io(source) => CassetteError::Io {
os_kind: io_kind_name(source.kind()).to_owned(),
message: source.to_string(),
},
other => CassetteError::Other {
message: other.to_string(),
},
})
}
fn to_error(&self, program: &str) -> Error {
match self {
CassetteError::Spawn { os_kind, message } => Error::Spawn {
program: program.to_owned(),
source: std::io::Error::new(io_kind_from_name(os_kind), message.clone()),
},
CassetteError::NotFound { searched } => Error::NotFound {
program: program.to_owned(),
searched: searched.clone(),
},
CassetteError::Stdin { os_kind, message } => Error::Stdin {
program: program.to_owned(),
source: std::io::Error::new(io_kind_from_name(os_kind), message.clone()),
},
CassetteError::OutputTooLarge {
max_lines,
max_bytes,
total_lines,
total_bytes,
} => Error::OutputTooLarge {
program: program.to_owned(),
max_lines: *max_lines,
max_bytes: *max_bytes,
total_lines: *total_lines,
total_bytes: *total_bytes,
},
CassetteError::Unsupported { operation } => Error::Unsupported {
operation: operation.clone(),
},
CassetteError::Io { os_kind, message } => Error::Io(std::io::Error::new(
io_kind_from_name(os_kind),
message.clone(),
)),
CassetteError::Other { message } => Error::Io(std::io::Error::other(message.clone())),
}
}
}
fn io_kind_name(kind: std::io::ErrorKind) -> &'static str {
use std::io::ErrorKind as K;
match kind {
K::NotFound => "NotFound",
K::PermissionDenied => "PermissionDenied",
K::Interrupted => "Interrupted",
K::WouldBlock => "WouldBlock",
K::InvalidInput => "InvalidInput",
K::InvalidData => "InvalidData",
K::TimedOut => "TimedOut",
K::WriteZero => "WriteZero",
K::UnexpectedEof => "UnexpectedEof",
K::ResourceBusy => "ResourceBusy",
K::ExecutableFileBusy => "ExecutableFileBusy",
K::NotADirectory => "NotADirectory",
K::BrokenPipe => "BrokenPipe",
K::AlreadyExists => "AlreadyExists",
_ => "Other",
}
}
fn io_kind_from_name(name: &str) -> std::io::ErrorKind {
use std::io::ErrorKind as K;
match name {
"NotFound" => K::NotFound,
"PermissionDenied" => K::PermissionDenied,
"Interrupted" => K::Interrupted,
"WouldBlock" => K::WouldBlock,
"InvalidInput" => K::InvalidInput,
"InvalidData" => K::InvalidData,
"TimedOut" => K::TimedOut,
"WriteZero" => K::WriteZero,
"UnexpectedEof" => K::UnexpectedEof,
"ResourceBusy" => K::ResourceBusy,
"ExecutableFileBusy" => K::ExecutableFileBusy,
"NotADirectory" => K::NotADirectory,
"BrokenPipe" => K::BrokenPipe,
"AlreadyExists" => K::AlreadyExists,
_ => K::Other,
}
}
struct KeyFields {
program: String,
args: Vec<String>,
cwd: Option<String>,
stdin_digest: Option<u64>,
has_stdin: bool,
env_names: Vec<String>,
}
#[derive(Debug, Clone, Default)]
struct MatchPolicy {
match_cwd: bool,
env_names: Vec<String>,
}
impl MatchPolicy {
fn is_empty(&self) -> bool {
!self.match_cwd && self.env_names.is_empty()
}
fn digest_of(&self, invocation: &Invocation) -> Option<u64> {
if self.is_empty() {
return None;
}
const OFFSET: u64 = 0xcbf2_9ce4_8422_2325;
const PRIME: u64 = 0x0000_0100_0000_01b3;
fn mix(mut h: u64, bytes: &[u8]) -> u64 {
for &b in bytes {
h ^= u64::from(b);
h = h.wrapping_mul(PRIME);
}
h
}
let mut h = OFFSET;
if self.match_cwd {
h = mix(h, b"cwd\0");
match &invocation.cwd {
Some(cwd) => {
h = mix(h, &[1]);
h = mix(h, cwd.as_os_str().as_encoded_bytes());
}
None => h = mix(h, &[0]),
}
}
for name in &self.env_names {
h = mix(h, b"env\0");
h = mix(h, name.as_bytes());
h = mix(h, &[0]); match invocation.env(name) {
Some(Some(value)) => {
h = mix(h, &[1]); h = mix(h, value.as_encoded_bytes());
}
Some(None) => h = mix(h, &[2]), None => h = mix(h, &[0]), }
}
Some(h)
}
}
#[allow(clippy::trivially_copy_pass_by_ref)] fn is_false(b: &bool) -> bool {
!*b
}
#[allow(clippy::trivially_copy_pass_by_ref)] fn is_zero_usize(n: &usize) -> bool {
*n == 0
}
#[allow(clippy::trivially_copy_pass_by_ref)] fn is_zero_u64(n: &u64) -> bool {
*n == 0
}
fn write_cassette(path: &Path, json: &str) -> std::io::Result<()> {
#[cfg(unix)]
if std::fs::symlink_metadata(path).is_ok_and(|m| m.file_type().is_symlink()) {
return Err(std::io::Error::from_raw_os_error(libc::ELOOP));
}
let _lock = acquire_save_lock(path)?;
let mut tmp = tmp_sibling(path);
let mut attempts = 0u32;
loop {
match write_new_file(&tmp, json) {
Ok(()) => break,
Err(e) if e.kind() == std::io::ErrorKind::AlreadyExists && attempts < 16 => {
attempts += 1;
tmp = tmp_sibling(path);
}
Err(e) => return Err(e),
}
}
match std::fs::rename(&tmp, path).and_then(|()| sync_parent_dir(path)) {
Ok(()) => Ok(()),
Err(e) => {
let _ = std::fs::remove_file(&tmp);
Err(e)
}
}
}
fn tmp_sibling(path: &Path) -> std::path::PathBuf {
static COUNTER: AtomicU64 = AtomicU64::new(0);
let seq = COUNTER.fetch_add(1, Ordering::Relaxed);
let nanos = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map_or(0, |d| d.as_nanos());
let mut name = path.as_os_str().to_owned();
name.push(format!(".{}.{}.{}.tmp", std::process::id(), seq, nanos));
std::path::PathBuf::from(name)
}
fn write_new_file(path: &Path, json: &str) -> std::io::Result<()> {
#[cfg(unix)]
{
use std::io::Write;
use std::os::unix::fs::{OpenOptionsExt, PermissionsExt};
let mut file = std::fs::OpenOptions::new()
.write(true)
.create_new(true)
.mode(0o600)
.custom_flags(libc::O_NOFOLLOW)
.open(path)?;
file.set_permissions(std::fs::Permissions::from_mode(0o600))?;
file.write_all(json.as_bytes())?;
file.sync_all()?; Ok(())
}
#[cfg(not(unix))]
{
use std::io::Write;
let mut file = std::fs::OpenOptions::new()
.write(true)
.create_new(true)
.open(path)?;
file.write_all(json.as_bytes())?;
file.sync_all()?;
Ok(())
}
}
fn lock_sibling(path: &Path) -> std::path::PathBuf {
let mut name = path.as_os_str().to_owned();
name.push(".lock");
std::path::PathBuf::from(name)
}
fn concurrent_save_conflict() -> std::io::Error {
std::io::Error::new(
std::io::ErrorKind::WouldBlock,
"another writer is saving this cassette concurrently — concurrent saves to \
one cassette path are serialized by an advisory lock and the loser is \
refused (a transient, retryable error) rather than silently overwriting \
the last good cassette",
)
}
struct SaveLock {
#[allow(dead_code)]
file: std::fs::File,
}
#[cfg(unix)]
fn acquire_save_lock(path: &Path) -> std::io::Result<SaveLock> {
use std::os::unix::fs::OpenOptionsExt;
use std::os::unix::io::AsRawFd;
let file = std::fs::OpenOptions::new()
.create(true)
.truncate(false)
.write(true)
.mode(0o600)
.custom_flags(libc::O_NOFOLLOW)
.open(lock_sibling(path))?;
if unsafe { libc::flock(file.as_raw_fd(), libc::LOCK_EX | libc::LOCK_NB) } != 0 {
let err = std::io::Error::last_os_error();
if err.kind() == std::io::ErrorKind::WouldBlock {
return Err(concurrent_save_conflict());
}
return Err(err);
}
Ok(SaveLock { file })
}
#[cfg(windows)]
fn acquire_save_lock(path: &Path) -> std::io::Result<SaveLock> {
use std::os::windows::fs::OpenOptionsExt;
const ERROR_SHARING_VIOLATION: i32 = 32;
match std::fs::OpenOptions::new()
.create(true)
.truncate(false)
.write(true)
.share_mode(0)
.open(lock_sibling(path))
{
Ok(file) => Ok(SaveLock { file }),
Err(err) if err.raw_os_error() == Some(ERROR_SHARING_VIOLATION) => {
Err(concurrent_save_conflict())
}
Err(err) => Err(err),
}
}
fn sync_parent_dir(path: &Path) -> std::io::Result<()> {
#[cfg(unix)]
{
let dir = match path.parent().filter(|p| !p.as_os_str().is_empty()) {
Some(parent) => std::fs::File::open(parent)?,
None => std::fs::File::open(".")?,
};
dir.sync_all()
}
#[cfg(not(unix))]
{
let _ = path;
Ok(())
}
}
impl Entry {
fn key_fields(invocation: &Invocation, stdin_digest: Option<u64>) -> KeyFields {
let mut env_names: Vec<String> = invocation
.envs
.iter()
.map(|(name, _value)| name.to_string_lossy().into_owned())
.collect();
env_names.sort();
env_names.dedup();
KeyFields {
program: invocation.program.to_string_lossy().into_owned(),
args: invocation
.args
.iter()
.map(|a| a.to_string_lossy().into_owned())
.collect(),
cwd: invocation
.cwd
.as_ref()
.map(|c| c.to_string_lossy().into_owned()),
stdin_digest,
has_stdin: invocation.has_stdin,
env_names,
}
}
fn from_parts(
invocation: &Invocation,
result: &ProcessResult<String>,
stdin_digest: Option<u64>,
match_digest: Option<u64>,
) -> Self {
let key = Self::key_fields(invocation, stdin_digest);
Self {
program: key.program,
args: key.args,
cwd: key.cwd,
stdin_digest: key.stdin_digest,
match_digest,
has_stdin: key.has_stdin,
env_names: key.env_names,
stdout: result.stdout().clone(),
stderr: result.stderr().to_owned(),
code: result.code(),
timed_out: result.timed_out(),
signal: match result.outcome() {
Outcome::Signalled(s) => s,
Outcome::Exited(_) | Outcome::TimedOut => None,
},
truncated: result.truncated(),
total_lines: result.total_lines(),
total_bytes: result.total_bytes(),
duration_ms: result.duration().as_millis() as u64,
error: None,
}
}
fn from_error(
invocation: &Invocation,
stdin_digest: Option<u64>,
match_digest: Option<u64>,
error: CassetteError,
) -> Self {
let key = Self::key_fields(invocation, stdin_digest);
Self {
program: key.program,
args: key.args,
cwd: key.cwd,
stdin_digest: key.stdin_digest,
match_digest,
has_stdin: key.has_stdin,
env_names: key.env_names,
stdout: String::new(),
stderr: String::new(),
code: None,
timed_out: false,
signal: None,
truncated: false,
total_lines: 0,
total_bytes: 0,
duration_ms: 0,
error: Some(error),
}
}
fn to_result(
&self,
timeout: Option<std::time::Duration>,
ok_codes: Vec<i32>,
) -> ProcessResult<String> {
let outcome = match (self.code, self.timed_out) {
(_, true) => Outcome::TimedOut,
(Some(code), false) => Outcome::Exited(code),
(None, false) => Outcome::Signalled(self.signal),
};
ProcessResult::new(
self.program.clone(),
self.stdout.clone(),
self.stderr.clone(),
outcome,
timeout,
)
.with_ok_codes(ok_codes)
.with_truncated(self.truncated)
.with_overflow_totals(self.total_lines, self.total_bytes)
.with_duration(std::time::Duration::from_millis(self.duration_ms))
}
}
type Key = (String, Vec<String>, bool, Option<u64>, Option<u64>);
fn stdin_digest_of(command: &Command) -> Option<u64> {
command
.stdin_source()
.filter(|s| !s.is_empty())
.map(|s| s.content_digest())
}
fn reject_unrecordable_stdin(command: &Command) -> Result<()> {
if command.stdin_source().is_some_and(|s| s.is_one_shot()) {
return Err(Error::Unsupported {
operation: "cassette record/replay with one-shot streaming stdin \
(from_reader/from_lines); use from_bytes/from_string/from_file"
.to_string(),
});
}
Ok(())
}
fn key_of(invocation: &Invocation, stdin_digest: Option<u64>, policy: &MatchPolicy) -> Key {
(
invocation.program.to_string_lossy().into_owned(),
invocation
.args
.iter()
.map(|a| a.to_string_lossy().into_owned())
.collect(),
invocation.has_stdin,
stdin_digest,
policy.digest_of(invocation),
)
}
fn key_of_entry(entry: &Entry) -> Key {
(
entry.program.clone(),
entry.args.clone(),
entry.has_stdin,
entry.stdin_digest,
entry.match_digest,
)
}
#[derive(Debug)]
struct ReplaySlot {
entries: Vec<Entry>,
next: usize,
}
impl ReplaySlot {
fn play(&mut self) -> &Entry {
let index = self.next.min(self.entries.len() - 1);
self.next = self.next.saturating_add(1);
&self.entries[index]
}
}
enum Mode<R> {
Record {
inner: R,
path: PathBuf,
recorded: Mutex<Vec<Entry>>,
dirty: AtomicBool,
},
Replay {
slots: Mutex<HashMap<Key, ReplaySlot>>,
},
}
pub struct RecordReplayRunner<R: ProcessRunner = JobRunner> {
mode: Mode<R>,
policy: MatchPolicy,
}
impl<R: ProcessRunner> RecordReplayRunner<R> {
pub fn record(path: impl Into<PathBuf>, inner: R) -> Self {
Self {
mode: Mode::Record {
inner,
path: path.into(),
recorded: Mutex::new(Vec::new()),
dirty: AtomicBool::new(false),
},
policy: MatchPolicy::default(),
}
}
#[must_use]
pub fn match_on_cwd(mut self) -> Self {
self.policy.match_cwd = true;
self
}
#[must_use]
pub fn match_on_env<I, S>(mut self, names: I) -> Self
where
I: IntoIterator<Item = S>,
S: Into<String>,
{
self.policy
.env_names
.extend(names.into_iter().map(Into::into));
self.policy.env_names.sort();
self.policy.env_names.dedup();
self
}
pub fn save(&self) -> Result<()> {
let Mode::Record {
path,
recorded,
dirty,
..
} = &self.mode
else {
return Ok(());
};
let entries = recorded.lock().expect("cassette mutex poisoned");
let cassette = Cassette {
version: CASSETTE_VERSION,
entries: entries.clone(),
};
let json = serde_json::to_string_pretty(&cassette)
.map_err(|e| Error::Io(std::io::Error::from(e)))?;
write_cassette(path, &json).map_err(Error::Io)?;
dirty.store(false, Ordering::SeqCst);
Ok(())
}
}
fn validate_entry_outcome(entry: &Entry) -> Result<()> {
let indicators = usize::from(entry.code.is_some())
+ usize::from(entry.timed_out)
+ usize::from(entry.signal.is_some());
if entry.error.is_some() {
if indicators > 0 {
return Err(Error::Io(std::io::Error::new(
std::io::ErrorKind::InvalidData,
format!(
"cassette entry for `{}` carries both a recorded `error` and an outcome \
indicator (`code`/`timed_out`/`signal`) — at most one may be set",
entry.program
),
)));
}
return Ok(());
}
if indicators > 1 {
return Err(Error::Io(std::io::Error::new(
std::io::ErrorKind::InvalidData,
format!(
"cassette entry for `{}` has a contradictory outcome: at most one of \
`code` (exited), `timed_out`, or `signal` (signalled) may be set — found {indicators}",
entry.program
),
)));
}
Ok(())
}
impl RecordReplayRunner<JobRunner> {
pub fn replay(path: impl AsRef<Path>) -> Result<Self> {
let path = path.as_ref();
const MAX_CASSETTE_BYTES: u64 = 64 << 20; if let Ok(meta) = std::fs::metadata(path)
&& meta.len() > MAX_CASSETTE_BYTES
{
return Err(Error::Io(std::io::Error::new(
std::io::ErrorKind::InvalidData,
format!(
"cassette is {} bytes, over the {MAX_CASSETTE_BYTES}-byte limit",
meta.len()
),
)));
}
let text = std::fs::read_to_string(path).map_err(Error::Io)?;
#[derive(Deserialize)]
struct CassetteHeader {
version: u32,
}
let header: CassetteHeader =
serde_json::from_str(&text).map_err(|e| Error::Io(std::io::Error::from(e)))?;
if header.version > CASSETTE_VERSION {
return Err(Error::Io(std::io::Error::new(
std::io::ErrorKind::InvalidData,
format!(
"cassette version {} is not supported (this build reads up to version {CASSETTE_VERSION})",
header.version
),
)));
}
let cassette: Cassette =
serde_json::from_str(&text).map_err(|e| Error::Io(std::io::Error::from(e)))?;
let mut slots: HashMap<Key, ReplaySlot> = HashMap::new();
for entry in cassette.entries {
validate_entry_outcome(&entry)?;
slots
.entry(key_of_entry(&entry))
.or_insert_with(|| ReplaySlot {
entries: Vec::new(),
next: 0,
})
.entries
.push(entry);
}
Ok(Self {
mode: Mode::Replay {
slots: Mutex::new(slots),
},
policy: MatchPolicy::default(),
})
}
}
#[async_trait::async_trait]
impl<R: ProcessRunner> ProcessRunner for RecordReplayRunner<R> {
async fn output_string(&self, command: &Command) -> Result<ProcessResult<String>> {
reject_unrecordable_stdin(command)?;
match &self.mode {
Mode::Record {
inner,
recorded,
dirty,
..
} => {
let invocation = Invocation::from_command(command);
let stdin_digest = stdin_digest_of(command);
let match_digest = self.policy.digest_of(&invocation);
match inner.output_string(command).await {
Ok(result) => {
let mut entries = recorded.lock().expect("cassette mutex poisoned");
entries.push(Entry::from_parts(
&invocation,
&result,
stdin_digest,
match_digest,
));
dirty.store(true, Ordering::SeqCst);
Ok(result)
}
Err(err) => {
if let Some(cassette_err) = CassetteError::from_error(&err) {
let mut entries = recorded.lock().expect("cassette mutex poisoned");
entries.push(Entry::from_error(
&invocation,
stdin_digest,
match_digest,
cassette_err,
));
dirty.store(true, Ordering::SeqCst);
}
Err(err)
}
}
}
Mode::Replay { slots } => {
if let Some(token) = command.cancel_token()
&& token.is_cancelled()
{
return Err(Error::Cancelled {
program: command.program_name(),
});
}
if !command.stdout_is_piped() {
return Err(crate::error::stdout_not_piped_error(
&command.program_name(),
));
}
let invocation = Invocation::from_command(command);
let stdin_digest = stdin_digest_of(command);
let entry = {
let mut slots = slots.lock().expect("cassette mutex poisoned");
let slot = match slots.get_mut(&key_of(&invocation, stdin_digest, &self.policy))
{
Some(slot) => slot,
None => {
return Err(Error::CassetteMiss {
program: command.program_name(),
});
}
};
slot.play().clone()
};
if let Some(cassette_err) = &entry.error {
return Err(cassette_err.to_error(&entry.program));
}
crate::doubles::replay_line_handlers(command, &entry.stdout, &entry.stderr);
Ok(entry.to_result(command.configured_timeout(), command.ok_codes_vec()))
}
}
}
async fn output_bytes(&self, _command: &Command) -> Result<ProcessResult<Vec<u8>>> {
Err(Error::Unsupported {
operation: "output_bytes on a cassette (a lossy-UTF-8 text fixture cannot \
reproduce exact bytes; capture them from a real or scripted runner)"
.to_string(),
})
}
async fn start(&self, command: &Command) -> Result<crate::RunningProcess> {
reject_unrecordable_stdin(command)?;
match &self.mode {
Mode::Record {
inner,
recorded,
dirty,
..
} => {
let invocation = Invocation::from_command(command);
let stdin_digest = stdin_digest_of(command);
let match_digest = self.policy.digest_of(&invocation);
match inner
.output_string(&command.without_line_side_effects())
.await
{
Ok(result) => {
let entry =
Entry::from_parts(&invocation, &result, stdin_digest, match_digest);
{
let mut entries = recorded.lock().expect("cassette mutex poisoned");
entries.push(entry.clone());
dirty.store(true, Ordering::SeqCst);
}
Ok(crate::doubles::scripted_running_from_parts(
command,
entry.stdout,
entry.stderr,
entry.code,
entry.timed_out,
entry.signal,
entry.truncated,
entry.total_lines,
entry.total_bytes,
std::time::Duration::from_millis(entry.duration_ms),
))
}
Err(err) => {
if let Some(cassette_err) = CassetteError::from_error(&err) {
let mut entries = recorded.lock().expect("cassette mutex poisoned");
entries.push(Entry::from_error(
&invocation,
stdin_digest,
match_digest,
cassette_err,
));
dirty.store(true, Ordering::SeqCst);
}
Err(err)
}
}
}
Mode::Replay { slots } => {
if let Some(token) = command.cancel_token()
&& token.is_cancelled()
{
return Err(Error::Cancelled {
program: command.program_name(),
});
}
let invocation = Invocation::from_command(command);
let stdin_digest = stdin_digest_of(command);
let entry = {
let mut slots = slots.lock().expect("cassette mutex poisoned");
let slot = match slots.get_mut(&key_of(&invocation, stdin_digest, &self.policy))
{
Some(slot) => slot,
None => {
return Err(Error::CassetteMiss {
program: command.program_name(),
});
}
};
slot.play().clone()
};
if let Some(cassette_err) = &entry.error {
return Err(cassette_err.to_error(&entry.program));
}
Ok(crate::doubles::scripted_running_from_parts(
command,
entry.stdout,
entry.stderr,
entry.code,
entry.timed_out,
entry.signal,
entry.truncated,
entry.total_lines,
entry.total_bytes,
std::time::Duration::from_millis(entry.duration_ms),
))
}
}
}
}
impl<R: ProcessRunner> std::fmt::Debug for RecordReplayRunner<R> {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match &self.mode {
Mode::Record {
path,
recorded,
dirty,
..
} => f
.debug_struct("RecordReplayRunner::Record")
.field("path", path)
.field(
"recorded",
&recorded.lock().expect("cassette mutex poisoned").len(),
)
.field("dirty", &dirty.load(Ordering::SeqCst))
.finish_non_exhaustive(),
Mode::Replay { slots } => f
.debug_struct("RecordReplayRunner::Replay")
.field(
"keys",
&slots.lock().expect("cassette mutex poisoned").len(),
)
.finish_non_exhaustive(),
}
}
}
impl<R: ProcessRunner> Drop for RecordReplayRunner<R> {
fn drop(&mut self) {
if let Mode::Record { dirty, .. } = &self.mode
&& dirty.load(Ordering::SeqCst)
&& !std::thread::panicking()
{
let _ = self.save();
}
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::doubles::{Reply, ScriptedRunner};
use crate::result::Outcome;
use crate::runner::ProcessRunnerExt;
use std::time::Duration;
fn scripted() -> ScriptedRunner {
ScriptedRunner::new()
.on(["tool", "--version"], Reply::ok("tool 1.2.3\n"))
.on(["tool", "fail"], Reply::fail(7, "boom"))
}
fn temp_cassette() -> (tempfile::TempDir, PathBuf) {
let dir = tempfile::tempdir().expect("create temp dir");
let path = dir.path().join("cassette.json");
(dir, path)
}
#[cfg(unix)]
#[test]
fn write_cassette_refuses_to_follow_a_symlink() {
let dir = tempfile::tempdir().expect("temp dir");
let target = dir.path().join("victim.txt");
std::fs::write(&target, "original").expect("seed victim");
let link = dir.path().join("cassette.json");
std::os::unix::fs::symlink(&target, &link).expect("create symlink");
let err = write_cassette(&link, "{\"secret\":true}")
.expect_err("writing through a symlink must fail (O_NOFOLLOW)");
assert_eq!(
err.raw_os_error(),
Some(libc::ELOOP),
"O_NOFOLLOW on a symlink yields ELOOP, got {err:?}"
);
assert_eq!(
std::fs::read_to_string(&target).expect("read victim"),
"original",
"the victim file must be untouched"
);
}
#[tokio::test]
async fn round_trip_is_identical() {
let (_dir, path) = temp_cassette();
let recorder = RecordReplayRunner::record(&path, scripted());
let ok = recorder
.output_string(&Command::new("tool").arg("--version"))
.await
.expect("record ok run");
let fail = recorder
.output_string(&Command::new("tool").arg("fail"))
.await
.expect("record failing run (non-zero exit is a result, not Err)");
recorder.save().expect("save cassette");
let replayer = RecordReplayRunner::replay(&path).expect("load cassette");
let ok2 = replayer
.output_string(&Command::new("tool").arg("--version"))
.await
.expect("replay ok run");
let fail2 = replayer
.output_string(&Command::new("tool").arg("fail"))
.await
.expect("replay failing run");
assert_eq!(ok, ok2, "replay must be identical to the recording");
assert_eq!(fail, fail2);
assert_eq!(fail2.code(), Some(7));
assert_eq!(fail2.stderr(), "boom");
}
#[tokio::test]
async fn start_records_then_replays_a_streaming_run() {
let (_dir, path) = temp_cassette();
let inner = ScriptedRunner::new().on(
["server", "--watch"],
Reply::lines(["starting", "listening on :8080", "ready"]),
);
let recorder = RecordReplayRunner::record(&path, inner);
let mut run = recorder
.start(&Command::new("server").arg("--watch"))
.await
.expect("record start");
let line = run
.wait_for_line(|l| l.contains("listening"), Duration::from_secs(5))
.await
.expect("readiness line during record");
assert_eq!(line, "listening on :8080");
assert_eq!(run.wait().await.expect("record finish"), Outcome::Exited(0));
recorder.save().expect("save cassette");
let replayer = RecordReplayRunner::replay(&path).expect("load cassette");
let mut replayed = replayer
.start(&Command::new("server").arg("--watch"))
.await
.expect("replay start");
assert_eq!(replayed.pid(), None, "a replayed handle has no OS identity");
let line2 = replayed
.wait_for_line(|l| l.contains("listening"), Duration::from_secs(5))
.await
.expect("readiness line during replay");
assert_eq!(line2, "listening on :8080");
assert_eq!(
replayed.wait().await.expect("replay finish"),
Outcome::Exited(0),
"replay must reproduce the recorded outcome through the streaming path"
);
}
#[tokio::test]
async fn start_record_fires_line_side_effects_exactly_once() {
use std::sync::Arc;
use std::sync::atomic::{AtomicUsize, Ordering as AtomicOrdering};
struct SharedSink(Arc<std::sync::Mutex<Vec<u8>>>);
impl tokio::io::AsyncWrite for SharedSink {
fn poll_write(
self: std::pin::Pin<&mut Self>,
_cx: &mut std::task::Context<'_>,
buf: &[u8],
) -> std::task::Poll<std::io::Result<usize>> {
self.0.lock().expect("sink mutex").extend_from_slice(buf);
std::task::Poll::Ready(Ok(buf.len()))
}
fn poll_flush(
self: std::pin::Pin<&mut Self>,
_cx: &mut std::task::Context<'_>,
) -> std::task::Poll<std::io::Result<()>> {
std::task::Poll::Ready(Ok(()))
}
fn poll_shutdown(
self: std::pin::Pin<&mut Self>,
_cx: &mut std::task::Context<'_>,
) -> std::task::Poll<std::io::Result<()>> {
std::task::Poll::Ready(Ok(()))
}
}
let (_dir, path) = temp_cassette();
let inner = ScriptedRunner::new().on(["tool"], Reply::lines(["a", "b", "c"]));
let recorder = RecordReplayRunner::record(&path, inner);
let hits = Arc::new(AtomicUsize::new(0));
let counter = Arc::clone(&hits);
let teed = Arc::new(std::sync::Mutex::new(Vec::<u8>::new()));
let cmd = Command::new("tool")
.on_stdout_line(move |_line| {
counter.fetch_add(1, AtomicOrdering::SeqCst);
})
.stdout_tee(SharedSink(Arc::clone(&teed)));
let run = recorder.start(&cmd).await.expect("record start");
let _ = run
.output_string()
.await
.expect("consume the scripted handle");
assert_eq!(
hits.load(AtomicOrdering::SeqCst),
3,
"stdout handler must fire once per line, not twice (the capture pass must stay silent)"
);
assert_eq!(
teed.lock().expect("teed mutex").as_slice(),
b"a\nb\nc\n",
"stdout tee must receive each line once, not twice"
);
}
#[tokio::test]
async fn start_replay_reproduces_signal_and_timeout_outcomes() {
let (_dir, path) = temp_cassette();
let json = serde_json::json!({
"version": 1,
"entries": [
{ "program": "killed", "args": [], "stdout": "", "stderr": "", "signal": 9 },
{ "program": "crashed", "args": [], "stdout": "", "stderr": "" },
{ "program": "slow", "args": [], "stdout": "", "stderr": "", "timed_out": true }
]
});
std::fs::write(&path, serde_json::to_string_pretty(&json).unwrap()).unwrap();
let replayer = RecordReplayRunner::replay(&path).expect("load cassette");
let signalled = replayer
.start(&Command::new("killed"))
.await
.expect("replay a signalled run through start")
.wait()
.await
.expect("wait the signalled handle");
assert_eq!(signalled, Outcome::Signalled(Some(9)));
let signalled_unknown = replayer
.start(&Command::new("crashed"))
.await
.expect("replay a signal-unknown run through start")
.wait()
.await
.expect("wait the signal-unknown handle");
assert_eq!(signalled_unknown, Outcome::Signalled(None));
let timed_out = replayer
.start(&Command::new("slow"))
.await
.expect("replay a timed-out run through start")
.wait()
.await
.expect("wait the timed-out handle");
assert_eq!(timed_out, Outcome::TimedOut);
}
#[tokio::test]
async fn output_bytes_is_unsupported_in_both_modes() {
let (_dir, path) = temp_cassette();
let recorder = RecordReplayRunner::record(&path, scripted());
let rec_err = recorder
.output_bytes(&Command::new("tool").arg("--version"))
.await
.expect_err("output_bytes must be unsupported in record mode");
assert!(
matches!(rec_err, Error::Unsupported { .. }),
"got {rec_err:?}"
);
let _ = recorder
.output_string(&Command::new("tool").arg("--version"))
.await
.expect("record a real entry");
recorder.save().expect("save");
let replayer = RecordReplayRunner::replay(&path).expect("load cassette");
let rep_err = replayer
.output_bytes(&Command::new("tool").arg("--version"))
.await
.expect_err("output_bytes must be unsupported in replay mode");
assert!(
matches!(rep_err, Error::Unsupported { .. }),
"got {rep_err:?}"
);
}
#[tokio::test]
async fn duplicate_key_plays_in_order_then_repeats_last() {
let (_dir, path) = temp_cassette();
let json = serde_json::json!({
"version": 1,
"entries": [
{
"program": "git", "args": ["head"],
"stdout": "aaa", "stderr": "", "code": 0
},
{
"program": "git", "args": ["head"],
"stdout": "bbb", "stderr": "", "code": 0
}
]
});
std::fs::write(&path, serde_json::to_string_pretty(&json).unwrap()).unwrap();
let cmd = Command::new("git").arg("head");
let replayer = RecordReplayRunner::replay(&path).expect("load cassette");
let first = replayer.run(&cmd).await.expect("first replay");
let second = replayer.run(&cmd).await.expect("second replay");
let third = replayer.run(&cmd).await.expect("third replay repeats last");
assert_eq!(first, "aaa");
assert_eq!(second, "bbb");
assert_eq!(third, "bbb", "exhausted key must repeat the last entry");
}
#[tokio::test]
async fn replay_rejects_an_entry_with_contradictory_outcome() {
let (_dir, path) = temp_cassette();
let json = serde_json::json!({
"version": 1,
"entries": [
{ "program": "x", "args": [], "stdout": "", "stderr": "", "code": 0, "signal": 9 }
]
});
std::fs::write(&path, serde_json::to_string_pretty(&json).unwrap()).unwrap();
let err = RecordReplayRunner::replay(&path)
.expect_err("a contradictory outcome must be rejected");
assert!(
matches!(&err, Error::Io(e) if e.kind() == std::io::ErrorKind::InvalidData),
"got {err:?}"
);
}
#[tokio::test]
async fn replay_miss_is_a_distinct_cassette_miss_error() {
let (_dir, path) = temp_cassette();
let recorder = RecordReplayRunner::record(&path, scripted());
let _ = recorder
.output_string(&Command::new("tool").arg("--version"))
.await
.expect("record");
recorder.save().expect("save");
let replayer = RecordReplayRunner::replay(&path).expect("load");
let err = replayer
.output_string(&Command::new("tool").arg("--other"))
.await
.expect_err("an unrecorded invocation must not be served");
match &err {
Error::CassetteMiss { program } => assert_eq!(program, "tool"),
other => panic!("expected Error::CassetteMiss, got {other:?}"),
}
assert!(
!err.is_not_found(),
"a cassette miss must not read as not-found: {err:?}"
);
}
#[tokio::test]
async fn replay_invokes_line_handlers() {
let (_dir, path) = temp_cassette();
let recorder = RecordReplayRunner::record(&path, scripted());
let _ = recorder
.output_string(&Command::new("tool").arg("--version"))
.await
.expect("record");
recorder.save().expect("save");
let seen = std::sync::Arc::new(Mutex::new(Vec::new()));
let replayer = RecordReplayRunner::replay(&path).expect("load");
let cmd = Command::new("tool").arg("--version").on_stdout_line({
let seen = seen.clone();
move |l| seen.lock().unwrap().push(l.to_owned())
});
let _ = replayer.output_string(&cmd).await.expect("replay");
assert_eq!(
*seen.lock().unwrap(),
["tool 1.2.3"],
"replay must invoke the command's line handler"
);
}
#[tokio::test]
async fn stdin_content_is_part_of_the_match_key() {
let (_dir, path) = temp_cassette();
let inner = ScriptedRunner::new()
.on_sequence(["tool"], [Reply::ok("out-A\n"), Reply::ok("out-B\n")]);
let recorder = RecordReplayRunner::record(&path, inner);
let _ = recorder
.output_string(&Command::new("tool").stdin(crate::Stdin::from_string("A")))
.await
.expect("record A");
let _ = recorder
.output_string(&Command::new("tool").stdin(crate::Stdin::from_string("B")))
.await
.expect("record B");
recorder.save().expect("save");
let replayer = RecordReplayRunner::replay(&path).expect("load");
let b = replayer
.output_string(&Command::new("tool").stdin(crate::Stdin::from_string("B")))
.await
.expect("replay B");
assert_eq!(
b.stdout(),
"out-B",
"stdin B must replay its own recording"
);
let a = replayer
.output_string(&Command::new("tool").stdin(crate::Stdin::from_string("A")))
.await
.expect("replay A");
assert_eq!(a.stdout(), "out-A", "stdin A must replay its own recording");
}
#[tokio::test]
async fn one_shot_streaming_stdin_is_rejected_in_both_modes() {
let (_dir, path) = temp_cassette();
let inner = ScriptedRunner::new().fallback(Reply::ok("out\n"));
let recorder = RecordReplayRunner::record(&path, inner);
let err = recorder
.output_string(&Command::new("tool").stdin(crate::Stdin::from_reader(&b"payload"[..])))
.await
.expect_err("record must reject a one-shot streaming stdin");
assert!(matches!(err, Error::Unsupported { .. }), "got {err:?}");
let _ = recorder
.output_string(&Command::new("tool"))
.await
.expect("record a replayable entry");
recorder.save().expect("save");
let replayer = RecordReplayRunner::replay(&path).expect("load");
let err = replayer
.output_string(&Command::new("tool").stdin(crate::Stdin::from_reader(&b"payload"[..])))
.await
.expect_err("replay must reject a one-shot streaming stdin");
assert!(matches!(err, Error::Unsupported { .. }), "got {err:?}");
}
#[tokio::test]
async fn no_stdin_replay_does_not_match_a_stdin_recorded_entry() {
let (_dir, path) = temp_cassette();
let recorder =
RecordReplayRunner::record(&path, ScriptedRunner::new().fallback(Reply::ok("out\n")));
let _ = recorder
.output_string(&Command::new("tool").stdin(crate::Stdin::from_string("input")))
.await
.expect("record with stdin");
recorder.save().expect("save");
let replayer = RecordReplayRunner::replay(&path).expect("load");
let err = replayer
.output_string(&Command::new("tool"))
.await
.expect_err("a no-stdin call must not match a stdin-recorded entry");
assert!(matches!(err, Error::CassetteMiss { .. }), "got {err:?}");
}
#[tokio::test]
async fn replayed_timeout_carries_the_commands_deadline() {
let (_dir, path) = temp_cassette();
let recorder = RecordReplayRunner::record(
&path,
ScriptedRunner::new().on(["tool", "slow"], Reply::timeout()),
);
let _ = recorder
.output_string(&Command::new("tool").arg("slow"))
.await
.expect("a captured timeout is a result, not an Err");
recorder.save().expect("save");
let replayer = RecordReplayRunner::replay(&path).expect("load");
let err = replayer
.run(
&Command::new("tool")
.arg("slow")
.timeout(Duration::from_secs(7)),
)
.await
.expect_err("run() raises the captured timeout");
match err {
Error::Timeout { timeout, .. } => assert_eq!(timeout, Duration::from_secs(7)),
other => panic!("expected Error::Timeout, got {other:?}"),
}
}
#[tokio::test]
async fn env_values_never_reach_the_file() {
let (_dir, path) = temp_cassette();
let recorder =
RecordReplayRunner::record(&path, ScriptedRunner::new().fallback(Reply::ok("done")));
let _ = recorder
.output_string(
&Command::new("tool")
.env("API_TOKEN", "hunter2-very-secret")
.env("MODE", "fast"),
)
.await
.expect("record");
recorder.save().expect("save");
let json = std::fs::read_to_string(&path).expect("read cassette");
assert!(json.contains("API_TOKEN"), "names are stored: {json}");
assert!(json.contains("MODE"));
assert!(
!json.contains("hunter2-very-secret") && !json.contains("fast"),
"values must never be written: {json}"
);
let replayer = RecordReplayRunner::replay(&path).expect("load");
let out = replayer
.run(&Command::new("tool"))
.await
.expect("env is not part of the match key");
assert_eq!(out, "done");
}
#[tokio::test]
async fn signal_number_survives_round_trip() {
let (_dir, path) = temp_cassette();
let json = r#"{"version":1,"entries":[{"program":"tool","args":[],"stdout":"","stderr":"","code":null,"signal":9}]}"#;
std::fs::write(&path, json).expect("write cassette");
let replayer = RecordReplayRunner::replay(&path).expect("load cassette");
let result = replayer
.output_string(&Command::new("tool"))
.await
.expect("replay");
assert_eq!(result.outcome(), Outcome::Signalled(Some(9)));
}
#[tokio::test]
async fn cassette_without_signal_field_loads_as_signalled_none() {
let (_dir, path) = temp_cassette();
let json = r#"{"version":1,"entries":[{"program":"tool","args":[],"stdout":"","stderr":"","code":null}]}"#;
std::fs::write(&path, json).expect("write cassette");
let replayer = RecordReplayRunner::replay(&path).expect("load cassette");
let result = replayer
.output_string(&Command::new("tool"))
.await
.expect("replay");
assert_eq!(result.outcome(), Outcome::Signalled(None));
}
#[tokio::test]
async fn load_errors_are_typed_io() {
let (_dir, path) = temp_cassette();
match RecordReplayRunner::replay(&path) {
Err(Error::Io(e)) => assert_eq!(e.kind(), std::io::ErrorKind::NotFound),
other => panic!("expected Io(NotFound), got {other:?}"),
}
std::fs::write(&path, "{ not json").unwrap();
match RecordReplayRunner::replay(&path) {
Err(Error::Io(e)) => assert_eq!(e.kind(), std::io::ErrorKind::InvalidData),
other => panic!("expected Io(InvalidData), got {other:?}"),
}
std::fs::write(&path, r#"{ "version": 99, "entries": [] }"#).unwrap();
match RecordReplayRunner::replay(&path) {
Err(Error::Io(e)) => {
assert_eq!(e.kind(), std::io::ErrorKind::InvalidData);
assert!(e.to_string().contains("version 99"), "got: {e}");
}
other => panic!("expected Io(InvalidData), got {other:?}"),
}
}
#[tokio::test]
async fn version_gate_fires_before_the_full_entries_decode() {
let (_dir, path) = temp_cassette();
std::fs::write(
&path,
r#"{ "version": 99, "entries": [ { "program": "x", "args": [], "code": "not-a-number" } ] }"#,
)
.unwrap();
match RecordReplayRunner::replay(&path) {
Err(Error::Io(e)) => {
assert_eq!(e.kind(), std::io::ErrorKind::InvalidData);
assert!(
e.to_string().contains("version 99"),
"expected the clear version-gate message, got: {e}"
);
}
other => panic!("expected the version-gate error, got {other:?}"),
}
}
#[tokio::test]
async fn replay_output_string_rejects_non_piped_stdout_even_on_a_match() {
let (_dir, path) = temp_cassette();
let recorder = RecordReplayRunner::record(&path, scripted());
let _ = recorder
.output_string(&Command::new("tool").arg("--version"))
.await
.expect("record ok run");
recorder.save().expect("save cassette");
let replayer = RecordReplayRunner::replay(&path).expect("load cassette");
let cmd = Command::new("tool")
.arg("--version")
.stdout(crate::StdioMode::Null);
let err = replayer
.output_string(&cmd)
.await
.expect_err("a non-piped stdout must error, even against a matching entry");
match err {
Error::Io(e) => assert_eq!(e.kind(), std::io::ErrorKind::InvalidInput),
other => panic!("expected Io(InvalidInput), got {other:?}"),
}
}
#[tokio::test]
async fn drop_without_save_flushes_best_effort() {
let (_dir, path) = temp_cassette();
{
let recorder = RecordReplayRunner::record(&path, scripted());
let _ = recorder
.output_string(&Command::new("tool").arg("--version"))
.await
.expect("record");
}
let replayer = RecordReplayRunner::replay(&path).expect("dropped recorder left a cassette");
let out = replayer
.run(&Command::new("tool").arg("--version"))
.await
.expect("replay after drop-flush");
assert_eq!(out, "tool 1.2.3");
}
#[tokio::test]
async fn cwd_is_not_part_of_the_match_key() {
let (_dir, path) = temp_cassette();
let recorder =
RecordReplayRunner::record(&path, ScriptedRunner::new().fallback(Reply::ok("from-a")));
let _ = recorder
.output_string(&Command::new("tool").current_dir("/home/dev/checkout"))
.await
.expect("record in one absolute cwd");
recorder.save().expect("save");
let replayer = RecordReplayRunner::replay(&path).expect("load");
let cross_dir = replayer
.run(&Command::new("tool").current_dir(r"C:\actions\work\checkout"))
.await
.expect(
"a differing absolute cwd (dev box -> CI workspace) must still replay: \
cwd is not part of the match key",
);
assert_eq!(cross_dir, "from-a");
let no_cwd = replayer
.run(&Command::new("tool"))
.await
.expect("no cwd at all must replay the same recorded entry too");
assert_eq!(no_cwd, "from-a");
let json = std::fs::read_to_string(&path).expect("read cassette");
assert!(
json.contains("/home/dev/checkout"),
"cwd must still be stored for visibility: {json}"
);
}
#[tokio::test]
async fn differing_program_or_args_still_miss_with_cwd_excluded() {
let (_dir, path) = temp_cassette();
let recorder =
RecordReplayRunner::record(&path, ScriptedRunner::new().fallback(Reply::ok("from-a")));
let _ = recorder
.output_string(&Command::new("tool").arg("build").current_dir("dir-a"))
.await
.expect("record");
recorder.save().expect("save");
let replayer = RecordReplayRunner::replay(&path).expect("load");
let err = replayer
.output_string(&Command::new("other").arg("build").current_dir("dir-b"))
.await
.expect_err("a different program is still a miss");
assert!(matches!(err, Error::CassetteMiss { .. }), "got {err:?}");
let err = replayer
.output_string(&Command::new("tool").arg("test").current_dir("dir-b"))
.await
.expect_err("different args are still a miss");
assert!(matches!(err, Error::CassetteMiss { .. }), "got {err:?}");
}
#[cfg(unix)]
#[tokio::test]
async fn cassette_file_is_written_owner_only() {
use std::os::unix::fs::PermissionsExt;
let (_dir, path) = temp_cassette();
let recorder = RecordReplayRunner::record(&path, scripted());
let _ = recorder
.output_string(&Command::new("tool").arg("--version"))
.await
.expect("record");
recorder.save().expect("save");
let mode = std::fs::metadata(&path)
.expect("stat cassette")
.permissions()
.mode();
assert_eq!(
mode & 0o777,
0o600,
"cassette must be owner-only, got {:o}",
mode & 0o777
);
}
#[tokio::test]
async fn drop_while_unwinding_does_not_persist_a_surprise_cassette() {
let (_dir, path) = temp_cassette();
let recorder = RecordReplayRunner::record(&path, scripted());
let _ = recorder
.output_string(&Command::new("tool").arg("--version"))
.await
.expect("record (now dirty, unsaved)");
let outcome = std::panic::catch_unwind(std::panic::AssertUnwindSafe(move || {
let _hold = recorder;
panic!("boom mid-recording");
}));
assert!(outcome.is_err(), "the scope must have panicked");
assert!(
!path.exists(),
"a recorder dropped during unwind must not persist a cassette: {path:?}"
);
}
#[tokio::test]
async fn save_then_record_more_then_drop_flushes_the_late_runs() {
let (_dir, path) = temp_cassette();
{
let recorder = RecordReplayRunner::record(&path, scripted());
let _ = recorder
.output_string(&Command::new("tool").arg("--version"))
.await
.expect("record first");
recorder.save().expect("first save");
let _ = recorder
.output_string(&Command::new("tool").arg("fail"))
.await
.expect("record second");
}
let replayer = RecordReplayRunner::replay(&path).expect("load");
let result = replayer
.output_string(&Command::new("tool").arg("fail"))
.await
.expect("the post-save run was flushed by drop");
assert_eq!(result.code(), Some(7));
}
#[tokio::test]
async fn concurrent_saves_to_one_path_never_corrupt_the_cassette() {
let (_dir, path) = temp_cassette();
let mut recorders = Vec::new();
for _ in 0..8 {
let r = RecordReplayRunner::record(
&path,
ScriptedRunner::new().on(["tool", "ping"], Reply::ok("pong")),
);
let _ = r
.output_string(&Command::new("tool").arg("ping"))
.await
.expect("record ping");
recorders.push(r);
}
let results: Vec<Result<()>> = std::thread::scope(|s| {
let handles: Vec<_> = recorders.iter().map(|r| s.spawn(|| r.save())).collect();
handles
.into_iter()
.map(|h| h.join().expect("a save thread must not panic"))
.collect()
});
for res in &results {
if let Err(err) = res {
assert!(
err.is_transient(),
"a losing concurrent save must be the transient WouldBlock \
conflict, got {err:?}"
);
}
}
assert!(
results.iter().any(|r| r.is_ok()),
"at least one concurrent save must win"
);
let replayer = RecordReplayRunner::replay(&path).expect("reopen after concurrent writes");
let out = replayer
.output_string(&Command::new("tool").arg("ping"))
.await
.expect("replay the surviving entry");
assert_eq!(out.stdout(), "pong");
}
#[tokio::test]
async fn a_save_racing_a_held_lock_is_refused_not_a_silent_clobber() {
let (_dir, path) = temp_cassette();
let winner = RecordReplayRunner::record(
&path,
ScriptedRunner::new().on(["tool", "keep"], Reply::ok("kept")),
);
let _ = winner
.output_string(&Command::new("tool").arg("keep"))
.await
.expect("record the good run");
winner.save().expect("the first save wins the free lock");
let held = acquire_save_lock(&path).expect("model a competing writer holding the lock");
let loser = RecordReplayRunner::record(
&path,
ScriptedRunner::new().on(["tool", "clobber"], Reply::ok("clobbered\n")),
);
let _ = loser
.output_string(&Command::new("tool").arg("clobber"))
.await
.expect("record the clobbering run");
let err = loser
.save()
.expect_err("a save racing a held lock must be refused, not silently applied");
assert!(
err.is_transient(),
"the conflict must be the transient WouldBlock error, got {err:?}"
);
drop(held);
let replayer = RecordReplayRunner::replay(&path).expect("reopen the preserved cassette");
assert_eq!(
replayer
.output_string(&Command::new("tool").arg("keep"))
.await
.expect("the good run is still there")
.stdout(),
"kept"
);
let miss = replayer
.output_string(&Command::new("tool").arg("clobber"))
.await
.expect_err("the refused save's run must be absent");
assert!(
matches!(miss, Error::CassetteMiss { .. }),
"the clobbering entry must be absent (a miss), got {miss:?}"
);
}
#[tokio::test]
async fn a_stale_temp_from_a_crashed_writer_is_left_untouched_and_does_not_block() {
let (_dir, path) = temp_cassette();
let stale = {
let mut name = path.as_os_str().to_owned();
name.push(".4294967295.0.0.tmp");
PathBuf::from(name)
};
std::fs::write(&stale, "garbage from a crashed writer").expect("seed stale temp");
let recorder = RecordReplayRunner::record(&path, scripted());
let _ = recorder
.output_string(&Command::new("tool").arg("--version"))
.await
.expect("record");
recorder
.save()
.expect("save must succeed despite the stale temp");
assert!(
stale.exists(),
"a stale temp (possibly another writer's live temp) must be left untouched"
);
assert_eq!(
std::fs::read_to_string(&stale).expect("read stale temp"),
"garbage from a crashed writer",
"the stale temp's contents must not be disturbed"
);
RecordReplayRunner::replay(&path).expect("the cassette is valid and reopens");
}
#[tokio::test]
async fn a_write_failure_surfaces_as_err_and_drop_stays_non_panic() {
let dir = tempfile::tempdir().expect("temp dir");
let path = dir.path().join("cassette.json");
std::fs::create_dir(&path).expect("occupy the target path with a directory");
let recorder = RecordReplayRunner::record(&path, scripted());
let _ = recorder
.output_string(&Command::new("tool").arg("--version"))
.await
.expect("record (now dirty)");
let err = recorder
.save()
.expect_err("renaming the temp over a directory must fail, not silently succeed");
assert!(
matches!(err, Error::Io(_)),
"a write/rename failure is an Io error, got {err:?}"
);
drop(recorder);
}
#[tokio::test]
async fn non_utf8_args_are_recorded_lossily_not_fatally() {
#[cfg(unix)]
let bad = {
use std::os::unix::ffi::OsStringExt;
std::ffi::OsString::from_vec(vec![b'a', 0xFF, b'b'])
};
#[cfg(windows)]
let bad = {
use std::os::windows::ffi::OsStringExt;
std::ffi::OsString::from_wide(&[0x61, 0xD800, 0x62])
};
let (_dir, path) = temp_cassette();
let recorder =
RecordReplayRunner::record(&path, ScriptedRunner::new().fallback(Reply::ok("ok")));
let cmd = Command::new("tool").arg(&bad);
let _ = recorder.output_string(&cmd).await.expect("record lossily");
recorder.save().expect("save");
let replayer = RecordReplayRunner::replay(&path).expect("load");
let out = replayer.run(&cmd).await.expect("replay matches lossily");
assert_eq!(out, "ok");
}
struct TruncatedInner;
#[async_trait::async_trait]
impl ProcessRunner for TruncatedInner {
async fn output_string(&self, command: &Command) -> Result<ProcessResult<String>> {
Ok(ProcessResult::new(
command.program_name(),
"clipped".to_owned(),
String::new(),
Outcome::Exited(0),
None,
)
.with_truncated(true)
.with_overflow_totals(100, 9999)
.with_duration(Duration::from_millis(1234)))
}
}
#[tokio::test]
async fn truncation_and_duration_survive_replay() {
let (_dir, path) = temp_cassette();
let recorder = RecordReplayRunner::record(&path, TruncatedInner);
let recorded = recorder
.output_string(&Command::new("tool"))
.await
.expect("record");
assert!(recorded.truncated());
recorder.save().expect("save");
let replayer = RecordReplayRunner::replay(&path).expect("load");
let replayed = replayer
.output_string(&Command::new("tool"))
.await
.expect("replay");
assert!(replayed.truncated(), "truncation must survive replay (D1)");
assert_eq!(
replayed.duration(),
Duration::from_millis(1234),
"the recorded duration must survive replay (D12)"
);
let err = replayer
.run(&Command::new("tool"))
.await
.expect_err("run must reject a truncated replay");
assert!(matches!(err, Error::OutputTooLarge { .. }), "got {err:?}");
}
#[tokio::test]
async fn truncation_and_duration_survive_start_replay() {
let (_dir, path) = temp_cassette();
let recorder = RecordReplayRunner::record(&path, TruncatedInner);
let _ = recorder
.start(&Command::new("tool"))
.await
.expect("record start");
recorder.save().expect("save");
let replayer = RecordReplayRunner::replay(&path).expect("load");
let replayed = replayer
.start(&Command::new("tool"))
.await
.expect("replay start")
.output_string()
.await
.expect("consume the replayed handle");
assert!(
replayed.truncated(),
"recorded truncation must survive a start replay"
);
assert_eq!(
replayed.duration(),
Duration::from_millis(1234),
"recorded duration must survive a start replay"
);
assert_eq!(replayed.total_lines(), 100);
assert_eq!(replayed.total_bytes(), 9999);
}
#[tokio::test]
async fn replay_short_circuits_a_cancelled_token() {
let (_dir, path) = temp_cassette();
let recorder = RecordReplayRunner::record(&path, scripted());
let _ = recorder
.output_string(&Command::new("tool").arg("--version"))
.await
.expect("record");
recorder.save().expect("save");
let replayer = RecordReplayRunner::replay(&path).expect("load");
let token = crate::CancellationToken::new();
token.cancel();
let err = replayer
.output_string(&Command::new("tool").arg("--version").cancel_on(token))
.await
.expect_err("a pre-cancelled token must short-circuit replay");
assert!(matches!(err, Error::Cancelled { .. }), "got {err:?}");
}
#[tokio::test]
async fn start_replay_short_circuits_a_cancelled_token() {
let (_dir, path) = temp_cassette();
let recorder = RecordReplayRunner::record(&path, scripted());
let _ = recorder
.output_string(&Command::new("tool").arg("--version"))
.await
.expect("record");
recorder.save().expect("save");
let replayer = RecordReplayRunner::replay(&path).expect("load");
let token = crate::CancellationToken::new();
token.cancel();
let err = replayer
.start(&Command::new("tool").arg("--version").cancel_on(token))
.await
.expect_err("a pre-cancelled token must short-circuit start replay");
assert!(matches!(err, Error::Cancelled { .. }), "got {err:?}");
}
struct FailingInner(fn(&str) -> Error);
#[async_trait::async_trait]
impl ProcessRunner for FailingInner {
async fn output_string(&self, command: &Command) -> Result<ProcessResult<String>> {
Err((self.0)(&command.program_name()))
}
}
#[tokio::test]
async fn record_of_a_not_found_err_replays_the_same_not_found_error() {
let (_dir, path) = temp_cassette();
let inner = FailingInner(|program| Error::NotFound {
program: program.to_owned(),
searched: Some("/usr/bin:/bin".to_owned()),
});
let recorder = RecordReplayRunner::record(&path, inner);
let record_err = recorder
.output_string(&Command::new("ghost"))
.await
.expect_err("the inner runner's Err must still reach the record-mode caller");
assert!(
matches!(&record_err, Error::NotFound { .. }),
"got {record_err:?}"
);
recorder.save().expect("save");
let replayer = RecordReplayRunner::replay(&path).expect("load cassette");
let replay_err = replayer
.output_string(&Command::new("ghost"))
.await
.expect_err("replay must reproduce the recorded NotFound, not CassetteMiss");
match &replay_err {
Error::NotFound { program, searched } => {
assert_eq!(program, "ghost");
assert_eq!(searched.as_deref(), Some("/usr/bin:/bin"));
}
other => panic!("expected Error::NotFound, got {other:?}"),
}
assert!(
replay_err.is_not_found(),
"is_not_found() must still classify the replayed error"
);
let (_dir2, path2) = temp_cassette();
let inner2 = FailingInner(|program| Error::NotFound {
program: program.to_owned(),
searched: None,
});
let recorder2 = RecordReplayRunner::record(&path2, inner2);
let _ = recorder2
.start(&Command::new("ghost"))
.await
.expect_err("start must also surface the inner runner's Err in record mode");
recorder2.save().expect("save");
let replayer2 = RecordReplayRunner::replay(&path2).expect("load cassette");
let err2 = replayer2
.start(&Command::new("ghost"))
.await
.expect_err("start replay must reproduce the recorded NotFound");
assert!(matches!(err2, Error::NotFound { .. }), "got {err2:?}");
}
#[tokio::test]
async fn record_of_a_spawn_err_preserves_the_permission_denied_classification() {
let (_dir, path) = temp_cassette();
let inner = FailingInner(|program| Error::Spawn {
program: program.to_owned(),
source: std::io::Error::new(std::io::ErrorKind::PermissionDenied, "denied"),
});
let recorder = RecordReplayRunner::record(&path, inner);
let _ = recorder
.output_string(&Command::new("tool"))
.await
.expect_err("record");
recorder.save().expect("save");
let replayer = RecordReplayRunner::replay(&path).expect("load");
let err = replayer
.output_string(&Command::new("tool"))
.await
.expect_err("replay reproduces the recorded Spawn error");
assert!(matches!(err, Error::Spawn { .. }), "got {err:?}");
assert!(
err.is_permission_denied(),
"the recorded os error kind must survive replay: {err:?}"
);
}
#[tokio::test]
async fn a_cancelled_record_mode_call_is_never_persisted_to_the_cassette() {
let (_dir, path) = temp_cassette();
let inner = FailingInner(|program| Error::Cancelled {
program: program.to_owned(),
});
let recorder = RecordReplayRunner::record(&path, inner);
let err = recorder
.output_string(&Command::new("tool"))
.await
.expect_err("the cancelled call still surfaces its real error to the caller");
assert!(matches!(err, Error::Cancelled { .. }));
recorder.save().expect("save");
let replayer = RecordReplayRunner::replay(&path).expect("load");
let err = replayer
.output_string(&Command::new("tool"))
.await
.expect_err("a cassette with no recorded entry for this invocation must miss");
assert!(
matches!(err, Error::CassetteMiss { .. }),
"Cancelled must never be persisted, got {err:?}"
);
}
#[tokio::test]
async fn a_version_1_cassette_with_no_error_field_still_loads_and_replays() {
let (_dir, path) = temp_cassette();
let json = serde_json::json!({
"version": 1,
"entries": [
{ "program": "tool", "args": ["--version"], "stdout": "tool 1.2.3\n", "stderr": "", "code": 0 }
]
});
std::fs::write(&path, serde_json::to_string_pretty(&json).unwrap()).unwrap();
let replayer = RecordReplayRunner::replay(&path).expect("a version-1 cassette must load");
let out = replayer
.run(&Command::new("tool").arg("--version"))
.await
.expect("replay a version-1 entry with no error field");
assert_eq!(out, "tool 1.2.3");
}
#[tokio::test]
async fn unmodeled_error_variant_falls_back_to_other_rather_than_dropping_silently() {
let (_dir, path) = temp_cassette();
let inner = FailingInner(|program| Error::Parse {
program: program.to_owned(),
message: "unexpected token at line 3".to_owned(),
});
let recorder = RecordReplayRunner::record(&path, inner);
let record_err = recorder
.output_string(&Command::new("tool"))
.await
.expect_err("record");
let expected_message = record_err.to_string();
recorder.save().expect("save");
let replayer = RecordReplayRunner::replay(&path).expect("load");
let err = replayer
.output_string(&Command::new("tool"))
.await
.expect_err("replay must reproduce the Other fallback, not miss the cassette");
match err {
Error::Io(source) => {
assert_eq!(source.kind(), std::io::ErrorKind::Other, "got {source:?}");
assert_eq!(source.to_string(), expected_message);
}
other => panic!("expected Error::Io(ErrorKind::Other), got {other:?}"),
}
}
#[tokio::test]
async fn modeled_unsupported_error_round_trips_exactly_through_replay() {
let (_dir, path) = temp_cassette();
let inner = FailingInner(|_program| Error::Unsupported {
operation: "signal(Hup)".to_owned(),
});
let recorder = RecordReplayRunner::record(&path, inner);
let _ = recorder
.output_string(&Command::new("tool"))
.await
.expect_err("record");
recorder.save().expect("save");
let replayer = RecordReplayRunner::replay(&path).expect("load");
let err = replayer
.output_string(&Command::new("tool"))
.await
.expect_err("replay must reproduce Unsupported, not miss the cassette");
match err {
Error::Unsupported { operation } => assert_eq!(operation, "signal(Hup)"),
other => panic!("expected Error::Unsupported, got {other:?}"),
}
}
#[tokio::test]
async fn default_ignores_differing_env_values_without_a_policy() {
let (_dir, path) = temp_cassette();
let recorder =
RecordReplayRunner::record(&path, ScriptedRunner::new().fallback(Reply::ok("from-a")));
let _ = recorder
.output_string(&Command::new("tool").env("MODE", "a"))
.await
.expect("record with MODE=a");
recorder.save().expect("save");
let replayer = RecordReplayRunner::replay(&path).expect("load");
let out = replayer
.run(&Command::new("tool").env("MODE", "b"))
.await
.expect("default: env is not keyed, so a differing value still hits");
assert_eq!(out, "from-a");
}
#[tokio::test]
async fn match_on_env_makes_a_differing_value_miss_and_ignores_unnamed_vars() {
let (_dir, path) = temp_cassette();
let recorder = RecordReplayRunner::record(
&path,
ScriptedRunner::new().fallback(Reply::ok("recorded")),
)
.match_on_env(["MODE"]);
let _ = recorder
.output_string(&Command::new("tool").env("MODE", "fast"))
.await
.expect("record MODE=fast");
recorder.save().expect("save");
let replayer = RecordReplayRunner::replay(&path)
.expect("load")
.match_on_env(["MODE"]);
let hit = replayer
.run(&Command::new("tool").env("MODE", "fast"))
.await
.expect("the same selected env value must replay its recording");
assert_eq!(hit, "recorded");
let err = replayer
.output_string(&Command::new("tool").env("MODE", "slow"))
.await
.expect_err("a differing selected env value must miss under the policy");
assert!(matches!(err, Error::CassetteMiss { .. }), "got {err:?}");
let still_hit = replayer
.run(
&Command::new("tool")
.env("MODE", "fast")
.env("UNRELATED", "x"),
)
.await
.expect("an env var the policy does not name must not perturb the key");
assert_eq!(still_hit, "recorded");
}
#[tokio::test]
async fn match_on_env_records_distinct_entries_per_value() {
let (_dir, path) = temp_cassette();
let recorder = RecordReplayRunner::record(
&path,
ScriptedRunner::new()
.on_sequence(["tool"], [Reply::ok("out-fast\n"), Reply::ok("out-slow\n")]),
)
.match_on_env(["MODE"]);
let _ = recorder
.output_string(&Command::new("tool").env("MODE", "fast"))
.await
.expect("record fast");
let _ = recorder
.output_string(&Command::new("tool").env("MODE", "slow"))
.await
.expect("record slow");
recorder.save().expect("save");
let replayer = RecordReplayRunner::replay(&path)
.expect("load")
.match_on_env(["MODE"]);
let slow = replayer
.run(&Command::new("tool").env("MODE", "slow"))
.await
.expect("replay slow");
assert_eq!(slow, "out-slow");
let fast = replayer
.run(&Command::new("tool").env("MODE", "fast"))
.await
.expect("replay fast");
assert_eq!(fast, "out-fast");
}
#[tokio::test]
async fn match_on_env_never_writes_raw_values_only_a_digest() {
let (_dir, path) = temp_cassette();
let recorder =
RecordReplayRunner::record(&path, ScriptedRunner::new().fallback(Reply::ok("done")))
.match_on_env(["API_TOKEN"]);
let _ = recorder
.output_string(&Command::new("tool").env("API_TOKEN", "hunter2-very-secret"))
.await
.expect("record");
recorder.save().expect("save");
let json = std::fs::read_to_string(&path).expect("read cassette");
assert!(
json.contains("API_TOKEN"),
"the name is still stored: {json}"
);
assert!(
!json.contains("hunter2-very-secret"),
"the raw value must never be written, even under the env match policy: {json}"
);
assert!(
json.contains("match_digest"),
"the opaque policy digest is what keys the env value: {json}"
);
}
#[tokio::test]
async fn match_on_env_distinguishes_a_set_var_from_an_untouched_one() {
let (_dir, path) = temp_cassette();
let recorder = RecordReplayRunner::record(
&path,
ScriptedRunner::new().fallback(Reply::ok("with-flag")),
)
.match_on_env(["FLAG"]);
let _ = recorder
.output_string(&Command::new("tool").env("FLAG", "on"))
.await
.expect("record with FLAG set");
recorder.save().expect("save");
let replayer = RecordReplayRunner::replay(&path)
.expect("load")
.match_on_env(["FLAG"]);
let err = replayer
.output_string(&Command::new("tool")) .await
.expect_err("an untouched selected var must not match a set-var recording");
assert!(matches!(err, Error::CassetteMiss { .. }), "got {err:?}");
}
#[tokio::test]
async fn match_on_cwd_keys_on_working_directory() {
let (_dir, path) = temp_cassette();
let recorder = RecordReplayRunner::record(
&path,
ScriptedRunner::new().fallback(Reply::ok("from-dir-a")),
)
.match_on_cwd();
let _ = recorder
.output_string(&Command::new("tool").current_dir("/work/a"))
.await
.expect("record in /work/a");
recorder.save().expect("save");
let replayer = RecordReplayRunner::replay(&path)
.expect("load")
.match_on_cwd();
let hit = replayer
.run(&Command::new("tool").current_dir("/work/a"))
.await
.expect("the same cwd must replay under match_on_cwd");
assert_eq!(hit, "from-dir-a");
let err = replayer
.output_string(&Command::new("tool").current_dir("/work/b"))
.await
.expect_err("a differing cwd must miss under match_on_cwd");
assert!(matches!(err, Error::CassetteMiss { .. }), "got {err:?}");
}
#[tokio::test]
async fn a_policy_keyed_cassette_replayed_without_the_policy_misses() {
let (_dir, path) = temp_cassette();
let recorder =
RecordReplayRunner::record(&path, ScriptedRunner::new().fallback(Reply::ok("x")))
.match_on_env(["MODE"]);
let _ = recorder
.output_string(&Command::new("tool").env("MODE", "a"))
.await
.expect("record under a policy");
recorder.save().expect("save");
let replayer = RecordReplayRunner::replay(&path).expect("load"); let err = replayer
.output_string(&Command::new("tool").env("MODE", "a"))
.await
.expect_err("a policy-keyed entry must not match a no-policy replay");
assert!(matches!(err, Error::CassetteMiss { .. }), "got {err:?}");
}
#[tokio::test]
async fn match_on_env_names_are_order_independent() {
let (_dir, path) = temp_cassette();
let recorder =
RecordReplayRunner::record(&path, ScriptedRunner::new().fallback(Reply::ok("ok")))
.match_on_env(["ALPHA", "BETA"]);
let _ = recorder
.output_string(&Command::new("tool").env("ALPHA", "1").env("BETA", "2"))
.await
.expect("record");
recorder.save().expect("save");
let replayer = RecordReplayRunner::replay(&path)
.expect("load")
.match_on_env(["BETA"])
.match_on_env(["ALPHA"]);
let out = replayer
.run(&Command::new("tool").env("ALPHA", "1").env("BETA", "2"))
.await
.expect("policy name order must not affect the key");
assert_eq!(out, "ok");
}
}