18346462606

Committed 08 Oct 2025 01:33PM UTC coverage: 54.854% (-0.9%) from 55.73%

Build # 18346462606

Build Type

Pull #45

github

Committed by

web-flow

Commit Message

Merge 9146fc44e into c1d3177fb

Pull Request Pull Request #45: HTTP client with configurable IPFS gateways into its own module

Run Details

399 of 536 new or added lines in 11 files covered. (74.44%)

1 existing line in 1 file now uncovered.

3554 of 6479 relevant lines covered (54.85%)

2.92 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

81.58

/src/httpclient/stream.rs

use anyhow::Result;
use async_compression::tokio::bufread::GzipDecoder;
use futures_util::TryStreamExt;
use std::path::Path;
use tokio::io::{AsyncRead, AsyncWriteExt, BufReader};
use tokio_util::io::StreamReader;

pub(crate) async fn stream_reader_to_file<R: AsyncRead + Unpin>(
    reader: &mut R,
    file: &mut tokio::fs::File,
    file_path: &Path,
) -> anyhow::Result<std::path::PathBuf> {
    tokio::io::copy(reader, file).await.map_err(|e| {
        anyhow::anyhow!(
            "Failed to stream content to file {}: {}",
            file_path.display(),
            e
        )
    })?;
    file.flush()
        .await
        .map_err(|e| anyhow::anyhow!("Failed to flush file {}: {}", file_path.display(), e))?;
    Ok(file_path.to_path_buf())
}

pub(crate) async fn stream_http_to_file(
    response: reqwest::Response,
    file_path: &Path,
) -> anyhow::Result<std::path::PathBuf> {
    let stream = response.bytes_stream().map_err(std::io::Error::other);
    let mut reader = StreamReader::new(stream);

    let mut file_path = file_path.to_path_buf();
    let (detected_ext, prefix_buf) =
        crate::content::extensions::detect_extension_from_stream(&mut reader).await;
    if !crate::content::extensions::has_known_extension(&file_path) {
        if let Some(detected_ext) = detected_ext {
            let current_path_str = file_path.to_string_lossy();
            tracing::debug!("Appending detected media extension: {}", detected_ext);
            file_path = std::path::PathBuf::from(format!("{current_path_str}.{detected_ext}"));
        }
    }

    let mut file = tokio::fs::File::create(&file_path)
        .await
        .map_err(|e| anyhow::anyhow!("Failed to create file {}: {}", file_path.display(), e))?;
    if !prefix_buf.is_empty() {
        file.write_all(&prefix_buf).await.map_err(|e| {
            anyhow::anyhow!(
                "Failed to write prefix to file {}: {}",
                file_path.display(),
                e
            )
        })?;
    }
    let result = stream_reader_to_file(&mut reader, &mut file, &file_path).await?;

    file.sync_all().await.map_err(|e| {
        anyhow::anyhow!("Failed to sync file {} to disk: {}", file_path.display(), e)
    })?;

    Ok(result)
}

pub(crate) async fn stream_gzip_http_to_file(
    response: reqwest::Response,
    file_path: &Path,
) -> Result<std::path::PathBuf> {
    let mut file = tokio::fs::File::create(file_path).await?;
    let stream = response.bytes_stream().map_err(std::io::Error::other);
    let reader = StreamReader::new(stream);
    let mut decoder = GzipDecoder::new(BufReader::new(reader));
    stream_reader_to_file(&mut decoder, &mut file, file_path).await
}

#[cfg(test)]
mod tests {
    use super::*;
    use tempfile::TempDir;
    use wiremock::{
        matchers::{method, path},
        Mock, MockServer, ResponseTemplate,
    };

    #[tokio::test]
    async fn test_large_file_streaming() {
        let mock_server = MockServer::start().await;
        let url = format!("{}/large-file", mock_server.uri());

        let large_body = "x".repeat(1024 * 1024);
        Mock::given(method("GET"))
            .and(path("/large-file"))
            .respond_with(ResponseTemplate::new(200).set_body_string(large_body))
            .mount(&mock_server)
            .await;

        let temp_dir = TempDir::new().unwrap();
        let file_path = temp_dir.path().join("large.txt");

        let client = crate::httpclient::HttpClient::new();
        let (resp, _status) = client.try_fetch_response(&url).await;
        let resp = resp.expect("response ok");
        let result = stream_http_to_file(resp, &file_path).await;
        assert!(result.is_ok());
        let metadata = std::fs::metadata(&file_path).unwrap();
        assert_eq!(metadata.len(), 1024 * 1024);
    }
}

