//! Local filesystem storage implementation use std::collections::HashMap; use std::path::{Path, PathBuf}; use std::time::UNIX_EPOCH; use async_trait::async_trait; use chrono::{DateTime, Utc}; use serde::{Deserialize, Serialize}; use tokio::fs; use tracing::{debug, info}; use crate::error::{StorageError, StorageResult}; use crate::{Storage, StorageMetadata}; /// Configuration for local filesystem storage #[derive(Debug, Clone, Serialize, Deserialize)] #[allow(clippy::module_name_repetitions)] pub struct LocalStorageConfig { /// Base directory for storage operations pub base_path: PathBuf, /// Enable atomic writes using temporary files pub atomic_writes: bool, /// Enable file locking for concurrent access pub enable_locking: bool, /// Create parent directories if they don't exist pub create_directories: bool, /// File permissions (Unix only) pub file_permissions: Option, /// Directory permissions (Unix only) pub directory_permissions: Option, /// Enable compression for stored files pub enable_compression: bool, /// Buffer size for file operations pub buffer_size: usize, } impl Default for LocalStorageConfig { fn default() -> Self { Self { base_path: PathBuf::from("./storage"), atomic_writes: true, enable_locking: true, create_directories: true, file_permissions: Some(0o644), directory_permissions: Some(0o755), enable_compression: false, buffer_size: 8192, } } } impl LocalStorageConfig { /// Load configuration from environment variables /// /// # Errors /// /// This function currently never returns an error but is marked as Result for future extensibility. pub fn from_env() -> StorageResult { let mut config = Self::default(); if let Ok(base_path) = std::env::var("STORAGE_LOCAL_BASE_PATH") { config.base_path = PathBuf::from(base_path); } if let Ok(atomic_str) = std::env::var("STORAGE_LOCAL_ATOMIC_WRITES") { config.atomic_writes = atomic_str.to_lowercase() == "true"; } if let Ok(locking_str) = std::env::var("STORAGE_LOCAL_ENABLE_LOCKING") { config.enable_locking = locking_str.to_lowercase() == "true"; } if let Ok(create_str) = std::env::var("STORAGE_LOCAL_CREATE_DIRECTORIES") { config.create_directories = create_str.to_lowercase() == "true"; } if let Ok(compression_str) = std::env::var("STORAGE_LOCAL_ENABLE_COMPRESSION") { config.enable_compression = compression_str.to_lowercase() == "true"; } if let Ok(buffer_str) = std::env::var("STORAGE_LOCAL_BUFFER_SIZE") { if let Ok(buffer_size) = buffer_str.parse::() { config.buffer_size = buffer_size; } } Ok(config) } /// Validate the configuration /// /// # Errors /// /// Returns `StorageError::ConfigError` if: /// - Base path is not absolute /// - Buffer size is 0 pub fn validate(&self) -> StorageResult<()> { if !self.base_path.is_absolute() { return Err(StorageError::ConfigError { message: "Base path must be absolute".to_owned(), }); } if self.buffer_size == 0 { return Err(StorageError::ConfigError { message: "Buffer size must be greater than 0".to_owned(), }); } Ok(()) } } /// File operation types for metrics and logging #[derive(Debug, Clone, Copy)] pub enum FileOperation { /// Read operation - retrieving file contents Read, /// Write operation - storing file contents Write, /// Delete operation - removing files Delete, /// List operation - listing directory contents List, /// Exists operation - checking file existence Exists, /// Metadata operation - retrieving file metadata Metadata, } impl FileOperation { /// Convert the operation to a string representation pub fn as_str(&self) -> &'static str { match self { FileOperation::Read => "read", FileOperation::Write => "write", FileOperation::Delete => "delete", FileOperation::List => "list", FileOperation::Exists => "exists", FileOperation::Metadata => "metadata", } } } /// Local filesystem storage implementation #[allow(clippy::module_name_repetitions)] pub struct LocalStorage { config: LocalStorageConfig, base_path: PathBuf, } impl LocalStorage { /// Create new local storage instance /// /// # Errors /// Returns error if the operation fails /// /// # Errors /// Returns error if the operation fails pub async fn new(config: LocalStorageConfig) -> StorageResult { config.validate()?; let base_path = config.base_path.clone(); // Create base directory if it doesn't exist if config.create_directories { Self::ensure_directory_exists(&base_path, config.directory_permissions).await?; } info!("Local storage initialized at: {}", base_path.display()); Ok(Self { config, base_path }) } /// Get the full filesystem path for a storage path fn get_full_path(&self, path: &str) -> PathBuf { // Sanitize path to prevent directory traversal let sanitized = path.replace("..", "").replace("\\", "/"); let sanitized = sanitized.trim_start_matches('/'); self.base_path.join(sanitized) } /// Ensure a directory exists, creating it if necessary async fn ensure_directory_exists(path: &Path, permissions: Option) -> StorageResult<()> { if !path.exists() { fs::create_dir_all(path) .await .map_err(|e| StorageError::IoError { message: format!("Failed to create directory {}: {}", path.display(), e), })?; // Set permissions on Unix systems #[cfg(unix)] if let Some(perms) = permissions { use std::os::unix::fs::PermissionsExt; let std_perms = std::fs::Permissions::from_mode(perms); std::fs::set_permissions(path, std_perms).map_err(|e| StorageError::IoError { message: format!( "Failed to set directory permissions for {}: {}", path.display(), e ), })?; } debug!("Created directory: {}", path.display()); } Ok(()) } /// Compress data if compression is enabled fn compress_data(&self, data: &[u8]) -> StorageResult> { if !self.config.enable_compression { return Ok(data.to_vec()); } use flate2::write::GzEncoder; use flate2::Compression; use std::io::Write; let mut encoder = GzEncoder::new(Vec::new(), Compression::default()); encoder .write_all(data) .map_err(|e| StorageError::CompressionError { message: format!("Compression failed: {}", e), })?; let compressed = encoder .finish() .map_err(|e| StorageError::CompressionError { message: format!("Compression finalization failed: {}", e), })?; debug!( "Compressed {} bytes to {} bytes", data.len(), compressed.len() ); Ok(compressed) } /// Decompress data if it was compressed fn decompress_data(&self, data: &[u8]) -> StorageResult> { if !self.config.enable_compression { return Ok(data.to_vec()); } use flate2::read::GzDecoder; use std::io::Read; let mut decoder = GzDecoder::new(data); let mut decompressed = Vec::new(); decoder .read_to_end(&mut decompressed) .map_err(|e| StorageError::CompressionError { message: format!("Decompression failed: {}", e), })?; debug!( "Decompressed {} bytes to {} bytes", data.len(), decompressed.len() ); Ok(decompressed) } /// Write data to file with optional atomic operation async fn write_file_data(&self, full_path: &Path, data: &[u8]) -> StorageResult<()> { // Ensure parent directory exists if let Some(parent) = full_path.parent() { if self.config.create_directories { Self::ensure_directory_exists(parent, self.config.directory_permissions).await?; } } let compressed_data = self.compress_data(data)?; if self.config.atomic_writes { // Write to temporary file first, then rename let temp_path = full_path.with_extension(format!( "tmp.{}.{}", std::process::id(), full_path .extension() .and_then(|s| s.to_str()) .unwrap_or("bin") )); fs::write(&temp_path, &compressed_data) .await .map_err(|e| StorageError::IoError { message: format!( "Failed to write temporary file {}: {}", temp_path.display(), e ), })?; // Set file permissions #[cfg(unix)] if let Some(perms) = self.config.file_permissions { use std::os::unix::fs::PermissionsExt; let std_perms = std::fs::Permissions::from_mode(perms); std::fs::set_permissions(&temp_path, std_perms).map_err(|e| { StorageError::IoError { message: format!( "Failed to set file permissions for {}: {}", temp_path.display(), e ), } })?; } // Atomic rename fs::rename(&temp_path, full_path) .await .map_err(|e| StorageError::IoError { message: format!( "Failed to rename {} to {}: {}", temp_path.display(), full_path.display(), e ), })?; debug!( "Atomically wrote {} bytes to {}", compressed_data.len(), full_path.display() ); } else { // Direct write fs::write(full_path, &compressed_data) .await .map_err(|e| StorageError::IoError { message: format!("Failed to write file {}: {}", full_path.display(), e), })?; // Set file permissions #[cfg(unix)] if let Some(perms) = self.config.file_permissions { use std::os::unix::fs::PermissionsExt; let std_perms = std::fs::Permissions::from_mode(perms); std::fs::set_permissions(full_path, std_perms).map_err(|e| { StorageError::IoError { message: format!( "Failed to set file permissions for {}: {}", full_path.display(), e ), } })?; } debug!( "Wrote {} bytes to {}", compressed_data.len(), full_path.display() ); } Ok(()) } /// Record operation metrics fn record_metrics(operation: FileOperation, duration: std::time::Duration, success: bool) { crate::metrics::get_metrics().record_operation( operation.as_str(), "local", duration, success, ); } } #[async_trait] impl Storage for LocalStorage { async fn store(&self, path: &str, data: &[u8]) -> StorageResult<()> { let start = std::time::Instant::now(); let full_path = self.get_full_path(path); let result = self.write_file_data(&full_path, data).await; let duration = start.elapsed(); Self::record_metrics(FileOperation::Write, duration, result.is_ok()); if result.is_ok() { crate::metrics::get_metrics().record_transfer( "store", "local", data.len() as u64, duration, ); } result } async fn retrieve(&self, path: &str) -> StorageResult> { let start = std::time::Instant::now(); let full_path = self.get_full_path(path); let result: StorageResult> = async { let compressed_data = fs::read(&full_path).await.map_err(|e| match e.kind() { std::io::ErrorKind::NotFound => StorageError::NotFound { path: path.to_string(), }, std::io::ErrorKind::PermissionDenied => StorageError::PermissionDenied { path: path.to_string(), }, _ => StorageError::IoError { message: format!("Failed to read file {}: {}", full_path.display(), e), }, })?; let data = self.decompress_data(&compressed_data)?; debug!( "Retrieved {} bytes from {}", data.len(), full_path.display() ); Ok(data) } .await; let duration = start.elapsed(); Self::record_metrics(FileOperation::Read, duration, result.is_ok()); if let Ok(ref data) = result { crate::metrics::get_metrics().record_transfer( "retrieve", "local", data.len() as u64, duration, ); } result } async fn exists(&self, path: &str) -> StorageResult { let start = std::time::Instant::now(); let full_path = self.get_full_path(path); let exists = full_path.exists(); debug!("Path {} exists: {}", path, exists); let duration = start.elapsed(); Self::record_metrics(FileOperation::Exists, duration, true); Ok(exists) } async fn delete(&self, path: &str) -> StorageResult { let start = std::time::Instant::now(); let full_path = self.get_full_path(path); let result = if full_path.exists() { fs::remove_file(&full_path) .await .map_err(|e| match e.kind() { std::io::ErrorKind::NotFound => StorageError::NotFound { path: path.to_string(), }, std::io::ErrorKind::PermissionDenied => StorageError::PermissionDenied { path: path.to_string(), }, _ => StorageError::IoError { message: format!("Failed to delete file {}: {}", full_path.display(), e), }, })?; debug!("Deleted file: {}", full_path.display()); Ok(true) } else { debug!("File not found for deletion: {}", full_path.display()); Ok(false) }; let duration = start.elapsed(); Self::record_metrics(FileOperation::Delete, duration, result.is_ok()); result } async fn list(&self, prefix: &str) -> StorageResult> { let start = std::time::Instant::now(); let prefix_path = self.get_full_path(prefix); let result = async { let mut paths = Vec::new(); // Handle both directory listing and prefix matching let (search_dir, file_prefix) = if prefix_path.is_dir() { (prefix_path, String::new()) } else { let parent = match prefix_path.parent() { Some(p) => p, None => &self.base_path, }; let filename = prefix_path .file_name() .and_then(|n| n.to_str()) .unwrap_or("_invalid_") .to_string(); (parent.to_path_buf(), filename) }; if !search_dir.exists() { return Ok(paths); } // Recursive helper function to traverse directories fn collect_files_recursive( dir: &std::path::Path, base_path: &std::path::Path, file_prefix: &str, paths: &mut Vec, ) -> Result<(), std::io::Error> { for entry in std::fs::read_dir(dir)? { let entry = entry?; let entry_path = entry.path(); if entry_path.is_file() { if let Some(filename) = entry_path.file_name().and_then(|n| n.to_str()) { if file_prefix.is_empty() || filename.starts_with(file_prefix) { // Convert back to relative path if let Ok(relative) = entry_path.strip_prefix(base_path) { if let Some(path_str) = relative.to_str() { paths.push(path_str.replace('\\', "/")); } } } } } else if entry_path.is_dir() { // Recursively collect from subdirectories collect_files_recursive(&entry_path, base_path, file_prefix, paths)?; } } Ok(()) } // Use recursive collection collect_files_recursive(&search_dir, &self.base_path, &file_prefix, &mut paths) .map_err(|e| StorageError::IoError { message: format!("Failed to collect files recursively: {}", e), })?; paths.sort(); debug!("Listed {} files with prefix '{}'", paths.len(), prefix); Ok(paths) } .await; let duration = start.elapsed(); Self::record_metrics(FileOperation::List, duration, result.is_ok()); result } async fn metadata(&self, path: &str) -> StorageResult { let start = std::time::Instant::now(); let full_path = self.get_full_path(path); let result = async { let metadata = fs::metadata(&full_path).await.map_err(|e| match e.kind() { std::io::ErrorKind::NotFound => StorageError::NotFound { path: path.to_string(), }, std::io::ErrorKind::PermissionDenied => StorageError::PermissionDenied { path: path.to_string(), }, _ => StorageError::IoError { message: format!("Failed to get metadata for {}: {}", full_path.display(), e), }, })?; let last_modified_dt = match metadata.modified() { Ok(modified_time) => { match modified_time.duration_since(UNIX_EPOCH) { Ok(duration) => DateTime::from_timestamp(duration.as_secs() as i64, 0) .unwrap_or_else(Utc::now), Err(_) => { Utc::now() // If duration calculation fails, use current time }, } }, Err(_) => { Utc::now() // If modified time not available, use current time }, }; let storage_metadata = StorageMetadata { path: path.to_string(), size: metadata.len(), content_type: Some("application/octet-stream".to_owned()), last_modified: last_modified_dt, etag: None, // Could implement checksum-based ETag tags: HashMap::new(), }; debug!("Retrieved metadata for {}: {} bytes", path, metadata.len()); Ok(storage_metadata) } .await; let duration = start.elapsed(); Self::record_metrics(FileOperation::Metadata, duration, result.is_ok()); result } } #[cfg(test)] mod tests { use super::*; use tempfile::TempDir; async fn create_test_storage() -> (LocalStorage, TempDir) { let temp_dir = TempDir::new().unwrap(); let config = LocalStorageConfig { base_path: temp_dir.path().to_path_buf(), ..Default::default() }; let storage = LocalStorage::new(config).await.unwrap(); (storage, temp_dir) } #[tokio::test] async fn test_store_and_retrieve() { let (storage, _temp_dir) = create_test_storage().await; let test_data = b"Hello, World!"; storage.store("test.txt", test_data).await.unwrap(); let retrieved = storage.retrieve("test.txt").await.unwrap(); assert_eq!(retrieved, test_data); } #[tokio::test] async fn test_exists() { let (storage, _temp_dir) = create_test_storage().await; assert!(!storage.exists("nonexistent.txt").await.unwrap()); storage.store("test.txt", b"data").await.unwrap(); assert!(storage.exists("test.txt").await.unwrap()); } #[tokio::test] async fn test_delete() { let (storage, _temp_dir) = create_test_storage().await; storage.store("test.txt", b"data").await.unwrap(); assert!(storage.exists("test.txt").await.unwrap()); let deleted = storage.delete("test.txt").await.unwrap(); assert!(deleted); assert!(!storage.exists("test.txt").await.unwrap()); // Deleting non-existent file should return false let deleted = storage.delete("nonexistent.txt").await.unwrap(); assert!(!deleted); } #[tokio::test] async fn test_list() { let (storage, _temp_dir) = create_test_storage().await; storage.store("file1.txt", b"data1").await.unwrap(); storage.store("file2.txt", b"data2").await.unwrap(); storage.store("other.dat", b"data3").await.unwrap(); let all_files = storage.list("").await.unwrap(); assert_eq!(all_files.len(), 3); assert!(all_files.contains(&"file1.txt".to_owned())); assert!(all_files.contains(&"file2.txt".to_owned())); assert!(all_files.contains(&"other.dat".to_owned())); } #[tokio::test] async fn test_metadata() { let (storage, _temp_dir) = create_test_storage().await; let test_data = b"Hello, World!"; storage.store("test.txt", test_data).await.unwrap(); let metadata = storage.metadata("test.txt").await.unwrap(); assert_eq!(metadata.path, "test.txt"); assert!(metadata.size > 0); // May be larger due to compression assert_eq!( metadata.content_type, Some("application/octet-stream".to_owned()) ); } #[tokio::test] async fn test_nested_directories() { let (storage, _temp_dir) = create_test_storage().await; storage .store("nested/dir/file.txt", b"nested data") .await .unwrap(); let retrieved = storage.retrieve("nested/dir/file.txt").await.unwrap(); assert_eq!(retrieved, b"nested data"); assert!(storage.exists("nested/dir/file.txt").await.unwrap()); } #[tokio::test] async fn test_path_sanitization() { let (storage, _temp_dir) = create_test_storage().await; // Directory traversal attempts should be sanitized storage .store("../../../etc/passwd", b"malicious") .await .unwrap(); // Should be stored safely within base directory assert!(storage.exists("etc/passwd").await.unwrap()); } // COMPRESSION EDGE CASE TESTS #[tokio::test] async fn test_compression_empty_data() { let temp_dir = TempDir::new().unwrap(); let config = LocalStorageConfig { base_path: temp_dir.path().to_path_buf(), enable_compression: true, ..Default::default() }; let storage = LocalStorage::new(config).await.unwrap(); // Empty data should compress/decompress without error storage.store("empty.bin", b"").await.unwrap(); let retrieved = storage.retrieve("empty.bin").await.unwrap(); assert_eq!(retrieved, b""); } #[tokio::test] async fn test_compression_highly_compressible_data() { let temp_dir = TempDir::new().unwrap(); let config = LocalStorageConfig { base_path: temp_dir.path().to_path_buf(), enable_compression: true, ..Default::default() }; let storage = LocalStorage::new(config).await.unwrap(); // Highly compressible data (repeated bytes) let data = vec![0u8; 10000]; storage.store("compressible.bin", &data).await.unwrap(); let retrieved = storage.retrieve("compressible.bin").await.unwrap(); assert_eq!(retrieved, data); } #[tokio::test] async fn test_compression_random_data() { let temp_dir = TempDir::new().unwrap(); let config = LocalStorageConfig { base_path: temp_dir.path().to_path_buf(), enable_compression: true, ..Default::default() }; let storage = LocalStorage::new(config).await.unwrap(); // Random data (not compressible) let data: Vec = (0..1000).map(|i| (i * 7 + 13) as u8).collect(); storage.store("random.bin", &data).await.unwrap(); let retrieved = storage.retrieve("random.bin").await.unwrap(); assert_eq!(retrieved, data); } #[tokio::test] async fn test_no_compression_mode() { let temp_dir = TempDir::new().unwrap(); let config = LocalStorageConfig { base_path: temp_dir.path().to_path_buf(), enable_compression: false, ..Default::default() }; let storage = LocalStorage::new(config).await.unwrap(); let data = b"test data without compression"; storage.store("uncompressed.bin", data).await.unwrap(); let retrieved = storage.retrieve("uncompressed.bin").await.unwrap(); assert_eq!(retrieved, data); } #[tokio::test] async fn test_compression_large_data() { let temp_dir = TempDir::new().unwrap(); let config = LocalStorageConfig { base_path: temp_dir.path().to_path_buf(), enable_compression: true, ..Default::default() }; let storage = LocalStorage::new(config).await.unwrap(); // Large data (1MB) let data = vec![42u8; 1024 * 1024]; storage.store("large.bin", &data).await.unwrap(); let retrieved = storage.retrieve("large.bin").await.unwrap(); assert_eq!(retrieved, data); } // ERROR HANDLING TESTS #[tokio::test] async fn test_retrieve_nonexistent_file() { let (storage, _temp_dir) = create_test_storage().await; let result = storage.retrieve("nonexistent.txt").await; assert!(result.is_err()); match result { Err(StorageError::NotFound { path }) => { assert_eq!(path, "nonexistent.txt"); }, _ => panic!("Expected NotFound error"), } } #[tokio::test] async fn test_metadata_nonexistent_file() { let (storage, _temp_dir) = create_test_storage().await; let result = storage.metadata("nonexistent.txt").await; assert!(result.is_err()); assert!(matches!(result, Err(StorageError::NotFound { .. }))); } #[tokio::test] async fn test_atomic_write_with_temp_file() { let temp_dir = TempDir::new().unwrap(); let config = LocalStorageConfig { base_path: temp_dir.path().to_path_buf(), atomic_writes: true, ..Default::default() }; let storage = LocalStorage::new(config).await.unwrap(); let data = b"atomic write test"; storage.store("atomic.txt", data).await.unwrap(); // Verify no temporary files are left behind let paths = storage.list("").await.unwrap(); for path in paths { assert!(!path.contains(".tmp"), "Temporary file found: {}", path); } } #[tokio::test] async fn test_non_atomic_write() { let temp_dir = TempDir::new().unwrap(); let config = LocalStorageConfig { base_path: temp_dir.path().to_path_buf(), atomic_writes: false, ..Default::default() }; let storage = LocalStorage::new(config).await.unwrap(); let data = b"non-atomic write test"; storage.store("nonatomic.txt", data).await.unwrap(); let retrieved = storage.retrieve("nonatomic.txt").await.unwrap(); assert_eq!(retrieved, data); } #[tokio::test] async fn test_overwrite_existing_file() { let (storage, _temp_dir) = create_test_storage().await; // Store initial data storage.store("overwrite.txt", b"original").await.unwrap(); // Overwrite with new data storage.store("overwrite.txt", b"updated").await.unwrap(); let retrieved = storage.retrieve("overwrite.txt").await.unwrap(); assert_eq!(retrieved, b"updated"); } #[tokio::test] async fn test_deep_nested_directories() { let (storage, _temp_dir) = create_test_storage().await; let deep_path = "a/b/c/d/e/f/g/h/i/j/file.txt"; storage.store(deep_path, b"deeply nested").await.unwrap(); assert!(storage.exists(deep_path).await.unwrap()); let retrieved = storage.retrieve(deep_path).await.unwrap(); assert_eq!(retrieved, b"deeply nested"); } #[tokio::test] async fn test_special_characters_in_path() { let (storage, _temp_dir) = create_test_storage().await; // Test with spaces and underscores let path = "test file_with-special.txt"; storage.store(path, b"special chars").await.unwrap(); assert!(storage.exists(path).await.unwrap()); } #[tokio::test] async fn test_list_with_prefix_matching() { let (storage, _temp_dir) = create_test_storage().await; storage.store("prefix_test1.txt", b"1").await.unwrap(); storage.store("prefix_test2.txt", b"2").await.unwrap(); storage.store("other.txt", b"3").await.unwrap(); let files = storage.list("prefix_").await.unwrap(); assert_eq!(files.len(), 2); assert!(files.iter().all(|f| f.starts_with("prefix_"))); } #[tokio::test] async fn test_list_empty_directory() { let (storage, _temp_dir) = create_test_storage().await; let files = storage.list("nonexistent_dir/").await.unwrap(); assert!(files.is_empty()); } #[tokio::test] async fn test_multiple_operations_sequence() { let (storage, _temp_dir) = create_test_storage().await; // Sequence of operations storage.store("seq1.txt", b"data1").await.unwrap(); assert!(storage.exists("seq1.txt").await.unwrap()); storage.store("seq2.txt", b"data2").await.unwrap(); assert_eq!(storage.list("seq").await.unwrap().len(), 2); storage.delete("seq1.txt").await.unwrap(); assert!(!storage.exists("seq1.txt").await.unwrap()); assert_eq!(storage.list("seq").await.unwrap().len(), 1); } #[tokio::test] async fn test_metadata_fields() { let (storage, _temp_dir) = create_test_storage().await; let data = b"metadata test data"; storage.store("meta.txt", data).await.unwrap(); let metadata = storage.metadata("meta.txt").await.unwrap(); assert_eq!(metadata.path, "meta.txt"); assert!(metadata.size > 0); assert_eq!( metadata.content_type, Some("application/octet-stream".to_owned()) ); assert!(metadata.last_modified <= Utc::now()); } #[tokio::test] async fn test_config_validation_absolute_path() { let config = LocalStorageConfig { base_path: std::path::PathBuf::from("relative/path"), ..Default::default() }; let result = config.validate(); assert!(result.is_err()); assert!(matches!(result, Err(StorageError::ConfigError { .. }))); } #[tokio::test] async fn test_config_validation_buffer_size() { let temp_dir = TempDir::new().unwrap(); let config = LocalStorageConfig { base_path: temp_dir.path().to_path_buf(), buffer_size: 0, ..Default::default() }; let result = config.validate(); assert!(result.is_err()); assert!(matches!(result, Err(StorageError::ConfigError { .. }))); } #[tokio::test] async fn test_binary_data_storage() { let (storage, _temp_dir) = create_test_storage().await; // Binary data with all byte values let data: Vec = (0..=255).collect(); storage.store("binary.bin", &data).await.unwrap(); let retrieved = storage.retrieve("binary.bin").await.unwrap(); assert_eq!(retrieved, data); } #[tokio::test] async fn test_concurrent_operations() { let (storage, _temp_dir) = create_test_storage().await; let storage = std::sync::Arc::new(storage); // Spawn multiple concurrent operations let mut handles = vec![]; for i in 0..10 { let storage_clone = storage.clone(); let handle = tokio::spawn(async move { let path = format!("concurrent_{}.txt", i); let data = format!("data_{}", i).into_bytes(); storage_clone.store(&path, &data).await.unwrap(); let retrieved = storage_clone.retrieve(&path).await.unwrap(); assert_eq!(retrieved, data); }); handles.push(handle); } // Wait for all operations to complete for handle in handles { handle.await.unwrap(); } // Verify all files were created let files = storage.list("concurrent_").await.unwrap(); assert_eq!(files.len(), 10); } }