1use std::{
17 cmp,
18 collections::BTreeMap,
19 ffi::OsString,
20 fmt::Display,
21 fs::{File, remove_file},
22 io::{BufReader, BufWriter, Read, Write},
23 path::{Path, PathBuf},
24 sync::OnceLock,
25 thread::sleep,
26 time::{Duration, Instant},
27};
28
29use anyhow::Context;
30use aws_lc_rs::digest;
31use nautilus_core::hex;
32use nautilus_network::{
33 http::{HttpClient, HttpResponseStream},
34 retry::RetryConfig,
35};
36use parking_lot::Mutex;
37use rand::{RngExt, rng};
38use serde_json::Value;
39
40static LARGE_CHECKSUMS_LOCK: OnceLock<Mutex<()>> = OnceLock::new();
41
42fn lock_large_checksums() -> parking_lot::MutexGuard<'static, ()> {
43 LARGE_CHECKSUMS_LOCK.get_or_init(|| Mutex::new(())).lock()
44}
45
46#[derive(Debug)]
47enum DownloadError {
48 Retryable(String),
49 NonRetryable(String),
50}
51
52impl Display for DownloadError {
53 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
54 match self {
55 Self::Retryable(msg) => write!(f, "Retryable error: {msg}"),
56 Self::NonRetryable(msg) => write!(f, "Non-retryable error: {msg}"),
57 }
58 }
59}
60
61impl std::error::Error for DownloadError {}
62
63fn execute_with_retry_blocking<T, E, F>(
64 config: &RetryConfig,
65 mut op: F,
66 should_retry: impl Fn(&E) -> bool,
67) -> Result<T, E>
68where
69 E: std::error::Error,
70 F: FnMut() -> Result<T, E>,
71{
72 let start = Instant::now();
73 let mut delay = Duration::from_millis(config.initial_delay_ms);
74
75 for attempt in 0..=config.max_retries {
76 if attempt > 0 && !config.immediate_first {
77 let jitter = rng().random_range(0..=config.jitter_ms);
78 let sleep_for = delay + Duration::from_millis(jitter);
79 sleep(sleep_for);
80 delay = cmp::min(
81 next_retry_delay(delay, config.backoff_factor),
82 Duration::from_millis(config.max_delay_ms),
83 );
84 }
85
86 if let Some(max_total) = config.max_elapsed_ms
87 && start.elapsed() >= Duration::from_millis(max_total)
88 {
89 break;
90 }
91
92 match op() {
93 Ok(v) => return Ok(v),
94 Err(e) if attempt < config.max_retries && should_retry(&e) => {}
95 Err(e) => return Err(e),
96 }
97 }
98
99 op()
100}
101
102fn next_retry_delay(delay: Duration, backoff_factor: f64) -> Duration {
103 let next = delay.as_secs_f64() * backoff_factor;
104 if next.is_nan() || next.is_sign_negative() {
105 return Duration::ZERO;
106 }
107
108 Duration::try_from_secs_f64(next).unwrap_or(Duration::MAX)
109}
110
111pub fn prepare_test_data() -> anyhow::Result<()> {
119 let checksums = crate::common::get_test_data_large_checksums_filepath();
120 let manifest: BTreeMap<String, String> =
121 serde_json::from_reader(BufReader::new(File::open(&checksums)?))?;
122 let directory = nautilus_core::paths::get_test_data_path().join("large");
123
124 for filename in manifest.keys() {
125 let filepath = directory.join(filename);
126 let url = format!("https://test-data.nautechsystems.io/large/{filename}");
127 prepare_test_data_file(&filepath, &url, &checksums)?;
128 }
129 Ok(())
130}
131
132fn prepare_test_data_file(filepath: &Path, url: &str, checksums: &Path) -> anyhow::Result<()> {
133 if filepath.exists() {
134 if verify_sha256_checksum(filepath, checksums)? {
135 return Ok(());
136 }
137 remove_file(filepath)?;
138 }
139
140 download_file(filepath, url, 30, None)?;
141 if !verify_sha256_checksum(filepath, checksums)? {
142 remove_file(filepath)?;
143 anyhow::bail!("Checksum mismatch for {}", filepath.display());
144 }
145 Ok(())
146}
147
148pub fn ensure_file_exists_or_download_http(
169 filepath: &Path,
170 url: &str,
171 checksums: Option<&Path>,
172 timeout_secs: Option<u64>,
173) -> anyhow::Result<()> {
174 ensure_file_exists_or_download_http_with_config(
175 filepath,
176 url,
177 checksums,
178 timeout_secs.unwrap_or(30),
179 None,
180 None,
181 )
182}
183
184pub fn ensure_file_exists_or_download_http_with_timeout(
193 filepath: &Path,
194 url: &str,
195 checksums: Option<&Path>,
196 timeout_secs: u64,
197) -> anyhow::Result<()> {
198 ensure_file_exists_or_download_http_with_config(
199 filepath,
200 url,
201 checksums,
202 timeout_secs,
203 None,
204 None,
205 )
206}
207
208pub fn ensure_file_exists_or_download_http_with_config(
227 filepath: &Path,
228 url: &str,
229 checksums: Option<&Path>,
230 timeout_secs: u64,
231 retry_config: Option<RetryConfig>,
232 initial_jitter_ms: Option<u64>,
233) -> anyhow::Result<()> {
234 if filepath.exists() {
238 println!("File already exists (local/cached): {}", filepath.display());
239
240 if let Some(checksums_file) = checksums {
241 let _guard = lock_large_checksums();
242
243 if verify_sha256_checksum(filepath, checksums_file)? {
244 println!("Checksum verified");
245 return Ok(());
246 }
247
248 let new_checksum = calculate_sha256(filepath)?;
249 println!("Updating checksum for local file: {new_checksum}");
250 update_sha256_checksums(filepath, checksums_file, &new_checksum)?;
251 return Ok(());
252 }
253 return Ok(());
254 }
255
256 if let Some(jitter_ms) = initial_jitter_ms {
259 if jitter_ms > 0 {
260 sleep(Duration::from_millis(jitter_ms));
261 }
262 } else {
263 let jitter_delay = {
264 let mut r = rng();
265 Duration::from_millis(r.random_range(100..=600))
266 };
267 sleep(jitter_delay);
268 }
269
270 download_file(filepath, url, timeout_secs, retry_config.clone())?;
271
272 if let Some(checksums_file) = checksums {
274 let guard = lock_large_checksums();
275
276 if !verify_sha256_checksum(filepath, checksums_file)? {
277 let actual = calculate_sha256(filepath)?;
278 println!("Checksum mismatch after download (calculated {actual}), retrying...");
279 remove_file(filepath)?;
280 drop(guard);
281
282 download_file(filepath, url, timeout_secs, retry_config)?;
283
284 let _guard = lock_large_checksums();
285
286 if !verify_sha256_checksum(filepath, checksums_file)? {
287 let actual = calculate_sha256(filepath)?;
288 remove_file(filepath)?;
289 anyhow::bail!(
290 "Checksum mismatch after retry for {} (calculated {actual})",
291 filepath.file_name().unwrap_or_default().display(),
292 );
293 }
294 }
295 }
296
297 Ok(())
298}
299
300fn download_file(
301 filepath: &Path,
302 url: &str,
303 timeout_secs: u64,
304 retry_config: Option<RetryConfig>,
305) -> anyhow::Result<()> {
306 #[cfg(not(test))]
310 if !url.starts_with("https://") {
311 anyhow::bail!("URL must use HTTPS protocol for security: {url}");
312 }
313
314 println!("Downloading file from {url} to {}", filepath.display());
315
316 if let Some(parent) = filepath.parent() {
317 std::fs::create_dir_all(parent)?;
318 }
319
320 let runtime = tokio::runtime::Builder::new_current_thread()
321 .enable_all()
322 .build()?;
323 let client = HttpClient::builder().build()?;
324 let timeout = Duration::from_secs(timeout_secs);
325
326 let cfg = if let Some(config) = retry_config {
327 config
328 } else {
329 let max_retries = 5u32;
331 let op_timeout_ms = timeout_secs.saturating_mul(1000);
332 let per_attempt_ms = std::cmp::max(1000u64, op_timeout_ms / (u64::from(max_retries) + 1));
335 RetryConfig {
336 max_retries,
337 initial_delay_ms: 1_000,
338 max_delay_ms: 10_000,
339 backoff_factor: 2.0,
340 jitter_ms: 1_000,
341 operation_timeout_ms: Some(per_attempt_ms),
342 immediate_first: false,
343 max_elapsed_ms: Some(op_timeout_ms),
344 }
345 };
346
347 remove_stale_partials(filepath);
348 let partial_path = partial_path_for(filepath);
349
350 let op = || -> Result<(), DownloadError> {
351 let _ = remove_file(&partial_path);
353
354 match runtime.block_on(async {
355 Ok::<_, anyhow::Error>(
356 tokio::time::timeout(timeout, client.get_stream(url.to_owned())).await??,
357 )
358 }) {
359 Ok(mut response) => {
360 let status = response.status();
361 if status.is_success() {
362 store_response_body(&runtime, &mut response, timeout, &partial_path, filepath)
363 } else if status.is_server_error()
364 || status.as_u16() == 429
365 || status.as_u16() == 408
366 {
367 println!("HTTP error {status}, retrying...");
368 Err(DownloadError::Retryable(format!("HTTP {status}")))
369 } else {
370 Err(DownloadError::NonRetryable(format!(
372 "Client error: HTTP {status}"
373 )))
374 }
375 }
376 Err(e) => {
377 println!("Request failed: {e}");
378 Err(DownloadError::Retryable(e.to_string()))
379 }
380 }
381 };
382
383 let should_retry = |e: &DownloadError| matches!(e, DownloadError::Retryable(_));
384
385 execute_with_retry_blocking(&cfg, op, should_retry).map_err(|e| anyhow::anyhow!(e.to_string()))
386}
387
388fn store_response_body(
390 runtime: &tokio::runtime::Runtime,
391 response: &mut HttpResponseStream,
392 timeout: Duration,
393 partial_path: &Path,
394 filepath: &Path,
395) -> Result<(), DownloadError> {
396 let mut out =
397 File::create(partial_path).map_err(|e| DownloadError::NonRetryable(e.to_string()))?;
398
399 let copied = runtime.block_on(async {
400 while let Some(chunk) = tokio::time::timeout(timeout, response.chunk())
401 .await
402 .map_err(|e| DownloadError::Retryable(format!("body stream error: {e}")))?
403 .map_err(|e| DownloadError::Retryable(format!("body stream error: {e}")))?
404 {
405 out.write_all(&chunk).map_err(|e| {
406 DownloadError::NonRetryable(format!("write {} failed: {e}", partial_path.display()))
407 })?;
408 }
409
410 out.sync_all().map_err(|e| {
412 DownloadError::NonRetryable(format!("sync {} failed: {e}", partial_path.display()))
413 })
414 });
415
416 drop(out);
417
418 if let Err(e) = copied {
419 let _ = remove_file(partial_path);
420 return Err(e);
421 }
422
423 if let Err(e) = std::fs::rename(partial_path, filepath) {
424 let _ = remove_file(partial_path);
425 return Err(DownloadError::NonRetryable(format!(
426 "rename {} -> {} failed: {e}",
427 partial_path.display(),
428 filepath.display(),
429 )));
430 }
431
432 println!("File downloaded to {}", filepath.display());
433 Ok(())
434}
435
436fn remove_stale_partials(filepath: &Path) {
438 const STALE_PARTIAL_AGE: Duration = Duration::from_secs(3600);
439
440 let (Some(parent), Some(file_name)) = (filepath.parent(), filepath.file_name()) else {
441 return;
442 };
443
444 let parent = if parent.as_os_str().is_empty() {
446 Path::new(".")
447 } else {
448 parent
449 };
450
451 let prefix = format!("{}.partial.", file_name.to_string_lossy());
452
453 let entries = match std::fs::read_dir(parent) {
454 Ok(entries) => entries,
455 Err(e) => {
456 println!(
457 "Failed to scan {} for stale partial downloads: {e}",
458 parent.display()
459 );
460 return;
461 }
462 };
463
464 for entry in entries.flatten() {
465 let name = entry.file_name();
466
467 let Some(suffix) = name.to_str().and_then(|name| name.strip_prefix(&prefix)) else {
468 continue;
469 };
470
471 let is_partial_name = suffix.split_once('.').is_some_and(|(pid, nanos)| {
473 pid.parse::<u32>().is_ok() && nanos.parse::<u128>().is_ok()
474 });
475
476 if !is_partial_name {
477 continue;
478 }
479
480 let is_stale = entry
481 .metadata()
482 .and_then(|metadata| metadata.modified())
483 .ok()
484 .and_then(|modified| modified.elapsed().ok())
485 .is_some_and(|age| age >= STALE_PARTIAL_AGE);
486
487 if !is_stale {
488 continue;
489 }
490
491 let path = entry.path();
492 match remove_file(&path) {
493 Ok(()) => println!("Removed stale partial download {}", path.display()),
494 Err(e) => println!(
495 "Failed to remove stale partial download {}: {e}",
496 path.display()
497 ),
498 }
499 }
500}
501
502fn partial_path_for(filepath: &Path) -> PathBuf {
503 let nanos = std::time::SystemTime::now()
506 .duration_since(std::time::UNIX_EPOCH)
507 .map_or(0, |d| d.as_nanos());
508 let mut p: OsString = filepath.as_os_str().to_owned();
509 p.push(format!(".partial.{}.{}", std::process::id(), nanos));
510 PathBuf::from(p)
511}
512
513fn calculate_sha256(filepath: &Path) -> anyhow::Result<String> {
514 let mut file = File::open(filepath)?;
515 let mut ctx = digest::Context::new(&digest::SHA256);
516 let mut buffer = [0u8; 4096];
517
518 loop {
519 let count = file.read(&mut buffer)?;
520 if count == 0 {
521 break;
522 }
523 ctx.update(&buffer[..count]);
524 }
525
526 let digest = ctx.finish();
527 Ok(hex::encode(digest.as_ref()))
528}
529
530fn verify_sha256_checksum(filepath: &Path, checksums: &Path) -> anyhow::Result<bool> {
531 let file = File::open(checksums)?;
532 let reader = BufReader::new(file);
533 let manifest: Value = serde_json::from_reader(reader)?;
534
535 let filename = checksum_key(filepath)?;
536 if let Some(expected_checksum) = manifest.get(filename) {
537 let expected_checksum_str = expected_checksum.as_str().with_context(|| {
538 format!(
539 "checksum for {filename} in {} is not a string",
540 checksums.display()
541 )
542 })?;
543
544 let expected_hash = expected_checksum_str
545 .strip_prefix("sha256:")
546 .unwrap_or(expected_checksum_str);
547 let calculated_checksum = calculate_sha256(filepath)?;
548 if expected_hash == calculated_checksum {
549 return Ok(true);
550 }
551 }
552
553 Ok(false)
554}
555
556fn update_sha256_checksums(
557 filepath: &Path,
558 checksums_file: &Path,
559 new_checksum: &str,
560) -> anyhow::Result<()> {
561 let mut checksums_map = if checksums_file.exists() {
562 let file = File::open(checksums_file)?;
563 let reader = BufReader::new(file);
564 match serde_json::from_reader(reader)? {
565 Value::Object(map) => map,
566 _ => anyhow::bail!(
567 "checksums file {} is not a JSON object",
568 checksums_file.display()
569 ),
570 }
571 } else {
572 serde_json::Map::new()
573 };
574
575 let filename = checksum_key(filepath)?.to_string();
577 let prefixed_checksum = format!("sha256:{new_checksum}");
578 checksums_map.insert(filename, Value::String(prefixed_checksum));
579
580 let temp_path = partial_path_for(checksums_file);
582 let result = write_checksums(&temp_path, &Value::Object(checksums_map))
583 .and_then(|()| std::fs::rename(&temp_path, checksums_file).map_err(Into::into));
584
585 if result.is_err() {
586 let _ = remove_file(&temp_path);
587 }
588
589 result
590}
591
592fn write_checksums(path: &Path, checksums: &Value) -> anyhow::Result<()> {
593 let mut writer = BufWriter::new(File::create(path)?);
594 serde_json::to_writer_pretty(&mut writer, checksums)?;
595 writer.into_inner()?.sync_all()?;
596 Ok(())
597}
598
599fn checksum_key(filepath: &Path) -> anyhow::Result<&str> {
600 filepath
601 .file_name()
602 .and_then(|name| name.to_str())
603 .with_context(|| format!("no UTF-8 file name in {}", filepath.display()))
604}
605
606#[cfg(test)]
607mod tests {
608 use std::{
609 fs,
610 io::{BufWriter, Write},
611 net::SocketAddr,
612 sync::{
613 Arc,
614 atomic::{AtomicUsize, Ordering},
615 },
616 };
617
618 use axum::{Router, http::StatusCode, routing::get, serve};
619 use rstest::*;
620 use serde_json::{json, to_writer};
621 use tempfile::TempDir;
622 use tokio::{
623 net::TcpListener,
624 task,
625 time::{Duration, sleep},
626 };
627
628 use super::*;
629
630 fn test_retry_config() -> RetryConfig {
633 RetryConfig {
634 max_retries: 5,
635 initial_delay_ms: 10,
636 max_delay_ms: 50,
637 backoff_factor: 2.0,
638 jitter_ms: 5,
639 operation_timeout_ms: Some(500),
640 immediate_first: false,
641 max_elapsed_ms: Some(2000),
642 }
643 }
644
645 #[rstest]
646 #[case::nan(f64::NAN, Duration::ZERO)]
647 #[case::negative(-1.0, Duration::ZERO)]
648 #[case::infinite(f64::INFINITY, Duration::MAX)]
649 #[case::normal(2.0, Duration::from_millis(20))]
650 fn next_retry_delay_handles_edge_factors(
651 #[case] backoff_factor: f64,
652 #[case] expected: Duration,
653 ) {
654 let delay = Duration::from_millis(10);
655
656 assert_eq!(next_retry_delay(delay, backoff_factor), expected);
657 }
658
659 async fn setup_test_server(
660 server_content: Option<String>,
661 status_code: StatusCode,
662 ) -> SocketAddr {
663 let server_content = Arc::new(server_content);
664 let server_content_clone = server_content.clone();
665 let app = Router::new().route(
666 "/testfile.txt",
667 get(move || {
668 let server_content = server_content_clone.clone();
669 async move {
670 let response_body = match &*server_content {
671 Some(content) => content.clone(),
672 None => "File not found".to_string(),
673 };
674 (status_code, response_body)
675 }
676 }),
677 );
678
679 let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
680 let addr = listener.local_addr().unwrap();
681 let server = serve(listener, app);
682
683 task::spawn(async move {
684 if let Err(e) = server.await {
685 eprintln!("server error: {e}");
686 }
687 });
688
689 sleep(Duration::from_millis(100)).await;
690
691 addr
692 }
693
694 #[rstest]
695 #[case::missing(None, "verified fixture", true)]
696 #[case::cached(Some("verified fixture"), "verified fixture", true)]
697 #[case::stale_cached(Some("stale fixture"), "verified fixture", true)]
698 #[case::corrupt_download(None, "corrupt download", false)]
699 #[tokio::test]
700 async fn test_prepare_test_data_file(
701 #[case] cached: Option<&str>,
702 #[case] downloaded: &str,
703 #[case] valid: bool,
704 ) {
705 let temp_dir = TempDir::new().unwrap();
706 let filepath = temp_dir.path().join("testfile.txt");
707 let checksums = temp_dir.path().join("checksums.json");
708 let content = "verified fixture";
709 let manifest = serde_json::to_vec(&json!({
710 "testfile.txt": format!("sha256:{}", calculate_sha256_bytes(content.as_bytes()))
711 }))
712 .unwrap();
713 fs::write(&checksums, &manifest).unwrap();
714 if let Some(cached) = cached {
715 fs::write(&filepath, cached).unwrap();
716 }
717
718 let url = if cached == Some(content) {
719 "http://127.0.0.1:0/testfile.txt".to_string()
720 } else {
721 let addr = setup_test_server(Some(downloaded.to_string()), StatusCode::OK).await;
722 format!("http://{addr}/testfile.txt")
723 };
724 let filepath_clone = filepath.clone();
725 let checksums_clone = checksums.clone();
726 let result = tokio::task::spawn_blocking(move || {
727 prepare_test_data_file(&filepath_clone, &url, &checksums_clone)
728 })
729 .await
730 .unwrap();
731
732 if valid {
733 result.unwrap();
734 assert_eq!(fs::read_to_string(&filepath).unwrap(), content);
735 } else {
736 assert_eq!(
737 result.unwrap_err().to_string(),
738 format!("Checksum mismatch for {}", filepath.display()),
739 );
740 assert!(!filepath.exists());
741 }
742 assert_eq!(fs::read(&checksums).unwrap(), manifest);
743 }
744
745 #[tokio::test]
746 async fn test_file_already_exists() {
747 let temp_dir = TempDir::new().unwrap();
748 let file_path = temp_dir.path().join("testfile.txt");
749 fs::write(&file_path, "Existing file content").unwrap();
750
751 let url = "http://example.com/testfile.txt".to_string();
752 let result = ensure_file_exists_or_download_http(&file_path, &url, None, Some(5));
753
754 assert!(result.is_ok());
755 let content = fs::read_to_string(&file_path).unwrap();
756 assert_eq!(content, "Existing file content");
757 }
758
759 #[tokio::test]
760 async fn test_download_file_success() {
761 let temp_dir = TempDir::new().unwrap();
762 let filepath = temp_dir.path().join("testfile.txt");
763 let filepath_clone = filepath.clone();
764
765 let server_content = "Server file content".to_string();
766 let status_code = StatusCode::OK;
767 let addr = setup_test_server(Some(server_content.clone()), status_code).await;
768 let url = format!("http://{addr}/testfile.txt");
769
770 let result = tokio::task::spawn_blocking(move || {
771 ensure_file_exists_or_download_http_with_config(
772 &filepath_clone,
773 &url,
774 None,
775 5,
776 Some(test_retry_config()),
777 Some(0),
778 )
779 })
780 .await
781 .unwrap();
782
783 assert!(result.is_ok());
784 let content = fs::read_to_string(&filepath).unwrap();
785 assert_eq!(content, server_content);
786 }
787
788 #[tokio::test]
789 async fn test_download_file_not_found() {
790 let temp_dir = TempDir::new().unwrap();
791 let file_path = temp_dir.path().join("testfile.txt");
792
793 let server_content = None;
794 let status_code = StatusCode::NOT_FOUND;
795 let addr = setup_test_server(server_content, status_code).await;
796 let url = format!("http://{addr}/testfile.txt");
797
798 let result = tokio::task::spawn_blocking(move || {
799 ensure_file_exists_or_download_http_with_config(
800 &file_path,
801 &url,
802 None,
803 1,
804 Some(test_retry_config()),
805 Some(0),
806 )
807 })
808 .await
809 .unwrap();
810
811 assert!(result.is_err());
812 let err_msg = format!("{}", result.unwrap_err());
813 assert!(
814 err_msg.contains("Client error: HTTP"),
815 "Unexpected error message: {err_msg}"
816 );
817 }
818
819 #[tokio::test]
820 async fn test_network_error() {
821 let temp_dir = TempDir::new().unwrap();
822 let file_path = temp_dir.path().join("testfile.txt");
823
824 let url = "http://127.0.0.1:0/testfile.txt".to_string();
826
827 let result = tokio::task::spawn_blocking(move || {
828 ensure_file_exists_or_download_http_with_config(
829 &file_path,
830 &url,
831 None,
832 2,
833 Some(test_retry_config()),
834 Some(0),
835 )
836 })
837 .await
838 .unwrap();
839
840 assert!(result.is_err());
841 let err_msg = format!("{}", result.unwrap_err());
842 assert!(
843 err_msg.contains("error"),
844 "Unexpected error message: {err_msg}"
845 );
846 }
847
848 #[tokio::test]
849 async fn test_retry_then_success_on_500() {
850 let temp_dir = TempDir::new().unwrap();
851 let filepath = temp_dir.path().join("testfile.txt");
852 let filepath_clone = filepath.clone();
853
854 let counter = Arc::new(AtomicUsize::new(0));
855 let counter_clone = counter.clone();
856
857 let app = Router::new().route(
858 "/testfile.txt",
859 get(move || {
860 let c = counter_clone.clone();
861 async move {
862 let n = c.fetch_add(1, Ordering::SeqCst);
863 if n < 2 {
864 (StatusCode::INTERNAL_SERVER_ERROR, "temporary error")
865 } else {
866 (StatusCode::OK, "eventual success")
867 }
868 }
869 }),
870 );
871
872 let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
873 let addr = listener.local_addr().unwrap();
874 let server = serve(listener, app);
875 task::spawn(async move {
876 let _ = server.await;
877 });
878 sleep(Duration::from_millis(100)).await;
879
880 let url = format!("http://{addr}/testfile.txt");
881
882 let result = tokio::task::spawn_blocking(move || {
883 ensure_file_exists_or_download_http_with_config(
884 &filepath_clone,
885 &url,
886 None,
887 5,
888 Some(test_retry_config()),
889 Some(0),
890 )
891 })
892 .await
893 .unwrap();
894
895 assert!(result.is_ok());
896 let content = std::fs::read_to_string(&filepath).unwrap();
897 assert_eq!(content, "eventual success");
898 assert!(counter.load(Ordering::SeqCst) >= 2);
899 }
900
901 #[tokio::test]
902 async fn test_retry_then_success_on_429() {
903 let temp_dir = TempDir::new().unwrap();
904 let filepath = temp_dir.path().join("testfile.txt");
905 let filepath_clone = filepath.clone();
906
907 let counter = Arc::new(AtomicUsize::new(0));
908 let counter_clone = counter.clone();
909
910 let app = Router::new().route(
911 "/testfile.txt",
912 get(move || {
913 let c = counter_clone.clone();
914 async move {
915 let n = c.fetch_add(1, Ordering::SeqCst);
916 if n < 1 {
917 (StatusCode::TOO_MANY_REQUESTS, "rate limited")
918 } else {
919 (StatusCode::OK, "ok after retry")
920 }
921 }
922 }),
923 );
924
925 let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
926 let addr = listener.local_addr().unwrap();
927 let server = serve(listener, app);
928 task::spawn(async move {
929 let _ = server.await;
930 });
931 sleep(Duration::from_millis(100)).await;
932
933 let url = format!("http://{addr}/testfile.txt");
934
935 let result = tokio::task::spawn_blocking(move || {
936 ensure_file_exists_or_download_http_with_config(
937 &filepath_clone,
938 &url,
939 None,
940 5,
941 Some(test_retry_config()),
942 Some(0),
943 )
944 })
945 .await
946 .unwrap();
947
948 assert!(result.is_ok());
949 let content = std::fs::read_to_string(&filepath).unwrap();
950 assert_eq!(content, "ok after retry");
951 assert!(counter.load(Ordering::SeqCst) >= 2);
952 }
953
954 #[tokio::test]
955 async fn test_no_retry_on_404() {
956 let temp_dir = TempDir::new().unwrap();
957 let filepath = temp_dir.path().join("testfile.txt");
958 let filepath_clone = filepath.clone();
959
960 let counter = Arc::new(AtomicUsize::new(0));
961 let counter_clone = counter.clone();
962
963 let app = Router::new().route(
964 "/testfile.txt",
965 get(move || {
966 let c = counter_clone.clone();
967 async move {
968 c.fetch_add(1, Ordering::SeqCst);
969 (StatusCode::NOT_FOUND, "missing")
970 }
971 }),
972 );
973
974 let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
975 let addr = listener.local_addr().unwrap();
976 let server = serve(listener, app);
977 task::spawn(async move {
978 let _ = server.await;
979 });
980 sleep(Duration::from_millis(100)).await;
981
982 let url = format!("http://{addr}/testfile.txt");
983
984 let result = tokio::task::spawn_blocking(move || {
985 ensure_file_exists_or_download_http_with_config(
986 &filepath_clone,
987 &url,
988 None,
989 5,
990 Some(test_retry_config()),
991 Some(0),
992 )
993 })
994 .await
995 .unwrap();
996
997 assert!(result.is_err());
998 assert_eq!(counter.load(Ordering::SeqCst), 1, "should not retry on 404");
999 }
1000
1001 #[tokio::test]
1002 async fn test_checksum_mismatch_retry_then_success() {
1003 let temp_dir = TempDir::new().unwrap();
1004 let filepath = temp_dir.path().join("testfile.txt");
1005 let filepath_clone = filepath.clone();
1006
1007 let good_content = "correct content";
1008 let good_checksum = calculate_sha256_bytes(good_content.as_bytes());
1009
1010 let checksums_path = temp_dir.path().join("checksums.json");
1011 let checksums_data = json!({
1012 "testfile.txt": format!("sha256:{good_checksum}")
1013 });
1014 let checksums_file = File::create(&checksums_path).unwrap();
1015 to_writer(BufWriter::new(checksums_file), &checksums_data).unwrap();
1016 let checksums_clone = checksums_path.clone();
1017
1018 let counter = Arc::new(AtomicUsize::new(0));
1020 let counter_clone = counter.clone();
1021
1022 let app = Router::new().route(
1023 "/testfile.txt",
1024 get(move || {
1025 let c = counter_clone.clone();
1026 async move {
1027 let n = c.fetch_add(1, Ordering::SeqCst);
1028 if n == 0 {
1029 (StatusCode::OK, "corrupt data")
1030 } else {
1031 (StatusCode::OK, "correct content")
1032 }
1033 }
1034 }),
1035 );
1036
1037 let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
1038 let addr = listener.local_addr().unwrap();
1039 let server = serve(listener, app);
1040 task::spawn(async move {
1041 let _ = server.await;
1042 });
1043 sleep(Duration::from_millis(100)).await;
1044
1045 let url = format!("http://{addr}/testfile.txt");
1046
1047 let result = tokio::task::spawn_blocking(move || {
1048 ensure_file_exists_or_download_http_with_config(
1049 &filepath_clone,
1050 &url,
1051 Some(&checksums_clone),
1052 5,
1053 Some(test_retry_config()),
1054 Some(0),
1055 )
1056 })
1057 .await
1058 .unwrap();
1059
1060 assert!(result.is_ok());
1061 let content = fs::read_to_string(&filepath).unwrap();
1062 assert_eq!(content, good_content);
1063 assert_eq!(counter.load(Ordering::SeqCst), 2);
1064 }
1065
1066 #[tokio::test]
1067 async fn test_checksum_mismatch_retry_then_fail() {
1068 let temp_dir = TempDir::new().unwrap();
1069 let filepath = temp_dir.path().join("testfile.txt");
1070 let filepath_clone = filepath.clone();
1071
1072 let checksums_path = temp_dir.path().join("checksums.json");
1074 let checksums_data = json!({
1075 "testfile.txt": "sha256:0000000000000000000000000000000000000000000000000000000000000000"
1076 });
1077 let checksums_file = File::create(&checksums_path).unwrap();
1078 to_writer(BufWriter::new(checksums_file), &checksums_data).unwrap();
1079 let checksums_clone = checksums_path.clone();
1080
1081 let counter = Arc::new(AtomicUsize::new(0));
1082 let counter_clone = counter.clone();
1083
1084 let app = Router::new().route(
1085 "/testfile.txt",
1086 get(move || {
1087 let c = counter_clone.clone();
1088 async move {
1089 c.fetch_add(1, Ordering::SeqCst);
1090 (StatusCode::OK, "always wrong content")
1091 }
1092 }),
1093 );
1094
1095 let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
1096 let addr = listener.local_addr().unwrap();
1097 let server = serve(listener, app);
1098 task::spawn(async move {
1099 let _ = server.await;
1100 });
1101 sleep(Duration::from_millis(100)).await;
1102
1103 let url = format!("http://{addr}/testfile.txt");
1104
1105 let result = tokio::task::spawn_blocking(move || {
1106 ensure_file_exists_or_download_http_with_config(
1107 &filepath_clone,
1108 &url,
1109 Some(&checksums_clone),
1110 5,
1111 Some(test_retry_config()),
1112 Some(0),
1113 )
1114 })
1115 .await
1116 .unwrap();
1117
1118 assert!(result.is_err());
1119 let err_msg = format!("{}", result.unwrap_err());
1120 assert!(err_msg.contains("Checksum mismatch after retry"));
1121 assert_eq!(
1122 counter.load(Ordering::SeqCst),
1123 2,
1124 "should download exactly twice"
1125 );
1126 assert!(!filepath.exists(), "corrupt file should be cleaned up");
1127 }
1128
1129 async fn truncated_then_full_server(good_body: &'static str) -> (SocketAddr, Arc<AtomicUsize>) {
1131 use tokio::io::{AsyncReadExt, AsyncWriteExt};
1132
1133 let counter = Arc::new(AtomicUsize::new(0));
1134 let counter_clone = counter.clone();
1135
1136 let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
1137 let addr = listener.local_addr().unwrap();
1138
1139 task::spawn(async move {
1140 loop {
1141 let Ok((mut sock, _)) = listener.accept().await else {
1142 break;
1143 };
1144 let counter = counter_clone.clone();
1145
1146 task::spawn(async move {
1147 let mut buf = [0u8; 1024];
1148 let _ = sock.read(&mut buf).await;
1149 let n = counter.fetch_add(1, Ordering::SeqCst);
1150
1151 if n == 0 {
1152 let resp = b"HTTP/1.1 200 OK\r\nContent-Length: 100\r\nConnection: close\r\n\r\nshort";
1153 let _ = sock.write_all(resp).await;
1154 } else {
1155 let resp = format!(
1156 "HTTP/1.1 200 OK\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}",
1157 good_body.len(),
1158 good_body,
1159 );
1160 let _ = sock.write_all(resp.as_bytes()).await;
1161 }
1162 let _ = sock.shutdown().await;
1163 });
1164 }
1165 });
1166
1167 sleep(Duration::from_millis(100)).await;
1168 (addr, counter)
1169 }
1170
1171 fn count_partial_siblings(filepath: &Path) -> usize {
1172 let parent = filepath.parent().unwrap();
1173 let stem = filepath.file_name().unwrap().to_string_lossy().into_owned();
1174 let prefix = format!("{stem}.partial.");
1175 fs::read_dir(parent)
1176 .unwrap()
1177 .filter_map(Result::ok)
1178 .filter(|e| e.file_name().to_string_lossy().starts_with(&prefix))
1179 .count()
1180 }
1181
1182 #[rstest]
1183 #[case::progressing(0, 400, Ok(()))]
1184 #[case::stalled_headers(1200, 0, Err("Retryable error: deadline has elapsed"))]
1185 #[case::stalled_body(
1186 0,
1187 1200,
1188 Err("Retryable error: body stream error: deadline has elapsed")
1189 )]
1190 #[tokio::test]
1191 async fn download_timeout_applies_to_each_read(
1192 #[case] header_delay_ms: u64,
1193 #[case] chunk_delay_ms: u64,
1194 #[case] expected: Result<(), &str>,
1195 ) {
1196 use tokio::io::{AsyncReadExt, AsyncWriteExt};
1197
1198 let temp_dir = TempDir::new().unwrap();
1199 let filepath = temp_dir.path().join("streamed.txt");
1200 let destination = filepath.clone();
1201 let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
1202 let addr = listener.local_addr().unwrap();
1203 let peer = task::spawn(async move {
1204 let (mut socket, _) = listener.accept().await.unwrap();
1205 let mut request = Vec::new();
1206 while !request.ends_with(b"\r\n\r\n") {
1207 request.push(socket.read_u8().await.unwrap());
1208 }
1209 sleep(Duration::from_millis(header_delay_ms)).await;
1210
1211 if socket
1212 .write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 4\r\nConnection: close\r\n\r\na")
1213 .await
1214 .is_err()
1215 {
1216 return;
1217 }
1218
1219 for byte in b"bcd" {
1220 sleep(Duration::from_millis(chunk_delay_ms)).await;
1221
1222 if socket.write_all(&[*byte]).await.is_err() {
1223 return;
1224 }
1225 }
1226 });
1227 let result = task::spawn_blocking(move || {
1228 let cfg = RetryConfig {
1229 max_retries: 0,
1230 max_elapsed_ms: None,
1231 ..test_retry_config()
1232 };
1233 download_file(
1234 &destination,
1235 &format!("http://{addr}/streamed"),
1236 1,
1237 Some(cfg),
1238 )
1239 .map_err(|e| e.to_string())
1240 })
1241 .await
1242 .unwrap();
1243 peer.await.unwrap();
1244
1245 assert_eq!(
1246 result.as_ref().map_err(String::as_str),
1247 expected.as_ref().map_err(|e| *e)
1248 );
1249 assert_eq!(count_partial_siblings(&filepath), 0);
1250 if expected.is_ok() {
1251 assert_eq!(fs::read(&filepath).unwrap(), b"abcd");
1252 } else {
1253 assert!(!filepath.exists());
1254 }
1255 }
1256
1257 #[tokio::test]
1258 async fn test_body_truncation_retries_and_recovers() {
1259 let temp_dir = TempDir::new().unwrap();
1260 let filepath = temp_dir.path().join("testfile.txt");
1261 let filepath_clone = filepath.clone();
1262
1263 let (addr, counter) = truncated_then_full_server("complete payload").await;
1264 let url = format!("http://{addr}/testfile.txt");
1265
1266 let result = tokio::task::spawn_blocking(move || {
1267 ensure_file_exists_or_download_http_with_config(
1268 &filepath_clone,
1269 &url,
1270 None,
1271 5,
1272 Some(test_retry_config()),
1273 Some(0),
1274 )
1275 })
1276 .await
1277 .unwrap();
1278
1279 assert!(result.is_ok(), "should retry past the truncated response");
1280 assert_eq!(counter.load(Ordering::SeqCst), 2);
1281 assert_eq!(
1282 count_partial_siblings(&filepath),
1283 0,
1284 "no .partial siblings must remain after success",
1285 );
1286 let content = fs::read_to_string(&filepath).unwrap();
1287 assert_eq!(content, "complete payload");
1288 }
1289
1290 #[tokio::test]
1291 async fn test_body_truncation_exhausts_retries_leaves_no_corrupt_file() {
1292 use tokio::io::{AsyncReadExt, AsyncWriteExt};
1293
1294 let temp_dir = TempDir::new().unwrap();
1295 let filepath = temp_dir.path().join("testfile.txt");
1296 let filepath_clone = filepath.clone();
1297
1298 let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
1299 let addr = listener.local_addr().unwrap();
1300
1301 task::spawn(async move {
1302 loop {
1303 let Ok((mut sock, _)) = listener.accept().await else {
1304 break;
1305 };
1306
1307 task::spawn(async move {
1308 let mut buf = [0u8; 1024];
1309 let _ = sock.read(&mut buf).await;
1310 let resp =
1311 b"HTTP/1.1 200 OK\r\nContent-Length: 100\r\nConnection: close\r\n\r\nshort";
1312 let _ = sock.write_all(resp).await;
1313 let _ = sock.shutdown().await;
1314 });
1315 }
1316 });
1317
1318 sleep(Duration::from_millis(100)).await;
1319 let url = format!("http://{addr}/testfile.txt");
1320
1321 let result = tokio::task::spawn_blocking(move || {
1322 ensure_file_exists_or_download_http_with_config(
1323 &filepath_clone,
1324 &url,
1325 None,
1326 5,
1327 Some(test_retry_config()),
1328 Some(0),
1329 )
1330 })
1331 .await
1332 .unwrap();
1333
1334 assert!(result.is_err(), "all retries should fail");
1335 assert!(!filepath.exists(), "no corrupt file at final path");
1336 assert_eq!(
1337 count_partial_siblings(&filepath),
1338 0,
1339 "no .partial siblings may leak after exhausted retries",
1340 );
1341 }
1342
1343 #[rstest]
1344 fn test_partial_path_for_is_unique_and_marked() {
1345 let target = Path::new("/tmp/data.parquet");
1346 let stem = target.file_name().unwrap().to_string_lossy().into_owned();
1347 let expected_prefix = format!("{stem}.partial.");
1348
1349 let a = partial_path_for(target);
1350 let b = partial_path_for(target);
1351 assert_ne!(a, b, "partial paths must be unique per call");
1352 for p in [&a, &b] {
1353 let name = p.file_name().unwrap().to_string_lossy().into_owned();
1354 assert!(name.starts_with(&expected_prefix), "got {name}");
1355 }
1356 }
1357
1358 #[tokio::test]
1359 async fn test_unrelated_partial_sibling_not_clobbered() {
1360 let temp_dir = TempDir::new().unwrap();
1361 let filepath = temp_dir.path().join("testfile.txt");
1362 let filepath_clone = filepath.clone();
1363 let bystander = temp_dir.path().join("testfile.txt.partial");
1365 fs::write(&bystander, b"do not touch").unwrap();
1366
1367 let server_content = "downloaded".to_string();
1368 let addr = setup_test_server(Some(server_content.clone()), StatusCode::OK).await;
1369 let url = format!("http://{addr}/testfile.txt");
1370
1371 let result = tokio::task::spawn_blocking(move || {
1372 ensure_file_exists_or_download_http_with_config(
1373 &filepath_clone,
1374 &url,
1375 None,
1376 5,
1377 Some(test_retry_config()),
1378 Some(0),
1379 )
1380 })
1381 .await
1382 .unwrap();
1383
1384 assert!(result.is_ok());
1385 assert_eq!(fs::read_to_string(&filepath).unwrap(), server_content);
1386 assert_eq!(
1387 fs::read_to_string(&bystander).unwrap(),
1388 "do not touch",
1389 "unrelated sibling file must be preserved",
1390 );
1391 }
1392
1393 #[tokio::test]
1394 async fn test_download_removes_stale_partial_downloads() {
1395 let temp_dir = TempDir::new().unwrap();
1396 let filepath = temp_dir.path().join("testfile.txt");
1397 let filepath_clone = filepath.clone();
1398 let stale_partial = temp_dir.path().join("testfile.txt.partial.123.456");
1399 File::create(&stale_partial)
1400 .unwrap()
1401 .set_modified(std::time::SystemTime::now() - Duration::from_secs(7200))
1402 .unwrap();
1403
1404 let server_content = "downloaded".to_string();
1405 let addr = setup_test_server(Some(server_content.clone()), StatusCode::OK).await;
1406 let url = format!("http://{addr}/testfile.txt");
1407
1408 let result = tokio::task::spawn_blocking(move || {
1409 ensure_file_exists_or_download_http_with_config(
1410 &filepath_clone,
1411 &url,
1412 None,
1413 5,
1414 Some(test_retry_config()),
1415 Some(0),
1416 )
1417 })
1418 .await
1419 .unwrap();
1420
1421 result.unwrap();
1422 assert_eq!(fs::read_to_string(&filepath).unwrap(), server_content);
1423 assert!(!stale_partial.exists());
1424 }
1425
1426 fn calculate_sha256_bytes(data: &[u8]) -> String {
1427 let mut ctx = digest::Context::new(&digest::SHA256);
1428 ctx.update(data);
1429 hex::encode(ctx.finish().as_ref())
1430 }
1431
1432 #[rstest]
1433 #[expect(clippy::panic_in_result_fn)]
1434 fn test_calculate_sha256() -> anyhow::Result<()> {
1435 let temp_dir = TempDir::new()?;
1436 let test_file_path = temp_dir.path().join("test_file.txt");
1437 let mut test_file = File::create(&test_file_path)?;
1438 let content = b"Hello, world!";
1439 test_file.write_all(content)?;
1440
1441 let expected_hash = "315f5bdb76d078c43b8ac0064e4a0164612b1fce77c869345bfc94c75894edd3";
1442 let calculated_hash = calculate_sha256(&test_file_path)?;
1443
1444 assert_eq!(calculated_hash, expected_hash);
1445 Ok(())
1446 }
1447
1448 #[rstest]
1449 #[expect(clippy::panic_in_result_fn)]
1450 fn test_verify_sha256_checksum() -> anyhow::Result<()> {
1451 let temp_dir = TempDir::new()?;
1452 let test_file_path = temp_dir.path().join("test_file.txt");
1453 let mut test_file = File::create(&test_file_path)?;
1454 let content = b"Hello, world!";
1455 test_file.write_all(content)?;
1456
1457 let calculated_checksum = calculate_sha256(&test_file_path)?;
1458
1459 let checksums_path = temp_dir.path().join("checksums.json");
1461 let checksums_data = json!({
1462 "test_file.txt": format!("sha256:{}", calculated_checksum)
1463 });
1464 let checksums_file = File::create(&checksums_path)?;
1465 let writer = BufWriter::new(checksums_file);
1466 to_writer(writer, &checksums_data)?;
1467
1468 let is_valid = verify_sha256_checksum(&test_file_path, &checksums_path)?;
1469 assert!(is_valid, "The checksum should be valid");
1470 Ok(())
1471 }
1472
1473 #[rstest]
1474 fn test_verify_sha256_checksum_rejects_non_string_entry() {
1475 let temp_dir = TempDir::new().unwrap();
1476 let test_file_path = temp_dir.path().join("test_file.txt");
1477 fs::write(&test_file_path, b"Hello, world!").unwrap();
1478 let checksums_path = temp_dir.path().join("checksums.json");
1479 fs::write(&checksums_path, r#"{"test_file.txt": 5}"#).unwrap();
1480
1481 let error = verify_sha256_checksum(&test_file_path, &checksums_path).unwrap_err();
1482
1483 assert_eq!(
1484 error.to_string(),
1485 format!(
1486 "checksum for test_file.txt in {} is not a string",
1487 checksums_path.display()
1488 )
1489 );
1490 }
1491
1492 #[cfg(unix)]
1493 #[rstest]
1494 fn test_verify_sha256_checksum_rejects_non_utf8_file_name() {
1495 use std::os::unix::ffi::OsStrExt;
1496
1497 let temp_dir = TempDir::new().unwrap();
1498 let test_file_path = temp_dir
1499 .path()
1500 .join(std::ffi::OsStr::from_bytes(b"bad\xff.txt"));
1501 let checksums_path = temp_dir.path().join("checksums.json");
1502 fs::write(&checksums_path, "{}").unwrap();
1503
1504 let error = verify_sha256_checksum(&test_file_path, &checksums_path).unwrap_err();
1505
1506 assert_eq!(
1507 error.to_string(),
1508 format!("no UTF-8 file name in {}", test_file_path.display())
1509 );
1510 }
1511
1512 #[rstest]
1513 fn test_update_sha256_checksums_replaces_manifest_without_leftovers() {
1514 let temp_dir = TempDir::new().unwrap();
1515 let checksums_path = temp_dir.path().join("checksums.json");
1516 fs::write(&checksums_path, r#"{"existing.txt": "sha256:aaa"}"#).unwrap();
1517
1518 update_sha256_checksums(&temp_dir.path().join("new.txt"), &checksums_path, "bbb").unwrap();
1519
1520 let manifest: Value =
1521 serde_json::from_str(&fs::read_to_string(&checksums_path).unwrap()).unwrap();
1522 assert_eq!(
1523 manifest,
1524 json!({"existing.txt": "sha256:aaa", "new.txt": "sha256:bbb"})
1525 );
1526 let entries: Vec<_> = fs::read_dir(temp_dir.path())
1527 .unwrap()
1528 .map(|entry| entry.unwrap().file_name())
1529 .collect();
1530 assert_eq!(entries, vec![OsString::from("checksums.json")]);
1531 }
1532
1533 #[rstest]
1534 fn test_update_sha256_checksums_rejects_non_object_manifest() {
1535 let temp_dir = TempDir::new().unwrap();
1536 let checksums_path = temp_dir.path().join("checksums.json");
1537 fs::write(&checksums_path, "[]").unwrap();
1538
1539 let error =
1540 update_sha256_checksums(&temp_dir.path().join("new.txt"), &checksums_path, "bbb")
1541 .unwrap_err();
1542
1543 assert_eq!(
1544 error.to_string(),
1545 format!(
1546 "checksums file {} is not a JSON object",
1547 checksums_path.display()
1548 )
1549 );
1550 assert_eq!(fs::read_to_string(&checksums_path).unwrap(), "[]");
1551 }
1552
1553 #[rstest]
1554 fn test_remove_stale_partials_only_removes_old_download_partials() {
1555 let temp_dir = TempDir::new().unwrap();
1556 let filepath = temp_dir.path().join("testfile.txt");
1557 let two_hours_ago = std::time::SystemTime::now() - Duration::from_secs(7200);
1558
1559 let write_file = |name: &str, stale: bool| {
1560 let path = temp_dir.path().join(name);
1561 let file = File::create(&path).unwrap();
1562
1563 if stale {
1564 file.set_modified(two_hours_ago).unwrap();
1565 }
1566
1567 path
1568 };
1569
1570 let stale_partial = write_file("testfile.txt.partial.123.456", true);
1571 let fresh_partial = write_file("testfile.txt.partial.124.457", false);
1572 let bystander = write_file("testfile.txt.partial", true);
1573 let non_numeric = write_file("testfile.txt.partial.abc.def", true);
1574 let other_target = write_file("other.txt.partial.123.456", true);
1575
1576 remove_stale_partials(&filepath);
1577
1578 assert!(!stale_partial.exists());
1579 assert!(fresh_partial.exists());
1580 assert!(bystander.exists());
1581 assert!(non_numeric.exists());
1582 assert!(other_target.exists());
1583 }
1584}