1	use anyhow::Result;
2	use async_compression::tokio::bufread::GzipDecoder;
3	use futures_util::TryStreamExt;
4	use std::path::Path;
5	use tokio::io::{AsyncRead, AsyncWriteExt, BufReader};
6	use tokio_util::io::StreamReader;
7
8	pub(crate) async fn stream_reader_to_file<R: AsyncRead + Unpin>(	28✔
9	reader: &mut R,	28✔
10	file: &mut tokio::fs::File,	28✔
11	file_path: &Path,	28✔
12	) -> anyhow::Result<std::path::PathBuf> {	28✔
13	tokio::io::copy(reader, file).await.map_err(\|e\| {	28✔
NEW 14	anyhow::anyhow!(	×
NEW 15	"Failed to stream content to file {}: {}",	×
NEW 16	file_path.display(),	×
17	e
18	)
NEW 19	})?;	×
20	file.flush()	28✔
21	.await	28✔
22	.map_err(\|e\| anyhow::anyhow!("Failed to flush file {}: {}", file_path.display(), e))?;	28✔
23	Ok(file_path.to_path_buf())	28✔
24	}	28✔
25
26	pub(crate) async fn stream_http_to_file(	19✔
27	response: reqwest::Response,	19✔
28	file_path: &Path,	19✔
29	) -> anyhow::Result<std::path::PathBuf> {	19✔
30	let stream = response.bytes_stream().map_err(std::io::Error::other);	19✔
31	let mut reader = StreamReader::new(stream);	19✔
32
33	let mut file_path = file_path.to_path_buf();	19✔
34	let (detected_ext, prefix_buf) =	19✔
35	crate::content::extensions::detect_extension_from_stream(&mut reader).await;	19✔
36	if !crate::content::extensions::has_known_extension(&file_path) {	19✔
37	if let Some(detected_ext) = detected_ext {	5✔
NEW 38	let current_path_str = file_path.to_string_lossy();	×
NEW 39	tracing::debug!("Appending detected media extension: {}", detected_ext);	×
NEW 40	file_path = std::path::PathBuf::from(format!("{current_path_str}.{detected_ext}"));	×
41	}	5✔
42	}	14✔
43
44	let mut file = tokio::fs::File::create(&file_path)	19✔
45	.await	19✔
46	.map_err(\|e\| anyhow::anyhow!("Failed to create file {}: {}", file_path.display(), e))?;	19✔
47	if !prefix_buf.is_empty() {	19✔
48	file.write_all(&prefix_buf).await.map_err(\|e\| {	19✔
NEW 49	anyhow::anyhow!(	×
NEW 50	"Failed to write prefix to file {}: {}",	×
NEW 51	file_path.display(),	×
52	e
53	)
NEW 54	})?;	×
NEW 55	}	×
56	let result = stream_reader_to_file(&mut reader, &mut file, &file_path).await?;	19✔
57
58	file.sync_all().await.map_err(\|e\| {	19✔
NEW 59	anyhow::anyhow!("Failed to sync file {} to disk: {}", file_path.display(), e)	×
NEW 60	})?;	×
61
62	Ok(result)	19✔
63	}	19✔
64
65	pub(crate) async fn stream_gzip_http_to_file(	9✔
66	response: reqwest::Response,	9✔
67	file_path: &Path,	9✔
68	) -> Result<std::path::PathBuf> {	9✔
69	let mut file = tokio::fs::File::create(file_path).await?;	9✔
70	let stream = response.bytes_stream().map_err(std::io::Error::other);	9✔
71	let reader = StreamReader::new(stream);	9✔
72	let mut decoder = GzipDecoder::new(BufReader::new(reader));	9✔
73	stream_reader_to_file(&mut decoder, &mut file, file_path).await	9✔
74	}	9✔
75
76	#[cfg(test)]
77	mod tests {
78	use super::*;
79	use tempfile::TempDir;
80	use wiremock::{
81	matchers::{method, path},
82	Mock, MockServer, ResponseTemplate,
83	};
84
85	#[tokio::test]
86	async fn test_large_file_streaming() {	1✔
87	let mock_server = MockServer::start().await;	1✔
88	let url = format!("{}/large-file", mock_server.uri());	1✔
89
90	let large_body = "x".repeat(1024 * 1024);	1✔
91	Mock::given(method("GET"))	1✔
92	.and(path("/large-file"))	1✔
93	.respond_with(ResponseTemplate::new(200).set_body_string(large_body))	1✔
94	.mount(&mock_server)	1✔
95	.await;	1✔
96
97	let temp_dir = TempDir::new().unwrap();	1✔
98	let file_path = temp_dir.path().join("large.txt");	1✔
99
100	let client = crate::httpclient::HttpClient::new();	1✔
101	let (resp, _status) = client.try_fetch_response(&url).await;	1✔
102	let resp = resp.expect("response ok");	1✔
103	let result = stream_http_to_file(resp, &file_path).await;	1✔
104	assert!(result.is_ok());	1✔
105	let metadata = std::fs::metadata(&file_path).unwrap();	1✔
106	assert_eq!(metadata.len(), 1024 * 1024);	1✔
107	}	1✔
108	}

0xmichalis / nftbk / 18346462606

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous