Spaces:
Sleeping
Sleeping
File size: 12,967 Bytes
b610d23 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 |
"""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
@patch('src.scheduler.monitoring_loop.DownloadTask')
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)
@patch('src.scheduler.monitoring_loop.DownloadTask')
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
@patch('src.scheduler.monitoring_loop.schedule')
@patch('src.scheduler.monitoring_loop.threading.Thread')
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()
@patch('src.scheduler.monitoring_loop.schedule')
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) |