Spaces:
Sleeping
Sleeping
| """Tests for monitoring loop and scheduler components.""" | |
| import pytest | |
| import time | |
| from unittest.mock import Mock, patch, MagicMock | |
| from datetime import datetime, timedelta | |
| from src.scheduler.monitoring_loop import MonitoringLoop, TaskCoordinator, StatusReporter | |
| class TestMonitoringLoop: | |
| """Test cases for MonitoringLoop class.""" | |
| def setup_method(self): | |
| """Set up test fixtures.""" | |
| self.mock_url_generator = Mock() | |
| self.mock_download_manager = Mock() | |
| self.mock_storage = Mock() | |
| # Set up download manager mocks | |
| self.mock_download_manager.get_download_count.return_value = 0 | |
| self.mock_download_manager.get_failed_tasks.return_value = [] | |
| self.monitoring_loop = MonitoringLoop( | |
| self.mock_url_generator, | |
| self.mock_download_manager, | |
| self.mock_storage, | |
| check_interval_minutes=1, # Short interval for testing | |
| monitoring_range_days=1 # Default range for testing | |
| ) | |
| def test_initialization(self): | |
| """Test monitoring loop initialization.""" | |
| assert self.monitoring_loop.check_interval == 1 | |
| assert self.monitoring_loop.monitoring_range_days == 1 | |
| assert self.monitoring_loop.is_running is False | |
| assert self.monitoring_loop.total_checks == 0 | |
| assert self.monitoring_loop.new_images_found == 0 | |
| def test_get_status_initial(self): | |
| """Test getting initial status.""" | |
| status = self.monitoring_loop.get_status() | |
| assert status['is_running'] is False | |
| assert status['check_interval_minutes'] == 1 | |
| assert status['monitoring_range_days'] == 1 | |
| assert status['total_checks'] == 0 | |
| assert status['last_check_time'] is None | |
| assert status['new_images_found'] == 0 | |
| def test_filter_new_images_all_new(self): | |
| """Test filtering when all images are new.""" | |
| urls = [ | |
| "https://sdo.gsfc.nasa.gov/assets/img/browse/2025/12/19/20251219_120000_4096_0211.jpg", | |
| "https://sdo.gsfc.nasa.gov/assets/img/browse/2025/12/19/20251219_130000_4096_0211.jpg" | |
| ] | |
| # Mock URL generator to extract metadata | |
| self.mock_url_generator.extract_metadata_from_url.side_effect = [ | |
| (datetime(2025, 12, 19), "120000"), | |
| (datetime(2025, 12, 19), "130000") | |
| ] | |
| # Mock storage to return False (files don't exist) | |
| self.mock_storage.file_exists.return_value = False | |
| new_urls = self.monitoring_loop._filter_new_images(urls) | |
| assert len(new_urls) == 2 | |
| assert new_urls == urls | |
| def test_filter_new_images_some_exist(self): | |
| """Test filtering when some images already exist.""" | |
| urls = [ | |
| "https://sdo.gsfc.nasa.gov/assets/img/browse/2025/12/19/20251219_120000_4096_0211.jpg", | |
| "https://sdo.gsfc.nasa.gov/assets/img/browse/2025/12/19/20251219_130000_4096_0211.jpg" | |
| ] | |
| # Mock URL generator | |
| self.mock_url_generator.extract_metadata_from_url.side_effect = [ | |
| (datetime(2025, 12, 19), "120000"), | |
| (datetime(2025, 12, 19), "130000") | |
| ] | |
| # Mock storage - first file exists, second doesn't | |
| self.mock_storage.file_exists.side_effect = [True, False] | |
| new_urls = self.monitoring_loop._filter_new_images(urls) | |
| assert len(new_urls) == 1 | |
| assert new_urls[0] == urls[1] # Only second URL should be returned | |
| def test_filter_new_images_invalid_url(self): | |
| """Test filtering with invalid URLs.""" | |
| urls = [ | |
| "https://sdo.gsfc.nasa.gov/assets/img/browse/2025/12/19/20251219_120000_4096_0211.jpg", | |
| "https://invalid-url.com/image.jpg" | |
| ] | |
| # Mock URL generator - first URL valid, second invalid | |
| self.mock_url_generator.extract_metadata_from_url.side_effect = [ | |
| (datetime(2025, 12, 19), "120000"), | |
| (None, None) # Invalid URL | |
| ] | |
| self.mock_storage.file_exists.return_value = False | |
| new_urls = self.monitoring_loop._filter_new_images(urls) | |
| assert len(new_urls) == 1 | |
| assert new_urls[0] == urls[0] # Only valid URL should be returned | |
| def test_download_new_images_success(self, mock_download_task): | |
| """Test downloading new images successfully.""" | |
| urls = ["https://sdo.gsfc.nasa.gov/assets/img/browse/2025/12/19/20251219_120000_4096_0211.jpg"] | |
| # Mock URL generator | |
| self.mock_url_generator.extract_metadata_from_url.return_value = ( | |
| datetime(2025, 12, 19), "120000" | |
| ) | |
| # Mock storage | |
| from pathlib import Path | |
| self.mock_storage.get_local_path.return_value = Path("data/2025/12/19/20251219_120000_4096_0211.jpg") | |
| # Mock download manager | |
| self.mock_download_manager.download_and_save.return_value = True | |
| # Mock download task | |
| mock_task = Mock() | |
| mock_download_task.return_value = mock_task | |
| self.monitoring_loop._download_new_images(urls) | |
| # Verify download was attempted | |
| self.mock_download_manager.download_and_save.assert_called_once_with(mock_task) | |
| def test_download_new_images_failure(self, mock_download_task): | |
| """Test downloading new images with failure.""" | |
| urls = ["https://sdo.gsfc.nasa.gov/assets/img/browse/2025/12/19/20251219_120000_4096_0211.jpg"] | |
| # Mock URL generator | |
| self.mock_url_generator.extract_metadata_from_url.return_value = ( | |
| datetime(2025, 12, 19), "120000" | |
| ) | |
| # Mock storage | |
| from pathlib import Path | |
| self.mock_storage.get_local_path.return_value = Path("data/2025/12/19/20251219_120000_4096_0211.jpg") | |
| # Mock download manager to fail | |
| self.mock_download_manager.download_and_save.return_value = False | |
| # Mock download task with error | |
| mock_task = Mock() | |
| mock_task.error_message = "Network error" | |
| mock_download_task.return_value = mock_task | |
| self.monitoring_loop._download_new_images(urls) | |
| # Verify download was attempted | |
| self.mock_download_manager.download_and_save.assert_called_once_with(mock_task) | |
| def test_force_check_not_running(self): | |
| """Test forcing check when not running.""" | |
| # Should not crash, just log warning | |
| self.monitoring_loop.force_check() | |
| # No actual check should have been performed | |
| assert self.monitoring_loop.total_checks == 0 | |
| def test_start_monitoring(self, mock_thread, mock_schedule): | |
| """Test starting monitoring loop.""" | |
| mock_thread_instance = Mock() | |
| mock_thread.return_value = mock_thread_instance | |
| self.monitoring_loop.start_monitoring() | |
| assert self.monitoring_loop.is_running is True | |
| mock_schedule.every.assert_called_once() | |
| mock_thread_instance.start.assert_called_once() | |
| def test_stop_monitoring_not_running(self): | |
| """Test stopping monitoring when not running.""" | |
| # Should not crash, just log warning | |
| self.monitoring_loop.stop_monitoring() | |
| assert self.monitoring_loop.is_running is False | |
| def test_set_monitoring_range(self): | |
| """Test setting monitoring range.""" | |
| # Test valid range | |
| self.monitoring_loop.set_monitoring_range(7) | |
| assert self.monitoring_loop.get_monitoring_range() == 7 | |
| # Test another valid range | |
| self.monitoring_loop.set_monitoring_range(30) | |
| assert self.monitoring_loop.get_monitoring_range() == 30 | |
| def test_set_monitoring_range_invalid(self): | |
| """Test setting invalid monitoring range.""" | |
| with pytest.raises(ValueError, match="Monitoring range must be at least 1 day"): | |
| self.monitoring_loop.set_monitoring_range(0) | |
| with pytest.raises(ValueError, match="Monitoring range must be at least 1 day"): | |
| self.monitoring_loop.set_monitoring_range(-1) | |
| def test_get_monitoring_range(self): | |
| """Test getting monitoring range.""" | |
| # Should return default value | |
| assert self.monitoring_loop.get_monitoring_range() == 1 | |
| def test_set_monitoring_range(self): | |
| """Test setting monitoring range.""" | |
| self.monitoring_loop.set_monitoring_range(7) | |
| assert self.monitoring_loop.get_monitoring_range() == 7 | |
| # Test invalid range | |
| with pytest.raises(ValueError): | |
| self.monitoring_loop.set_monitoring_range(0) | |
| def test_get_monitoring_range(self): | |
| """Test getting monitoring range.""" | |
| assert self.monitoring_loop.get_monitoring_range() == 1 | |
| class TestTaskCoordinator: | |
| """Test cases for TaskCoordinator class.""" | |
| def setup_method(self): | |
| """Set up test fixtures.""" | |
| self.mock_monitoring_loop = Mock() | |
| self.coordinator = TaskCoordinator(self.mock_monitoring_loop) | |
| def test_initialization(self): | |
| """Test task coordinator initialization.""" | |
| # Verify callbacks were set | |
| assert self.mock_monitoring_loop.on_check_start is not None | |
| assert self.mock_monitoring_loop.on_check_complete is not None | |
| assert self.mock_monitoring_loop.on_new_images_found is not None | |
| def test_on_check_start(self): | |
| """Test check start callback.""" | |
| check_time = datetime.now() | |
| check_number = 5 | |
| # Should not raise exception | |
| self.coordinator._on_check_start(check_time, check_number) | |
| def test_on_check_complete(self): | |
| """Test check complete callback.""" | |
| check_time = datetime.now() | |
| new_images = 3 | |
| duration = 2.5 | |
| # Should not raise exception | |
| self.coordinator._on_check_complete(check_time, new_images, duration) | |
| def test_on_new_images_found(self): | |
| """Test new images found callback.""" | |
| urls = [ | |
| "https://sdo.gsfc.nasa.gov/assets/img/browse/2025/12/19/20251219_120000_4096_0211.jpg", | |
| "https://sdo.gsfc.nasa.gov/assets/img/browse/2025/12/19/20251219_130000_4096_0211.jpg" | |
| ] | |
| # Should not raise exception | |
| self.coordinator._on_new_images_found(urls) | |
| class TestStatusReporter: | |
| """Test cases for StatusReporter class.""" | |
| def setup_method(self): | |
| """Set up test fixtures.""" | |
| self.mock_monitoring_loop = Mock() | |
| self.reporter = StatusReporter(self.mock_monitoring_loop) | |
| def test_initialization(self): | |
| """Test status reporter initialization.""" | |
| assert self.reporter.monitoring_loop == self.mock_monitoring_loop | |
| def test_print_status(self): | |
| """Test printing status.""" | |
| # Mock status | |
| self.mock_monitoring_loop.get_status.return_value = { | |
| 'is_running': True, | |
| 'check_interval_minutes': 5, | |
| 'monitoring_range_days': 1, | |
| 'total_checks': 10, | |
| 'last_check_time': datetime(2025, 12, 19, 12, 0, 0), | |
| 'new_images_found': 5, | |
| 'total_downloads': 3, | |
| 'failed_downloads': 2 | |
| } | |
| # Should not raise exception | |
| self.reporter.print_status() | |
| def test_print_status_no_last_check(self): | |
| """Test printing status when no checks have been performed.""" | |
| # Mock status with no last check time | |
| self.mock_monitoring_loop.get_status.return_value = { | |
| 'is_running': False, | |
| 'check_interval_minutes': 5, | |
| 'monitoring_range_days': 1, | |
| 'total_checks': 0, | |
| 'last_check_time': None, | |
| 'new_images_found': 0, | |
| 'total_downloads': 0, | |
| 'failed_downloads': 0 | |
| } | |
| # Should not raise exception | |
| self.reporter.print_status() | |
| def test_log_periodic_status(self, mock_schedule): | |
| """Test setting up periodic status logging.""" | |
| self.reporter.log_periodic_status(30) | |
| # Verify schedule was called | |
| mock_schedule.every.assert_called_once_with(30) |