Skip to main content

nautilus_testkit/
files.rs

1// -------------------------------------------------------------------------------------------------
2//  Copyright (C) 2015-2026 Nautech Systems Pty Ltd. All rights reserved.
3//  https://nautechsystems.io
4//
5//  Licensed under the GNU Lesser General Public License Version 3.0 (the "License");
6//  You may not use this file except in compliance with the License.
7//  You may obtain a copy of the License at https://www.gnu.org/licenses/lgpl-3.0.en.html
8//
9//  Unless required by applicable law or agreed to in writing, software
10//  distributed under the License is distributed on an "AS IS" BASIS,
11//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12//  See the License for the specific language governing permissions and
13//  limitations under the License.
14// -------------------------------------------------------------------------------------------------
15
16use 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
111/// Downloads missing large test fixtures and verifies every file against the checksum manifest.
112///
113/// Replaces cached files whose checksums differ. The manifest remains unchanged.
114///
115/// # Errors
116///
117/// Returns an error if the manifest cannot be read, a download fails, or a checksum differs.
118pub 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
148/// Ensures that a file exists at the specified path by downloading it if necessary.
149///
150/// If the file already exists, it checks the integrity of the file using a SHA-256 checksum
151/// from the optional `checksums` file. If the checksum is valid, the function exits early. If
152/// the checksum is invalid or missing, the function updates the checksums file with the correct
153/// hash for the existing file without redownloading it.
154///
155/// If the file does not exist, it downloads the file from the specified `url`. When a
156/// `checksums` file is provided, the download must match its entry for the file (retried once on
157/// mismatch), and a missing entry counts as a mismatch. Downloads never update the checksums file.
158///
159/// The `timeout_secs` parameter specifies the timeout in seconds for the HTTP request.
160/// If `None` is provided, a default timeout of 30 seconds will be used.
161///
162/// # Errors
163///
164/// Returns an error if:
165/// - The HTTP request cannot be sent or returns a non-success status code.
166/// - Any I/O operation fails during file creation, reading, or writing.
167/// - Checksum verification or JSON parsing fails.
168pub 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
184/// Ensures that a file exists at the specified path by downloading it if necessary, with a custom timeout.
185///
186/// # Errors
187///
188/// Returns an error if:
189/// - The HTTP request cannot be sent or returns a non-success status code after retries.
190/// - Any I/O operation fails during file creation, reading, or writing.
191/// - Checksum verification or JSON parsing fails.
192pub 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
208/// Ensures that a file exists at the specified path by downloading it if necessary,
209/// with custom timeout, retry config, and initial jitter delay.
210///
211/// # Parameters
212///
213/// - `filepath`: The path where the file should exist.
214/// - `url`: The URL to download from if the file doesn't exist.
215/// - `checksums`: Optional path to checksums file for verification.
216/// - `timeout_secs`: Timeout in seconds for response headers and each body read.
217/// - `retry_config`: Optional custom retry configuration (uses sensible defaults if None).
218/// - `initial_jitter_ms`: Optional initial jitter delay in milliseconds before download (defaults to 100-600ms if None).
219///
220/// # Errors
221///
222/// Returns an error if:
223/// - The HTTP request cannot be sent or returns a non-success status code after retries.
224/// - Any I/O operation fails during file creation, reading, or writing.
225/// - Checksum verification or JSON parsing fails.
226pub 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    // Local/cached file path: accept the file if it exists, updating the
235    // checksum record when it differs (e.g. after local regeneration).
236    // This is intentionally lenient - download verification below is strict.
237    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    // Add a small random delay to avoid bursting the remote server when
257    // many downloads start concurrently. Can be disabled by passing Some(0).
258    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    // Verify checksum after download, retry once on mismatch (corrupt download)
273    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    // Validate HTTPS for security in production builds,
307    // HTTP is intentionally allowed in test builds for local test servers (127.0.0.1),
308    // CodeQL flags this as "non-https-url" but it's a deliberate design choice for testkit.
309    #[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        // Default production config
330        let max_retries = 5u32;
331        let op_timeout_ms = timeout_secs.saturating_mul(1000);
332        // Make the provided timeout a hard ceiling for total elapsed time.
333        // Split it across attempts (at least 1000 ms per attempt) and cap total at op_timeout_ms.
334        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        // Discard any leftover partial from a prior attempt before writing
352        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                    // Preserve existing error text used by tests
371                    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
388// Stream errors are transient (Retryable), local write errors like disk full are not
389fn 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        // Persist before the rename makes the file visible at `filepath`
411        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
436// Active downloads keep writing and time out well within this age, so old partials are orphaned
437fn 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    // A bare file name has an empty parent, which names the current directory
445    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        // Only names produced by `partial_path_for`: `<pid>.<nanos>`
472        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    // Suffix with pid + nanos so the staging path is unique per invocation,
504    // preventing collision with a real user file or another concurrent downloader.
505    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    // Add or update the checksum
576    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    // Replace via a synced sibling file so a crash or full disk never truncates the manifest
581    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    /// Creates a fast, deterministic retry config for tests.
631    /// Uses very short delays to make tests run quickly without introducing flakiness.
632    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        // Use an unreachable address to simulate a network error
825        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        // First request returns corrupt data, second returns correct data
1019        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        // Checksum for content that the server will never return
1073        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    /// First call overstates Content-Length to fail the body read mid-stream; subsequent calls succeed.
1130    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        // A user-owned file whose name happens to start with `<target>.partial`
1364        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        // Create checksums.json containing the checksum
1460        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}