/*! * Source Scheduler Unit Tests * * Tests for the universal source scheduler functionality including: * - Auto-resume sync after server restart * - Background sync scheduling * - Manual sync triggering * - Source type detection and routing * - Error handling and recovery */ use std::sync::Arc; use std::time::Duration; use tokio::time::timeout; use uuid::Uuid; use chrono::Utc; use serde_json::json; use readur::{ AppState, config::Config, db::Database, models::{Source, SourceType, SourceStatus, WebDAVSourceConfig, LocalFolderSourceConfig, S3SourceConfig}, source_scheduler::SourceScheduler, }; /// Mock database for testing struct MockDatabase { sources: Vec, sources_for_sync: Vec, } impl MockDatabase { fn new() -> Self { Self { sources: Vec::new(), sources_for_sync: Vec::new(), } } fn with_interrupted_source(mut self, name: &str, source_type: SourceType) -> Self { let mut config = json!({}); match source_type { SourceType::WebDAV => { config = json!({ "server_url": "https://test.com", "username": "test", "password": "test", "watch_folders": ["/test"], "file_extensions": [".pdf", ".txt"], "auto_sync": true, "sync_interval_minutes": 60 }); }, SourceType::LocalFolder => { config = json!({ "paths": ["/test/folder"], "recursive": true, "follow_symlinks": false, "auto_sync": true, "sync_interval_minutes": 30 }); }, SourceType::S3 => { config = json!({ "bucket": "test-bucket", "region": "us-east-1", "access_key_id": "test", "secret_access_key": "test", "prefix": "", "auto_sync": true, "sync_interval_minutes": 120 }); } } self.sources.push(Source { id: Uuid::new_v4(), user_id: Uuid::new_v4(), name: name.to_string(), source_type, enabled: true, config, status: SourceStatus::Syncing, // Interrupted state last_sync_at: None, last_error: None, last_error_at: None, total_files_synced: 0, total_files_pending: 0, total_size_bytes: 0, created_at: Utc::now(), updated_at: Utc::now(), }); self.sources_for_sync = self.sources.clone(); self } fn with_due_sync_source(mut self, name: &str, source_type: SourceType, minutes_ago: i64) -> Self { let mut config = json!({}); match source_type { SourceType::WebDAV => { config = json!({ "server_url": "https://test.com", "username": "test", "password": "test", "watch_folders": ["/test"], "file_extensions": [".pdf", ".txt"], "auto_sync": true, "sync_interval_minutes": 30 }); }, SourceType::LocalFolder => { config = json!({ "paths": ["/test/folder"], "recursive": true, "follow_symlinks": false, "auto_sync": true, "sync_interval_minutes": 30 }); }, SourceType::S3 => { config = json!({ "bucket": "test-bucket", "region": "us-east-1", "access_key_id": "test", "secret_access_key": "test", "prefix": "", "auto_sync": true, "sync_interval_minutes": 30 }); } } self.sources.push(Source { id: Uuid::new_v4(), user_id: Uuid::new_v4(), name: name.to_string(), source_type, enabled: true, config, status: SourceStatus::Idle, last_sync_at: Some(Utc::now() - chrono::Duration::minutes(minutes_ago)), last_error: None, last_error_at: None, total_files_synced: 5, total_files_pending: 0, total_size_bytes: 1024, created_at: Utc::now(), updated_at: Utc::now(), }); self.sources_for_sync = self.sources.clone(); self } } async fn create_test_app_state() -> Arc { let database_url = std::env::var("TEST_DATABASE_URL") .or_else(|_| std::env::var("DATABASE_URL")) .unwrap_or_else(|_| "postgresql://readur:readur@localhost:5432/readur".to_string()); let config = Config { database_url, server_address: "127.0.0.1:8080".to_string(), jwt_secret: "test_secret".to_string(), upload_path: "/tmp/test_uploads".to_string(), watch_folder: "/tmp/test_watch".to_string(), allowed_file_types: vec!["pdf".to_string(), "txt".to_string()], watch_interval_seconds: Some(30), file_stability_check_ms: Some(500), max_file_age_hours: None, ocr_language: "eng".to_string(), concurrent_ocr_jobs: 2, ocr_timeout_seconds: 60, max_file_size_mb: 10, memory_limit_mb: 256, cpu_priority: "normal".to_string(), }; let db = Database::new(&config.database_url).await.unwrap(); let queue_service = std::sync::Arc::new(readur::ocr_queue::OcrQueueService::new(db.clone(), db.pool.clone(), 2)); Arc::new(AppState { db, config, webdav_scheduler: None, source_scheduler: None, queue_service, }) } #[tokio::test] async fn test_source_scheduler_creation() { let state = create_test_app_state().await; let scheduler = SourceScheduler::new(state.clone()); // Test that scheduler is created successfully // assert_eq!(scheduler.check_interval, Duration::from_secs(60)); // private field } #[tokio::test] async fn test_interrupted_sync_detection_webdav() { let state = create_test_app_state().await; let scheduler = SourceScheduler::new(state.clone()); // Create a mock source that was interrupted during sync let source = Source { id: Uuid::new_v4(), user_id: Uuid::new_v4(), name: "Test WebDAV".to_string(), source_type: SourceType::WebDAV, enabled: true, config: json!({ "server_url": "https://test.com", "username": "test", "password": "test", "watch_folders": ["/test"], "file_extensions": [".pdf", ".txt"], "auto_sync": true, "sync_interval_minutes": 60 }), status: SourceStatus::Syncing, // This indicates interruption last_sync_at: None, last_error: None, last_error_at: None, total_files_synced: 0, total_files_pending: 0, total_size_bytes: 0, created_at: Utc::now(), updated_at: Utc::now(), }; // Test that interrupted sync is detected assert_eq!(source.status, SourceStatus::Syncing); // Test config parsing for WebDAV let config: Result = serde_json::from_value(source.config.clone()); assert!(config.is_ok()); let webdav_config = config.unwrap(); assert!(webdav_config.auto_sync); assert_eq!(webdav_config.sync_interval_minutes, 60); } #[tokio::test] async fn test_interrupted_sync_detection_local_folder() { let state = create_test_app_state().await; let scheduler = SourceScheduler::new(state.clone()); let source = Source { id: Uuid::new_v4(), user_id: Uuid::new_v4(), name: "Test Local Folder".to_string(), source_type: SourceType::LocalFolder, enabled: true, config: json!({ "watch_folders": ["/test/folder"], "file_extensions": [".pdf", ".txt"], "recursive": true, "follow_symlinks": false, "auto_sync": true, "sync_interval_minutes": 30 }), status: SourceStatus::Syncing, last_sync_at: None, last_error: None, last_error_at: None, total_files_synced: 0, total_files_pending: 0, total_size_bytes: 0, created_at: Utc::now(), updated_at: Utc::now(), }; // Test config parsing for Local Folder let config: Result = serde_json::from_value(source.config.clone()); assert!(config.is_ok()); let local_config = config.unwrap(); assert!(local_config.auto_sync); assert_eq!(local_config.sync_interval_minutes, 30); } #[tokio::test] async fn test_interrupted_sync_detection_s3() { let state = create_test_app_state().await; let scheduler = SourceScheduler::new(state.clone()); let source = Source { id: Uuid::new_v4(), user_id: Uuid::new_v4(), name: "Test S3".to_string(), source_type: SourceType::S3, enabled: true, config: json!({ "bucket_name": "test-bucket", "region": "us-east-1", "access_key_id": "test", "secret_access_key": "test", "prefix": "", "watch_folders": ["/test/prefix"], "file_extensions": [".pdf", ".txt"], "auto_sync": true, "sync_interval_minutes": 120 }), status: SourceStatus::Syncing, last_sync_at: None, last_error: None, last_error_at: None, total_files_synced: 0, total_files_pending: 0, total_size_bytes: 0, created_at: Utc::now(), updated_at: Utc::now(), }; // Test config parsing for S3 let config: Result = serde_json::from_value(source.config.clone()); assert!(config.is_ok()); let s3_config = config.unwrap(); assert!(s3_config.auto_sync); assert_eq!(s3_config.sync_interval_minutes, 120); } #[tokio::test] async fn test_sync_due_calculation() { let state = create_test_app_state().await; let scheduler = SourceScheduler::new(state.clone()); // Test source that should be due for sync let old_sync_source = Source { id: Uuid::new_v4(), user_id: Uuid::new_v4(), name: "Old Sync".to_string(), source_type: SourceType::WebDAV, enabled: true, config: json!({ "server_url": "https://test.com", "username": "test", "password": "test", "watch_folders": ["/test"], "file_extensions": [".pdf"], "auto_sync": true, "sync_interval_minutes": 30 }), status: SourceStatus::Idle, last_sync_at: Some(Utc::now() - chrono::Duration::minutes(45)), // 45 minutes ago last_error: None, last_error_at: None, total_files_synced: 5, total_files_pending: 0, total_size_bytes: 1024, created_at: Utc::now(), updated_at: Utc::now(), }; // Test source that should NOT be due for sync let recent_sync_source = Source { id: Uuid::new_v4(), user_id: Uuid::new_v4(), name: "Recent Sync".to_string(), source_type: SourceType::LocalFolder, enabled: true, config: json!({ "paths": ["/test"], "recursive": true, "follow_symlinks": false, "auto_sync": true, "sync_interval_minutes": 60 }), status: SourceStatus::Idle, last_sync_at: Some(Utc::now() - chrono::Duration::minutes(15)), // 15 minutes ago last_error: None, last_error_at: None, total_files_synced: 10, total_files_pending: 0, total_size_bytes: 2048, created_at: Utc::now(), updated_at: Utc::now(), }; // Test that sync due calculation works correctly // let old_result = scheduler.is_sync_due(&old_sync_source).await; // assert!(old_result.is_ok()); // assert!(old_result.unwrap(), "Old sync should be due"); // let recent_result = scheduler.is_sync_due(&recent_sync_source).await; // assert!(recent_result.is_ok()); // assert!(!recent_result.unwrap(), "Recent sync should not be due"); } #[tokio::test] async fn test_auto_sync_disabled() { let state = create_test_app_state().await; let scheduler = SourceScheduler::new(state.clone()); let source_with_auto_sync_disabled = Source { id: Uuid::new_v4(), user_id: Uuid::new_v4(), name: "Auto Sync Disabled".to_string(), source_type: SourceType::WebDAV, enabled: true, config: json!({ "server_url": "https://test.com", "username": "test", "password": "test", "watch_folders": ["/test"], "file_extensions": [".pdf"], "auto_sync": false, // Disabled "sync_interval_minutes": 30 }), status: SourceStatus::Idle, last_sync_at: Some(Utc::now() - chrono::Duration::minutes(45)), last_error: None, last_error_at: None, total_files_synced: 0, total_files_pending: 0, total_size_bytes: 0, created_at: Utc::now(), updated_at: Utc::now(), }; // let result = scheduler.is_sync_due(&source_with_auto_sync_disabled).await; // assert!(result.is_ok()); // assert!(!result.unwrap(), "Source with auto_sync disabled should not be due"); } #[tokio::test] async fn test_currently_syncing_source() { let state = create_test_app_state().await; let scheduler = SourceScheduler::new(state.clone()); let syncing_source = Source { id: Uuid::new_v4(), user_id: Uuid::new_v4(), name: "Currently Syncing".to_string(), source_type: SourceType::S3, enabled: true, config: json!({ "bucket": "test-bucket", "region": "us-east-1", "access_key_id": "test", "secret_access_key": "test", "prefix": "", "auto_sync": true, "sync_interval_minutes": 30 }), status: SourceStatus::Syncing, // Currently syncing last_sync_at: Some(Utc::now() - chrono::Duration::minutes(45)), last_error: None, last_error_at: None, total_files_synced: 0, total_files_pending: 5, total_size_bytes: 0, created_at: Utc::now(), updated_at: Utc::now(), }; // let result = scheduler.is_sync_due(&syncing_source).await; // assert!(result.is_ok()); // assert!(!result.unwrap(), "Currently syncing source should not be due for another sync"); } #[tokio::test] async fn test_invalid_sync_interval() { let state = create_test_app_state().await; let scheduler = SourceScheduler::new(state.clone()); let invalid_interval_source = Source { id: Uuid::new_v4(), user_id: Uuid::new_v4(), name: "Invalid Interval".to_string(), source_type: SourceType::WebDAV, enabled: true, config: json!({ "server_url": "https://test.com", "username": "test", "password": "test", "watch_folders": ["/test"], "file_extensions": [".pdf"], "auto_sync": true, "sync_interval_minutes": 0 // Invalid interval }), status: SourceStatus::Idle, last_sync_at: Some(Utc::now() - chrono::Duration::minutes(45)), last_error: None, last_error_at: None, total_files_synced: 0, total_files_pending: 0, total_size_bytes: 0, created_at: Utc::now(), updated_at: Utc::now(), }; // let result = scheduler.is_sync_due(&invalid_interval_source).await; // assert!(result.is_ok()); // assert!(!result.unwrap(), "Source with invalid sync interval should not be due"); } #[tokio::test] async fn test_never_synced_source() { let state = create_test_app_state().await; let scheduler = SourceScheduler::new(state.clone()); let never_synced_source = Source { id: Uuid::new_v4(), user_id: Uuid::new_v4(), name: "Never Synced".to_string(), source_type: SourceType::LocalFolder, enabled: true, config: json!({ "paths": ["/test"], "recursive": true, "follow_symlinks": false, "auto_sync": true, "sync_interval_minutes": 60 }), status: SourceStatus::Idle, last_sync_at: None, // Never synced last_error: None, last_error_at: None, total_files_synced: 0, total_files_pending: 0, total_size_bytes: 0, created_at: Utc::now(), updated_at: Utc::now(), }; // let result = scheduler.is_sync_due(&never_synced_source).await; // assert!(result.is_ok()); // assert!(result.unwrap(), "Never synced source should be due for sync"); } #[tokio::test] async fn test_trigger_sync_nonexistent_source() { let state = create_test_app_state().await; let scheduler = SourceScheduler::new(state.clone()); let nonexistent_id = Uuid::new_v4(); let result = scheduler.trigger_sync(nonexistent_id).await; assert!(result.is_err()); assert_eq!(result.unwrap_err().to_string(), "Source not found"); } #[tokio::test] async fn test_source_status_enum() { // Test SourceStatus enum string conversion assert_eq!(SourceStatus::Idle.to_string(), "idle"); assert_eq!(SourceStatus::Syncing.to_string(), "syncing"); assert_eq!(SourceStatus::Error.to_string(), "error"); } #[tokio::test] async fn test_source_type_enum() { // Test SourceType enum let webdav = SourceType::WebDAV; let local = SourceType::LocalFolder; let s3 = SourceType::S3; assert_eq!(webdav.to_string(), "webdav"); assert_eq!(local.to_string(), "local_folder"); assert_eq!(s3.to_string(), "s3"); } #[tokio::test] async fn test_config_validation() { // Test WebDAV config validation let webdav_config = WebDAVSourceConfig { server_url: "https://test.com".to_string(), username: "user".to_string(), password: "pass".to_string(), watch_folders: vec!["/folder1".to_string(), "/folder2".to_string()], file_extensions: vec![".pdf".to_string(), ".txt".to_string()], auto_sync: true, sync_interval_minutes: 60, server_type: Some("nextcloud".to_string()), }; assert!(!webdav_config.server_url.is_empty()); assert!(!webdav_config.username.is_empty()); assert!(!webdav_config.password.is_empty()); assert!(!webdav_config.watch_folders.is_empty()); assert!(webdav_config.sync_interval_minutes > 0); // Test Local Folder config validation let local_config = LocalFolderSourceConfig { watch_folders: vec!["/test/path".to_string()], recursive: true, follow_symlinks: false, auto_sync: true, sync_interval_minutes: 30, file_extensions: vec![".pdf".to_string()], }; assert!(!local_config.watch_folders.is_empty()); assert!(local_config.sync_interval_minutes > 0); // Test S3 config validation let s3_config = S3SourceConfig { bucket_name: "test-bucket".to_string(), region: "us-east-1".to_string(), access_key_id: "key".to_string(), secret_access_key: "secret".to_string(), prefix: Some("docs/".to_string()), endpoint_url: Some("https://minio.example.com".to_string()), watch_folders: vec!["docs/".to_string()], auto_sync: true, sync_interval_minutes: 120, file_extensions: vec![".pdf".to_string()], }; assert!(!s3_config.bucket_name.is_empty()); assert!(!s3_config.region.is_empty()); assert!(!s3_config.access_key_id.is_empty()); assert!(!s3_config.secret_access_key.is_empty()); assert!(s3_config.sync_interval_minutes > 0); } #[tokio::test] async fn test_scheduler_timeout_handling() { let state = create_test_app_state().await; let scheduler = SourceScheduler::new(state.clone()); // Test that operations complete within reasonable time let start = std::time::Instant::now(); let dummy_source = Source { id: Uuid::new_v4(), user_id: Uuid::new_v4(), name: "Timeout Test".to_string(), source_type: SourceType::WebDAV, enabled: true, config: json!({ "server_url": "https://test.com", "username": "test", "password": "test", "watch_folders": ["/test"], "file_extensions": [".pdf"], "auto_sync": true, "sync_interval_minutes": 60 }), status: SourceStatus::Idle, last_sync_at: None, last_error: None, last_error_at: None, total_files_synced: 0, total_files_pending: 0, total_size_bytes: 0, created_at: Utc::now(), updated_at: Utc::now(), }; // let result = timeout(Duration::from_secs(1), scheduler.is_sync_due(&dummy_source)).await; // assert!(result.is_ok(), "Sync due calculation should complete quickly"); let elapsed = start.elapsed(); assert!(elapsed < Duration::from_millis(500), "Operation should be fast"); }