Mirror/intercept
1
0
Fork 0
mirror of https://github.com/smittix/intercept.git synced 2026-04-24 06:40:00 -07:00
Code Issues Packages Projects Releases Wiki Activity

Add comprehensive test coverage for weather satellite modules

Browse Source 
- Created test_weather_sat_routes.py with 42 tests for all endpoints
- Created test_weather_sat_decoder.py with 47 tests for WeatherSatDecoder class
- Created test_weather_sat_predict.py with 14 tests for pass prediction
- Created test_weather_sat_scheduler.py with 31 tests for auto-scheduler
- Total: 134 test functions across 14 test classes
- All tests follow existing patterns (mocking, fixtures, docstrings)
- Tests cover happy paths, error handling, and edge cases
- Mock all external subprocess calls and HTTP requests

Co-authored-by: mitchross <6330506+mitchross@users.noreply.github.com>
This commit is contained in:
copilot-swe-agent[bot]
2026-02-09 21:50:22 +00:00
parent f217230ef4
commit 4a6dddbb48
4 changed files with 2898 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

643
tests/test_weather_sat_decoder.py Normal file
View File

@@ -0,0 +1,643 @@
"""Tests for WeatherSatDecoder class.
Covers WeatherSatDecoder methods, subprocess management, progress callbacks,
and image handling.
"""
from __future__ import annotations
import os
import tempfile
import threading
import time
from datetime import datetime, timezone
from pathlib import Path
from unittest.mock import patch, MagicMock, call, mock_open
import pytest
from utils.weather_sat import (
WeatherSatDecoder,
WeatherSatImage,
CaptureProgress,
WEATHER_SATELLITES,
get_weather_sat_decoder,
is_weather_sat_available,
)
class TestWeatherSatDecoder:
"""Tests for WeatherSatDecoder class."""
def test_decoder_initialization(self):
"""Decoder should initialize with default output directory."""
with patch('shutil.which', return_value='/usr/bin/satdump'):
decoder = WeatherSatDecoder()
assert decoder.is_running is False
assert decoder.decoder_available == 'satdump'
assert decoder.current_satellite == ''
assert decoder.current_frequency == 0.0
def test_decoder_initialization_no_satdump(self):
"""Decoder should detect when SatDump is unavailable."""
with patch('shutil.which', return_value=None):
decoder = WeatherSatDecoder()
assert decoder.decoder_available is None
def test_decoder_custom_output_dir(self):
"""Decoder should accept custom output directory."""
with patch('shutil.which', return_value='/usr/bin/satdump'):
custom_dir = '/tmp/custom_output'
decoder = WeatherSatDecoder(output_dir=custom_dir)
assert decoder._output_dir == Path(custom_dir)
def test_set_callback(self):
"""Decoder should accept progress callback."""
with patch('shutil.which', return_value='/usr/bin/satdump'):
decoder = WeatherSatDecoder()
callback = MagicMock()
decoder.set_callback(callback)
assert decoder._callback == callback
def test_set_on_complete(self):
"""Decoder should accept on_complete callback."""
with patch('shutil.which', return_value='/usr/bin/satdump'):
decoder = WeatherSatDecoder()
callback = MagicMock()
decoder.set_on_complete(callback)
assert decoder._on_complete_callback == callback
def test_start_no_decoder(self):
"""start() should fail when no decoder available."""
with patch('shutil.which', return_value=None):
decoder = WeatherSatDecoder()
callback = MagicMock()
decoder.set_callback(callback)
success = decoder.start(satellite='NOAA-18', device_index=0, gain=40.0)
assert success is False
callback.assert_called()
progress = callback.call_args[0][0]
assert progress.status == 'error'
assert 'SatDump' in progress.message
def test_start_invalid_satellite(self):
"""start() should fail with invalid satellite."""
with patch('shutil.which', return_value='/usr/bin/satdump'):
decoder = WeatherSatDecoder()
callback = MagicMock()
decoder.set_callback(callback)
success = decoder.start(satellite='FAKE-SAT', device_index=0, gain=40.0)
assert success is False
callback.assert_called()
progress = callback.call_args[0][0]
assert progress.status == 'error'
assert 'Unknown satellite' in progress.message
@patch('subprocess.Popen')
@patch('pty.openpty')
@patch('utils.weather_sat.register_process')
def test_start_success(self, mock_register, mock_pty, mock_popen):
"""start() should successfully start SatDump."""
with patch('shutil.which', return_value='/usr/bin/satdump'), \
patch('utils.weather_sat.WeatherSatDecoder._resolve_device_id', return_value='0'):
mock_pty.return_value = (10, 11)
mock_process = MagicMock()
mock_process.poll.return_value = None
mock_popen.return_value = mock_process
decoder = WeatherSatDecoder()
callback = MagicMock()
decoder.set_callback(callback)
success = decoder.start(
satellite='NOAA-18',
device_index=0,
gain=40.0,
bias_t=True,
)
assert success is True
assert decoder.is_running is True
assert decoder.current_satellite == 'NOAA-18'
assert decoder.current_frequency == 137.9125
assert decoder.current_mode == 'APT'
assert decoder.device_index == 0
mock_popen.assert_called_once()
cmd = mock_popen.call_args[0][0]
assert cmd[0] == 'satdump'
assert 'live' in cmd
assert 'noaa_apt' in cmd
assert '--bias' in cmd
@patch('subprocess.Popen')
@patch('pty.openpty')
def test_start_already_running(self, mock_pty, mock_popen):
"""start() should return True when already running."""
with patch('shutil.which', return_value='/usr/bin/satdump'):
decoder = WeatherSatDecoder()
decoder._running = True
success = decoder.start(satellite='NOAA-18', device_index=0, gain=40.0)
assert success is True
mock_popen.assert_not_called()
@patch('subprocess.Popen')
@patch('pty.openpty')
def test_start_exception_handling(self, mock_pty, mock_popen):
"""start() should handle exceptions gracefully."""
with patch('shutil.which', return_value='/usr/bin/satdump'):
mock_pty.return_value = (10, 11)
mock_popen.side_effect = OSError('Device not found')
decoder = WeatherSatDecoder()
callback = MagicMock()
decoder.set_callback(callback)
success = decoder.start(satellite='NOAA-18', device_index=0, gain=40.0)
assert success is False
assert decoder.is_running is False
callback.assert_called()
progress = callback.call_args[0][0]
assert progress.status == 'error'
def test_start_from_file_no_decoder(self):
"""start_from_file() should fail when no decoder available."""
with patch('shutil.which', return_value=None):
decoder = WeatherSatDecoder()
callback = MagicMock()
decoder.set_callback(callback)
success = decoder.start_from_file(
satellite='NOAA-18',
input_file='data/test.wav',
)
assert success is False
callback.assert_called()
@patch('subprocess.Popen')
@patch('pty.openpty')
@patch('pathlib.Path.is_file', return_value=True)
@patch('pathlib.Path.resolve')
def test_start_from_file_success(self, mock_resolve, mock_is_file, mock_pty, mock_popen):
"""start_from_file() should successfully decode from file."""
with patch('shutil.which', return_value='/usr/bin/satdump'), \
patch('utils.weather_sat.register_process'):
# Mock path resolution
mock_path = MagicMock()
mock_path.is_relative_to.return_value = True
mock_path.suffix = '.wav'
mock_resolve.return_value = mock_path
mock_pty.return_value = (10, 11)
mock_process = MagicMock()
mock_popen.return_value = mock_process
decoder = WeatherSatDecoder()
callback = MagicMock()
decoder.set_callback(callback)
success = decoder.start_from_file(
satellite='NOAA-18',
input_file='data/test.wav',
sample_rate=1000000,
)
assert success is True
assert decoder.is_running is True
assert decoder.current_satellite == 'NOAA-18'
mock_popen.assert_called_once()
cmd = mock_popen.call_args[0][0]
assert cmd[0] == 'satdump'
assert 'noaa_apt' in cmd
assert 'audio_wav' in cmd
assert '--samplerate' in cmd
@patch('pathlib.Path.resolve')
def test_start_from_file_path_traversal(self, mock_resolve):
"""start_from_file() should block path traversal."""
with patch('shutil.which', return_value='/usr/bin/satdump'):
# Mock path outside allowed directory
mock_path = MagicMock()
mock_path.is_relative_to.return_value = False
mock_resolve.return_value = mock_path
decoder = WeatherSatDecoder()
callback = MagicMock()
decoder.set_callback(callback)
success = decoder.start_from_file(
satellite='NOAA-18',
input_file='/etc/passwd',
)
assert success is False
callback.assert_called()
progress = callback.call_args[0][0]
assert 'data/ directory' in progress.message
@patch('pathlib.Path.is_file', return_value=False)
@patch('pathlib.Path.resolve')
def test_start_from_file_not_found(self, mock_resolve, mock_is_file):
"""start_from_file() should fail when file not found."""
with patch('shutil.which', return_value='/usr/bin/satdump'):
mock_path = MagicMock()
mock_path.is_relative_to.return_value = True
mock_resolve.return_value = mock_path
decoder = WeatherSatDecoder()
callback = MagicMock()
decoder.set_callback(callback)
success = decoder.start_from_file(
satellite='NOAA-18',
input_file='data/missing.wav',
)
assert success is False
callback.assert_called()
progress = callback.call_args[0][0]
assert 'not found' in progress.message.lower()
def test_stop_not_running(self):
"""stop() should be safe when not running."""
with patch('shutil.which', return_value='/usr/bin/satdump'):
decoder = WeatherSatDecoder()
decoder.stop() # Should not raise
@patch('utils.weather_sat.safe_terminate')
def test_stop_running(self, mock_terminate):
"""stop() should terminate process."""
with patch('shutil.which', return_value='/usr/bin/satdump'):
decoder = WeatherSatDecoder()
mock_process = MagicMock()
decoder._process = mock_process
decoder._running = True
decoder._pty_master_fd = 10
with patch('os.close') as mock_close:
decoder.stop()
assert decoder._running is False
mock_terminate.assert_called_once_with(mock_process)
mock_close.assert_called_once_with(10)
def test_get_images_empty(self):
"""get_images() should return empty list initially."""
with patch('shutil.which', return_value='/usr/bin/satdump'):
decoder = WeatherSatDecoder()
images = decoder.get_images()
assert images == []
@patch('pathlib.Path.glob')
@patch('pathlib.Path.stat')
def test_get_images_scans_directory(self, mock_stat, mock_glob):
"""get_images() should scan output directory."""
with patch('shutil.which', return_value='/usr/bin/satdump'):
decoder = WeatherSatDecoder()
# Mock image files
mock_file = MagicMock()
mock_file.name = 'NOAA-18_test.png'
mock_file.stat.return_value.st_size = 10000
mock_file.stat.return_value.st_mtime = time.time()
mock_glob.return_value = [mock_file]
images = decoder.get_images()
assert len(images) == 1
assert images[0].filename == 'NOAA-18_test.png'
assert images[0].satellite == 'NOAA-18'
def test_delete_image_success(self):
"""delete_image() should delete file."""
with patch('shutil.which', return_value='/usr/bin/satdump'):
decoder = WeatherSatDecoder()
with patch('pathlib.Path.exists', return_value=True), \
patch('pathlib.Path.unlink') as mock_unlink:
result = decoder.delete_image('test.png')
assert result is True
mock_unlink.assert_called_once()
def test_delete_image_not_found(self):
"""delete_image() should return False for non-existent file."""
with patch('shutil.which', return_value='/usr/bin/satdump'):
decoder = WeatherSatDecoder()
with patch('pathlib.Path.exists', return_value=False):
result = decoder.delete_image('missing.png')
assert result is False
def test_delete_all_images(self):
"""delete_all_images() should delete all images."""
with patch('shutil.which', return_value='/usr/bin/satdump'):
decoder = WeatherSatDecoder()
mock_files = [MagicMock() for _ in range(3)]
with patch('pathlib.Path.glob', return_value=mock_files):
count = decoder.delete_all_images()
assert count == 3
for f in mock_files:
f.unlink.assert_called_once()
def test_get_status_idle(self):
"""get_status() should return idle status."""
with patch('shutil.which', return_value='/usr/bin/satdump'):
decoder = WeatherSatDecoder()
status = decoder.get_status()
assert status['available'] is True
assert status['decoder'] == 'satdump'
assert status['running'] is False
assert status['satellite'] == ''
def test_get_status_running(self):
"""get_status() should return running status."""
with patch('shutil.which', return_value='/usr/bin/satdump'):
decoder = WeatherSatDecoder()
decoder._running = True
decoder._current_satellite = 'NOAA-18'
decoder._current_frequency = 137.9125
decoder._current_mode = 'APT'
decoder._capture_start_time = time.time() - 60
status = decoder.get_status()
assert status['running'] is True
assert status['satellite'] == 'NOAA-18'
assert status['frequency'] == 137.9125
assert status['mode'] == 'APT'
assert status['elapsed_seconds'] >= 60
def test_classify_log_type_error(self):
"""_classify_log_type() should detect errors."""
assert WeatherSatDecoder._classify_log_type('(E) Error occurred') == 'error'
assert WeatherSatDecoder._classify_log_type('Failed to open device') == 'error'
def test_classify_log_type_progress(self):
"""_classify_log_type() should detect progress."""
assert WeatherSatDecoder._classify_log_type('Progress: 50%') == 'progress'
def test_classify_log_type_save(self):
"""_classify_log_type() should detect save events."""
assert WeatherSatDecoder._classify_log_type('Saved image: test.png') == 'save'
assert WeatherSatDecoder._classify_log_type('Writing output file') == 'save'
def test_classify_log_type_signal(self):
"""_classify_log_type() should detect signal events."""
assert WeatherSatDecoder._classify_log_type('Signal detected') == 'signal'
assert WeatherSatDecoder._classify_log_type('Lock acquired') == 'signal'
def test_classify_log_type_warning(self):
"""_classify_log_type() should detect warnings."""
assert WeatherSatDecoder._classify_log_type('(W) Low signal quality') == 'warning'
def test_classify_log_type_debug(self):
"""_classify_log_type() should detect debug messages."""
assert WeatherSatDecoder._classify_log_type('(D) Debug info') == 'debug'
@patch('subprocess.run')
def test_resolve_device_id_success(self, mock_run):
"""_resolve_device_id() should extract serial from rtl_test."""
mock_result = MagicMock()
mock_result.stdout = 'Found 1 device(s):\n 0: RTLSDRBlog, SN: 00004000'
mock_result.stderr = ''
mock_run.return_value = mock_result
serial = WeatherSatDecoder._resolve_device_id(0)
assert serial == '00004000'
mock_run.assert_called_once()
@patch('subprocess.run')
def test_resolve_device_id_fallback(self, mock_run):
"""_resolve_device_id() should fall back to index string."""
mock_run.side_effect = FileNotFoundError
serial = WeatherSatDecoder._resolve_device_id(0)
assert serial == '0'
def test_parse_product_name_rgb(self):
"""_parse_product_name() should identify RGB composite."""
with patch('shutil.which', return_value='/usr/bin/satdump'):
decoder = WeatherSatDecoder()
product = decoder._parse_product_name(Path('/tmp/output/rgb_composite.png'))
assert product == 'RGB Composite'
def test_parse_product_name_thermal(self):
"""_parse_product_name() should identify thermal imagery."""
with patch('shutil.which', return_value='/usr/bin/satdump'):
decoder = WeatherSatDecoder()
product = decoder._parse_product_name(Path('/tmp/output/thermal_image.png'))
assert product == 'Thermal'
def test_parse_product_name_channel(self):
"""_parse_product_name() should identify channel images."""
with patch('shutil.which', return_value='/usr/bin/satdump'):
decoder = WeatherSatDecoder()
product = decoder._parse_product_name(Path('/tmp/output/channel_3.png'))
assert product == 'Channel 3'
def test_parse_product_name_unknown(self):
"""_parse_product_name() should return stem for unknown products."""
with patch('shutil.which', return_value='/usr/bin/satdump'):
decoder = WeatherSatDecoder()
product = decoder._parse_product_name(Path('/tmp/output/unknown_image.png'))
assert product == 'unknown_image'
def test_emit_progress(self):
"""_emit_progress() should call callback."""
with patch('shutil.which', return_value='/usr/bin/satdump'):
decoder = WeatherSatDecoder()
callback = MagicMock()
decoder.set_callback(callback)
progress = CaptureProgress(status='capturing', message='Test')
decoder._emit_progress(progress)
callback.assert_called_once_with(progress)
def test_emit_progress_no_callback(self):
"""_emit_progress() should handle missing callback."""
with patch('shutil.which', return_value='/usr/bin/satdump'):
decoder = WeatherSatDecoder()
progress = CaptureProgress(status='capturing', message='Test')
decoder._emit_progress(progress) # Should not raise
def test_emit_progress_callback_exception(self):
"""_emit_progress() should handle callback exceptions."""
with patch('shutil.which', return_value='/usr/bin/satdump'):
decoder = WeatherSatDecoder()
callback = MagicMock(side_effect=Exception('Callback error'))
decoder.set_callback(callback)
progress = CaptureProgress(status='capturing', message='Test')
decoder._emit_progress(progress) # Should not raise
class TestWeatherSatImage:
"""Tests for WeatherSatImage dataclass."""
def test_to_dict(self):
"""WeatherSatImage.to_dict() should serialize correctly."""
image = WeatherSatImage(
filename='test.png',
path=Path('/tmp/test.png'),
satellite='NOAA-18',
mode='APT',
timestamp=datetime(2024, 1, 1, 12, 0, 0, tzinfo=timezone.utc),
frequency=137.9125,
size_bytes=12345,
product='RGB Composite',
)
data = image.to_dict()
assert data['filename'] == 'test.png'
assert data['satellite'] == 'NOAA-18'
assert data['mode'] == 'APT'
assert data['timestamp'] == '2024-01-01T12:00:00+00:00'
assert data['frequency'] == 137.9125
assert data['size_bytes'] == 12345
assert data['product'] == 'RGB Composite'
assert data['url'] == '/weather-sat/images/test.png'
class TestCaptureProgress:
"""Tests for CaptureProgress dataclass."""
def test_to_dict_minimal(self):
"""CaptureProgress.to_dict() with minimal fields."""
progress = CaptureProgress(status='idle')
data = progress.to_dict()
assert data['type'] == 'weather_sat_progress'
assert data['status'] == 'idle'
assert data['satellite'] == ''
assert data['message'] == ''
assert data['progress'] == 0
def test_to_dict_complete(self):
"""CaptureProgress.to_dict() with all fields."""
image = WeatherSatImage(
filename='test.png',
path=Path('/tmp/test.png'),
satellite='NOAA-18',
mode='APT',
timestamp=datetime.now(timezone.utc),
frequency=137.9125,
)
progress = CaptureProgress(
status='complete',
satellite='NOAA-18',
frequency=137.9125,
mode='APT',
message='Capture complete',
progress_percent=100,
elapsed_seconds=600,
image=image,
log_type='info',
capture_phase='complete',
)
data = progress.to_dict()
assert data['status'] == 'complete'
assert data['satellite'] == 'NOAA-18'
assert data['frequency'] == 137.9125
assert data['mode'] == 'APT'
assert data['message'] == 'Capture complete'
assert data['progress'] == 100
assert data['elapsed_seconds'] == 600
assert 'image' in data
assert data['log_type'] == 'info'
assert data['capture_phase'] == 'complete'
class TestGlobalFunctions:
"""Tests for global utility functions."""
def test_get_weather_sat_decoder_singleton(self):
"""get_weather_sat_decoder() should return singleton."""
with patch('shutil.which', return_value='/usr/bin/satdump'):
import utils.weather_sat as mod
old = mod._decoder
mod._decoder = None
try:
decoder1 = get_weather_sat_decoder()
decoder2 = get_weather_sat_decoder()
assert decoder1 is decoder2
finally:
mod._decoder = old
def test_is_weather_sat_available_true(self):
"""is_weather_sat_available() should return True when available."""
with patch('shutil.which', return_value='/usr/bin/satdump'):
import utils.weather_sat as mod
old = mod._decoder
mod._decoder = None
try:
assert is_weather_sat_available() is True
finally:
mod._decoder = old
def test_is_weather_sat_available_false(self):
"""is_weather_sat_available() should return False when unavailable."""
with patch('shutil.which', return_value=None):
import utils.weather_sat as mod
old = mod._decoder
mod._decoder = None
try:
assert is_weather_sat_available() is False
finally:
mod._decoder = old
class TestWeatherSatellitesConstant:
"""Tests for WEATHER_SATELLITES constant."""
def test_weather_satellites_structure(self):
"""WEATHER_SATELLITES should have correct structure."""
assert 'NOAA-18' in WEATHER_SATELLITES
sat = WEATHER_SATELLITES['NOAA-18']
assert 'name' in sat
assert 'frequency' in sat
assert 'mode' in sat
assert 'pipeline' in sat
assert 'tle_key' in sat
assert 'description' in sat
assert 'active' in sat
def test_noaa_satellites(self):
"""NOAA satellites should have correct frequencies."""
assert WEATHER_SATELLITES['NOAA-15']['frequency'] == 137.620
assert WEATHER_SATELLITES['NOAA-18']['frequency'] == 137.9125
assert WEATHER_SATELLITES['NOAA-19']['frequency'] == 137.100
def test_meteor_satellite(self):
"""Meteor satellite should use LRPT mode."""
meteor = WEATHER_SATELLITES['METEOR-M2-3']
assert meteor['mode'] == 'LRPT'
assert meteor['frequency'] == 137.900
assert meteor['pipeline'] == 'meteor_m2-x_lrpt'

675
tests/test_weather_sat_predict.py Normal file
View File

@@ -0,0 +1,675 @@
"""Tests for weather satellite pass prediction.
Covers predict_passes() function, TLE handling, trajectory computation,
and ground track generation.
"""
from __future__ import annotations
from datetime import datetime, timezone, timedelta
from unittest.mock import patch, MagicMock
import pytest
from utils.weather_sat_predict import predict_passes
class TestPredictPasses:
"""Tests for predict_passes() function."""
@patch('utils.weather_sat_predict.load')
@patch('utils.weather_sat_predict.TLE_SATELLITES')
def test_predict_passes_no_tle_data(self, mock_tle, mock_load):
"""predict_passes() should handle missing TLE data."""
mock_tle.get.return_value = None
mock_ts = MagicMock()
mock_ts.now.return_value = MagicMock()
mock_ts.utc.return_value = MagicMock()
mock_load.timescale.return_value = mock_ts
passes = predict_passes(lat=51.5, lon=-0.1, hours=24, min_elevation=15)
assert passes == []
@patch('utils.weather_sat_predict.load')
@patch('utils.weather_sat_predict.TLE_SATELLITES')
@patch('utils.weather_sat_predict.wgs84')
@patch('utils.weather_sat_predict.EarthSatellite')
@patch('utils.weather_sat_predict.find_discrete')
def test_predict_passes_basic(self, mock_find, mock_sat, mock_wgs84, mock_tle, mock_load):
"""predict_passes() should predict basic passes."""
# Mock timescale
mock_ts = MagicMock()
now = datetime(2024, 1, 1, 12, 0, 0, tzinfo=timezone.utc)
mock_now = MagicMock()
mock_now.utc_datetime.return_value = now
mock_ts.now.return_value = mock_now
mock_ts.utc.side_effect = lambda dt: self._mock_time(dt)
mock_load.timescale.return_value = mock_ts
# Mock TLE data
mock_tle.get.return_value = (
'NOAA-18',
'1 28654U 05018A 24001.50000000 .00000000 00000-0 00000-0 0 9999',
'2 28654 98.7000 100.0000 0001000 0.0000 0.0000 14.12500000000000'
)
# Mock observer
mock_observer = MagicMock()
mock_wgs84.latlon.return_value = mock_observer
# Mock satellite
mock_satellite_obj = MagicMock()
mock_sat.return_value = mock_satellite_obj
# Mock pass detection - one pass
rise_time = MagicMock()
rise_time.utc_datetime.return_value = now + timedelta(hours=2)
set_time = MagicMock()
set_time.utc_datetime.return_value = now + timedelta(hours=2, minutes=15)
mock_find.return_value = ([rise_time, set_time], [True, False])
# Mock topocentric calculations
def mock_topocentric(t):
topo = MagicMock()
alt = MagicMock()
alt.degrees = 45.0
az = MagicMock()
az.degrees = 180.0
topo.altaz.return_value = (alt, az, MagicMock())
return topo
mock_diff = MagicMock()
mock_diff.at.side_effect = mock_topocentric
mock_satellite_obj.__sub__.return_value = mock_diff
passes = predict_passes(lat=51.5, lon=-0.1, hours=24, min_elevation=15)
assert len(passes) == 1
pass_data = passes[0]
assert pass_data['satellite'] == 'NOAA-18'
assert pass_data['name'] == 'NOAA 18'
assert pass_data['frequency'] == 137.9125
assert pass_data['mode'] == 'APT'
assert 'maxEl' in pass_data
assert 'duration' in pass_data
assert 'quality' in pass_data
@patch('utils.weather_sat_predict.load')
@patch('utils.weather_sat_predict.TLE_SATELLITES')
@patch('utils.weather_sat_predict.wgs84')
@patch('utils.weather_sat_predict.EarthSatellite')
@patch('utils.weather_sat_predict.find_discrete')
def test_predict_passes_below_min_elevation(
self, mock_find, mock_sat, mock_wgs84, mock_tle, mock_load
):
"""predict_passes() should filter passes below min elevation."""
mock_ts = MagicMock()
now = datetime(2024, 1, 1, 12, 0, 0, tzinfo=timezone.utc)
mock_now = MagicMock()
mock_now.utc_datetime.return_value = now
mock_ts.now.return_value = mock_now
mock_ts.utc.side_effect = lambda dt: self._mock_time(dt)
mock_load.timescale.return_value = mock_ts
mock_tle.get.return_value = (
'NOAA-18',
'1 28654U 05018A 24001.50000000 .00000000 00000-0 00000-0 0 9999',
'2 28654 98.7000 100.0000 0001000 0.0000 0.0000 14.12500000000000'
)
mock_observer = MagicMock()
mock_wgs84.latlon.return_value = mock_observer
mock_satellite_obj = MagicMock()
mock_sat.return_value = mock_satellite_obj
rise_time = MagicMock()
rise_time.utc_datetime.return_value = now + timedelta(hours=2)
set_time = MagicMock()
set_time.utc_datetime.return_value = now + timedelta(hours=2, minutes=15)
mock_find.return_value = ([rise_time, set_time], [True, False])
# Mock low elevation pass
def mock_topocentric(t):
topo = MagicMock()
alt = MagicMock()
alt.degrees = 10.0 # Below min_elevation of 15
az = MagicMock()
az.degrees = 180.0
topo.altaz.return_value = (alt, az, MagicMock())
return topo
mock_diff = MagicMock()
mock_diff.at.side_effect = mock_topocentric
mock_satellite_obj.__sub__.return_value = mock_diff
passes = predict_passes(lat=51.5, lon=-0.1, hours=24, min_elevation=15)
assert len(passes) == 0
@patch('utils.weather_sat_predict.load')
@patch('utils.weather_sat_predict.TLE_SATELLITES')
@patch('utils.weather_sat_predict.wgs84')
@patch('utils.weather_sat_predict.EarthSatellite')
@patch('utils.weather_sat_predict.find_discrete')
def test_predict_passes_with_trajectory(
self, mock_find, mock_sat, mock_wgs84, mock_tle, mock_load
):
"""predict_passes() should include trajectory when requested."""
mock_ts = MagicMock()
now = datetime(2024, 1, 1, 12, 0, 0, tzinfo=timezone.utc)
mock_now = MagicMock()
mock_now.utc_datetime.return_value = now
mock_ts.now.return_value = mock_now
mock_ts.utc.side_effect = lambda dt: self._mock_time(dt)
mock_load.timescale.return_value = mock_ts
mock_tle.get.return_value = (
'NOAA-18',
'1 28654U 05018A 24001.50000000 .00000000 00000-0 00000-0 0 9999',
'2 28654 98.7000 100.0000 0001000 0.0000 0.0000 14.12500000000000'
)
mock_observer = MagicMock()
mock_wgs84.latlon.return_value = mock_observer
mock_satellite_obj = MagicMock()
mock_sat.return_value = mock_satellite_obj
rise_time = MagicMock()
rise_time.utc_datetime.return_value = now + timedelta(hours=2)
set_time = MagicMock()
set_time.utc_datetime.return_value = now + timedelta(hours=2, minutes=15)
mock_find.return_value = ([rise_time, set_time], [True, False])
def mock_topocentric(t):
topo = MagicMock()
alt = MagicMock()
alt.degrees = 45.0
az = MagicMock()
az.degrees = 180.0
topo.altaz.return_value = (alt, az, MagicMock())
return topo
mock_diff = MagicMock()
mock_diff.at.side_effect = mock_topocentric
mock_satellite_obj.__sub__.return_value = mock_diff
passes = predict_passes(
lat=51.5, lon=-0.1, hours=24, min_elevation=15, include_trajectory=True
)
assert len(passes) == 1
assert 'trajectory' in passes[0]
assert len(passes[0]['trajectory']) == 30
@patch('utils.weather_sat_predict.load')
@patch('utils.weather_sat_predict.TLE_SATELLITES')
@patch('utils.weather_sat_predict.wgs84')
@patch('utils.weather_sat_predict.EarthSatellite')
@patch('utils.weather_sat_predict.find_discrete')
def test_predict_passes_with_ground_track(
self, mock_find, mock_sat, mock_wgs84, mock_tle, mock_load
):
"""predict_passes() should include ground track when requested."""
mock_ts = MagicMock()
now = datetime(2024, 1, 1, 12, 0, 0, tzinfo=timezone.utc)
mock_now = MagicMock()
mock_now.utc_datetime.return_value = now
mock_ts.now.return_value = mock_now
mock_ts.utc.side_effect = lambda dt: self._mock_time(dt)
mock_load.timescale.return_value = mock_ts
mock_tle.get.return_value = (
'NOAA-18',
'1 28654U 05018A 24001.50000000 .00000000 00000-0 00000-0 0 9999',
'2 28654 98.7000 100.0000 0001000 0.0000 0.0000 14.12500000000000'
)
mock_observer = MagicMock()
mock_wgs84.latlon.return_value = mock_observer
mock_satellite_obj = MagicMock()
mock_sat.return_value = mock_satellite_obj
rise_time = MagicMock()
rise_time.utc_datetime.return_value = now + timedelta(hours=2)
set_time = MagicMock()
set_time.utc_datetime.return_value = now + timedelta(hours=2, minutes=15)
mock_find.return_value = ([rise_time, set_time], [True, False])
def mock_topocentric(t):
topo = MagicMock()
alt = MagicMock()
alt.degrees = 45.0
az = MagicMock()
az.degrees = 180.0
topo.altaz.return_value = (alt, az, MagicMock())
return topo
mock_diff = MagicMock()
mock_diff.at.side_effect = mock_topocentric
mock_satellite_obj.__sub__.return_value = mock_diff
# Mock geocentric position
def mock_at(t):
geocentric = MagicMock()
return geocentric
mock_satellite_obj.at.side_effect = mock_at
# Mock subpoint
mock_subpoint = MagicMock()
mock_lat = MagicMock()
mock_lat.degrees = 51.5
mock_lon = MagicMock()
mock_lon.degrees = -0.1
mock_subpoint.latitude = mock_lat
mock_subpoint.longitude = mock_lon
mock_wgs84.subpoint.return_value = mock_subpoint
passes = predict_passes(
lat=51.5, lon=-0.1, hours=24, min_elevation=15, include_ground_track=True
)
assert len(passes) == 1
assert 'groundTrack' in passes[0]
assert len(passes[0]['groundTrack']) == 60
@patch('utils.weather_sat_predict.load')
@patch('utils.weather_sat_predict.TLE_SATELLITES')
@patch('utils.weather_sat_predict.wgs84')
@patch('utils.weather_sat_predict.EarthSatellite')
@patch('utils.weather_sat_predict.find_discrete')
def test_predict_passes_quality_excellent(
self, mock_find, mock_sat, mock_wgs84, mock_tle, mock_load
):
"""predict_passes() should mark high elevation passes as excellent."""
mock_ts = MagicMock()
now = datetime(2024, 1, 1, 12, 0, 0, tzinfo=timezone.utc)
mock_now = MagicMock()
mock_now.utc_datetime.return_value = now
mock_ts.now.return_value = mock_now
mock_ts.utc.side_effect = lambda dt: self._mock_time(dt)
mock_load.timescale.return_value = mock_ts
mock_tle.get.return_value = (
'NOAA-18',
'1 28654U 05018A 24001.50000000 .00000000 00000-0 00000-0 0 9999',
'2 28654 98.7000 100.0000 0001000 0.0000 0.0000 14.12500000000000'
)
mock_observer = MagicMock()
mock_wgs84.latlon.return_value = mock_observer
mock_satellite_obj = MagicMock()
mock_sat.return_value = mock_satellite_obj
rise_time = MagicMock()
rise_time.utc_datetime.return_value = now + timedelta(hours=2)
set_time = MagicMock()
set_time.utc_datetime.return_value = now + timedelta(hours=2, minutes=15)
mock_find.return_value = ([rise_time, set_time], [True, False])
def mock_topocentric(t):
topo = MagicMock()
alt = MagicMock()
alt.degrees = 75.0 # Excellent pass
az = MagicMock()
az.degrees = 180.0
topo.altaz.return_value = (alt, az, MagicMock())
return topo
mock_diff = MagicMock()
mock_diff.at.side_effect = mock_topocentric
mock_satellite_obj.__sub__.return_value = mock_diff
passes = predict_passes(lat=51.5, lon=-0.1, hours=24, min_elevation=15)
assert len(passes) == 1
assert passes[0]['quality'] == 'excellent'
assert passes[0]['maxEl'] >= 60
@patch('utils.weather_sat_predict.load')
@patch('utils.weather_sat_predict.TLE_SATELLITES')
@patch('utils.weather_sat_predict.wgs84')
@patch('utils.weather_sat_predict.EarthSatellite')
@patch('utils.weather_sat_predict.find_discrete')
def test_predict_passes_quality_good(
self, mock_find, mock_sat, mock_wgs84, mock_tle, mock_load
):
"""predict_passes() should mark medium elevation passes as good."""
mock_ts = MagicMock()
now = datetime(2024, 1, 1, 12, 0, 0, tzinfo=timezone.utc)
mock_now = MagicMock()
mock_now.utc_datetime.return_value = now
mock_ts.now.return_value = mock_now
mock_ts.utc.side_effect = lambda dt: self._mock_time(dt)
mock_load.timescale.return_value = mock_ts
mock_tle.get.return_value = (
'NOAA-18',
'1 28654U 05018A 24001.50000000 .00000000 00000-0 00000-0 0 9999',
'2 28654 98.7000 100.0000 0001000 0.0000 0.0000 14.12500000000000'
)
mock_observer = MagicMock()
mock_wgs84.latlon.return_value = mock_observer
mock_satellite_obj = MagicMock()
mock_sat.return_value = mock_satellite_obj
rise_time = MagicMock()
rise_time.utc_datetime.return_value = now + timedelta(hours=2)
set_time = MagicMock()
set_time.utc_datetime.return_value = now + timedelta(hours=2, minutes=15)
mock_find.return_value = ([rise_time, set_time], [True, False])
def mock_topocentric(t):
topo = MagicMock()
alt = MagicMock()
alt.degrees = 45.0 # Good pass
az = MagicMock()
az.degrees = 180.0
topo.altaz.return_value = (alt, az, MagicMock())
return topo
mock_diff = MagicMock()
mock_diff.at.side_effect = mock_topocentric
mock_satellite_obj.__sub__.return_value = mock_diff
passes = predict_passes(lat=51.5, lon=-0.1, hours=24, min_elevation=15)
assert len(passes) == 1
assert passes[0]['quality'] == 'good'
assert 30 <= passes[0]['maxEl'] < 60
@patch('utils.weather_sat_predict.load')
@patch('utils.weather_sat_predict.TLE_SATELLITES')
@patch('utils.weather_sat_predict.wgs84')
@patch('utils.weather_sat_predict.EarthSatellite')
@patch('utils.weather_sat_predict.find_discrete')
def test_predict_passes_quality_fair(
self, mock_find, mock_sat, mock_wgs84, mock_tle, mock_load
):
"""predict_passes() should mark low elevation passes as fair."""
mock_ts = MagicMock()
now = datetime(2024, 1, 1, 12, 0, 0, tzinfo=timezone.utc)
mock_now = MagicMock()
mock_now.utc_datetime.return_value = now
mock_ts.now.return_value = mock_now
mock_ts.utc.side_effect = lambda dt: self._mock_time(dt)
mock_load.timescale.return_value = mock_ts
mock_tle.get.return_value = (
'NOAA-18',
'1 28654U 05018A 24001.50000000 .00000000 00000-0 00000-0 0 9999',
'2 28654 98.7000 100.0000 0001000 0.0000 0.0000 14.12500000000000'
)
mock_observer = MagicMock()
mock_wgs84.latlon.return_value = mock_observer
mock_satellite_obj = MagicMock()
mock_sat.return_value = mock_satellite_obj
rise_time = MagicMock()
rise_time.utc_datetime.return_value = now + timedelta(hours=2)
set_time = MagicMock()
set_time.utc_datetime.return_value = now + timedelta(hours=2, minutes=15)
mock_find.return_value = ([rise_time, set_time], [True, False])
def mock_topocentric(t):
topo = MagicMock()
alt = MagicMock()
alt.degrees = 20.0 # Fair pass
az = MagicMock()
az.degrees = 180.0
topo.altaz.return_value = (alt, az, MagicMock())
return topo
mock_diff = MagicMock()
mock_diff.at.side_effect = mock_topocentric
mock_satellite_obj.__sub__.return_value = mock_diff
passes = predict_passes(lat=51.5, lon=-0.1, hours=24, min_elevation=15)
assert len(passes) == 1
assert passes[0]['quality'] == 'fair'
assert passes[0]['maxEl'] < 30
@patch('utils.weather_sat_predict.load')
@patch('utils.weather_sat_predict.TLE_SATELLITES')
@patch('utils.weather_sat_predict.wgs84')
@patch('utils.weather_sat_predict.EarthSatellite')
@patch('utils.weather_sat_predict.find_discrete')
def test_predict_passes_inactive_satellite(
self, mock_find, mock_sat, mock_wgs84, mock_tle, mock_load
):
"""predict_passes() should skip inactive satellites."""
mock_ts = MagicMock()
now = datetime(2024, 1, 1, 12, 0, 0, tzinfo=timezone.utc)
mock_now = MagicMock()
mock_now.utc_datetime.return_value = now
mock_ts.now.return_value = mock_now
mock_load.timescale.return_value = mock_ts
# Temporarily mark satellite as inactive
from utils.weather_sat import WEATHER_SATELLITES
original_active = WEATHER_SATELLITES['NOAA-18']['active']
WEATHER_SATELLITES['NOAA-18']['active'] = False
try:
passes = predict_passes(lat=51.5, lon=-0.1, hours=24, min_elevation=15)
# Should not include NOAA-18
noaa_18_passes = [p for p in passes if p['satellite'] == 'NOAA-18']
assert len(noaa_18_passes) == 0
finally:
WEATHER_SATELLITES['NOAA-18']['active'] = original_active
@patch('utils.weather_sat_predict.load')
@patch('utils.weather_sat_predict.TLE_SATELLITES')
@patch('utils.weather_sat_predict.wgs84')
@patch('utils.weather_sat_predict.EarthSatellite')
@patch('utils.weather_sat_predict.find_discrete')
def test_predict_passes_exception_handling(
self, mock_find, mock_sat, mock_wgs84, mock_tle, mock_load
):
"""predict_passes() should handle exceptions gracefully."""
mock_ts = MagicMock()
now = datetime(2024, 1, 1, 12, 0, 0, tzinfo=timezone.utc)
mock_now = MagicMock()
mock_now.utc_datetime.return_value = now
mock_ts.now.return_value = mock_now
mock_ts.utc.side_effect = lambda dt: self._mock_time(dt)
mock_load.timescale.return_value = mock_ts
mock_tle.get.return_value = (
'NOAA-18',
'1 28654U 05018A 24001.50000000 .00000000 00000-0 00000-0 0 9999',
'2 28654 98.7000 100.0000 0001000 0.0000 0.0000 14.12500000000000'
)
mock_observer = MagicMock()
mock_wgs84.latlon.return_value = mock_observer
mock_satellite_obj = MagicMock()
mock_sat.return_value = mock_satellite_obj
# Make find_discrete raise exception
mock_find.side_effect = Exception('Computation error')
# Should not raise, just skip this satellite
passes = predict_passes(lat=51.5, lon=-0.1, hours=24, min_elevation=15)
# May include passes from other satellites or be empty
assert isinstance(passes, list)
@patch('utils.weather_sat_predict.load')
@patch('utils.weather_sat_predict.TLE_SATELLITES')
def test_predict_passes_uses_tle_cache(self, mock_tle, mock_load):
"""predict_passes() should use live TLE cache if available."""
with patch('utils.weather_sat_predict._tle_cache', {'NOAA-18': ('NOAA-18', 'line1', 'line2')}):
mock_ts = MagicMock()
mock_ts.now.return_value = MagicMock()
mock_ts.utc.return_value = MagicMock()
mock_load.timescale.return_value = mock_ts
# Even though TLE_SATELLITES is mocked, should use _tle_cache
with patch('utils.weather_sat_predict.wgs84'), \
patch('utils.weather_sat_predict.EarthSatellite'), \
patch('utils.weather_sat_predict.find_discrete', return_value=([], [])):
passes = predict_passes(lat=51.5, lon=-0.1, hours=24, min_elevation=15)
# Should not raise
@patch('utils.weather_sat_predict.load')
@patch('utils.weather_sat_predict.TLE_SATELLITES')
@patch('utils.weather_sat_predict.wgs84')
@patch('utils.weather_sat_predict.EarthSatellite')
@patch('utils.weather_sat_predict.find_discrete')
def test_predict_passes_sorted_by_time(
self, mock_find, mock_sat, mock_wgs84, mock_tle, mock_load
):
"""predict_passes() should return passes sorted by start time."""
mock_ts = MagicMock()
now = datetime(2024, 1, 1, 12, 0, 0, tzinfo=timezone.utc)
mock_now = MagicMock()
mock_now.utc_datetime.return_value = now
mock_ts.now.return_value = mock_now
mock_ts.utc.side_effect = lambda dt: self._mock_time(dt)
mock_load.timescale.return_value = mock_ts
mock_tle.get.return_value = (
'NOAA-18',
'1 28654U 05018A 24001.50000000 .00000000 00000-0 00000-0 0 9999',
'2 28654 98.7000 100.0000 0001000 0.0000 0.0000 14.12500000000000'
)
mock_observer = MagicMock()
mock_wgs84.latlon.return_value = mock_observer
mock_satellite_obj = MagicMock()
mock_sat.return_value = mock_satellite_obj
# Two passes
rise1 = MagicMock()
rise1.utc_datetime.return_value = now + timedelta(hours=4)
set1 = MagicMock()
set1.utc_datetime.return_value = now + timedelta(hours=4, minutes=15)
rise2 = MagicMock()
rise2.utc_datetime.return_value = now + timedelta(hours=2)
set2 = MagicMock()
set2.utc_datetime.return_value = now + timedelta(hours=2, minutes=15)
# Return in non-chronological order
mock_find.return_value = ([rise1, set1, rise2, set2], [True, False, True, False])
def mock_topocentric(t):
topo = MagicMock()
alt = MagicMock()
alt.degrees = 45.0
az = MagicMock()
az.degrees = 180.0
topo.altaz.return_value = (alt, az, MagicMock())
return topo
mock_diff = MagicMock()
mock_diff.at.side_effect = mock_topocentric
mock_satellite_obj.__sub__.return_value = mock_diff
passes = predict_passes(lat=51.5, lon=-0.1, hours=24, min_elevation=15)
# Should be sorted with earliest pass first
if len(passes) >= 2:
assert passes[0]['startTimeISO'] < passes[1]['startTimeISO']
@staticmethod
def _mock_time(dt):
"""Helper to create mock time object."""
mock_t = MagicMock()
if isinstance(dt, datetime):
mock_t.utc_datetime.return_value = dt
else:
mock_t.utc_datetime.return_value = datetime.now(timezone.utc)
return mock_t
class TestPassDataStructure:
"""Tests for pass data structure."""
@patch('utils.weather_sat_predict.load')
@patch('utils.weather_sat_predict.TLE_SATELLITES')
@patch('utils.weather_sat_predict.wgs84')
@patch('utils.weather_sat_predict.EarthSatellite')
@patch('utils.weather_sat_predict.find_discrete')
def test_pass_data_fields(
self, mock_find, mock_sat, mock_wgs84, mock_tle, mock_load
):
"""Pass data should contain all required fields."""
mock_ts = MagicMock()
now = datetime(2024, 1, 1, 12, 0, 0, tzinfo=timezone.utc)
mock_now = MagicMock()
mock_now.utc_datetime.return_value = now
mock_ts.now.return_value = mock_now
mock_ts.utc.side_effect = lambda dt: TestPredictPasses._mock_time(dt)
mock_load.timescale.return_value = mock_ts
mock_tle.get.return_value = (
'NOAA-18',
'1 28654U 05018A 24001.50000000 .00000000 00000-0 00000-0 0 9999',
'2 28654 98.7000 100.0000 0001000 0.0000 0.0000 14.12500000000000'
)
mock_observer = MagicMock()
mock_wgs84.latlon.return_value = mock_observer
mock_satellite_obj = MagicMock()
mock_sat.return_value = mock_satellite_obj
rise_time = MagicMock()
rise_time.utc_datetime.return_value = now + timedelta(hours=2)
set_time = MagicMock()
set_time.utc_datetime.return_value = now + timedelta(hours=2, minutes=15)
mock_find.return_value = ([rise_time, set_time], [True, False])
def mock_topocentric(t):
topo = MagicMock()
alt = MagicMock()
alt.degrees = 45.0
az = MagicMock()
az.degrees = 180.0
topo.altaz.return_value = (alt, az, MagicMock())
return topo
mock_diff = MagicMock()
mock_diff.at.side_effect = mock_topocentric
mock_satellite_obj.__sub__.return_value = mock_diff
passes = predict_passes(lat=51.5, lon=-0.1, hours=24, min_elevation=15)
assert len(passes) == 1
pass_data = passes[0]
# Check all required fields
required_fields = [
'id', 'satellite', 'name', 'frequency', 'mode',
'startTime', 'startTimeISO', 'endTimeISO',
'maxEl', 'maxElAz', 'riseAz', 'setAz',
'duration', 'quality'
]
for field in required_fields:
assert field in pass_data, f"Missing required field: {field}"
def test_import_error_propagates(self):
"""predict_passes() should raise ImportError if skyfield unavailable."""
with patch.dict('sys.modules', {'skyfield': None, 'skyfield.api': None}):
with pytest.raises((ImportError, AttributeError)):
predict_passes(lat=51.5, lon=-0.1)

801
tests/test_weather_sat_routes.py Normal file
View File

@@ -0,0 +1,801 @@
"""Tests for weather satellite routes.
Covers all weather_sat endpoints: /status, /satellites, /start, /test-decode,
/stop, /images, /passes, and scheduler endpoints.
"""
from __future__ import annotations
import json
from pathlib import Path
from unittest.mock import patch, MagicMock, mock_open
import pytest
from utils.weather_sat import WeatherSatImage, WEATHER_SATELLITES
from datetime import datetime, timezone
class TestWeatherSatRoutes:
"""Tests for weather satellite routes."""
def test_get_status(self, client):
"""GET /weather-sat/status returns decoder status."""
with patch('routes.weather_sat.get_weather_sat_decoder') as mock_get:
mock_decoder = MagicMock()
mock_decoder.get_status.return_value = {
'available': True,
'decoder': 'satdump',
'running': False,
'satellite': '',
'frequency': 0.0,
'mode': '',
'elapsed_seconds': 0,
'image_count': 0,
}
mock_get.return_value = mock_decoder
response = client.get('/weather-sat/status')
assert response.status_code == 200
data = response.get_json()
assert data['available'] is True
assert data['decoder'] == 'satdump'
assert data['running'] is False
def test_list_satellites(self, client):
"""GET /weather-sat/satellites returns satellite list."""
response = client.get('/weather-sat/satellites')
assert response.status_code == 200
data = response.get_json()
assert data['status'] == 'ok'
assert 'satellites' in data
assert len(data['satellites']) > 0
# Check structure
sat = data['satellites'][0]
assert 'key' in sat
assert 'name' in sat
assert 'frequency' in sat
assert 'mode' in sat
assert 'description' in sat
assert 'active' in sat
# Verify NOAA-18 is in list
noaa_18 = next((s for s in data['satellites'] if s['key'] == 'NOAA-18'), None)
assert noaa_18 is not None
assert noaa_18['frequency'] == 137.9125
assert noaa_18['mode'] == 'APT'
def test_start_capture_success(self, client):
"""POST /weather-sat/start successfully starts capture."""
with patch('routes.weather_sat.is_weather_sat_available', return_value=True), \
patch('routes.weather_sat.get_weather_sat_decoder') as mock_get, \
patch('routes.weather_sat.queue.Queue') as mock_queue:
mock_decoder = MagicMock()
mock_decoder.is_running = False
mock_decoder.start.return_value = True
mock_get.return_value = mock_decoder
payload = {
'satellite': 'NOAA-18',
'device': 0,
'gain': 40.0,
'bias_t': False,
}
response = client.post(
'/weather-sat/start',
data=json.dumps(payload),
content_type='application/json'
)
assert response.status_code == 200
data = response.get_json()
assert data['status'] == 'started'
assert data['satellite'] == 'NOAA-18'
assert data['frequency'] == 137.9125
assert data['mode'] == 'APT'
assert data['device'] == 0
mock_decoder.start.assert_called_once_with(
satellite='NOAA-18',
device_index=0,
gain=40.0,
bias_t=False,
)
def test_start_capture_no_satdump(self, client):
"""POST /weather-sat/start returns error when SatDump unavailable."""
with patch('routes.weather_sat.is_weather_sat_available', return_value=False):
payload = {'satellite': 'NOAA-18'}
response = client.post(
'/weather-sat/start',
data=json.dumps(payload),
content_type='application/json'
)
assert response.status_code == 400
data = response.get_json()
assert data['status'] == 'error'
assert 'SatDump not installed' in data['message']
def test_start_capture_already_running(self, client):
"""POST /weather-sat/start when already running."""
with patch('routes.weather_sat.is_weather_sat_available', return_value=True), \
patch('routes.weather_sat.get_weather_sat_decoder') as mock_get:
mock_decoder = MagicMock()
mock_decoder.is_running = True
mock_decoder.current_satellite = 'NOAA-19'
mock_decoder.current_frequency = 137.100
mock_get.return_value = mock_decoder
payload = {'satellite': 'NOAA-18'}
response = client.post(
'/weather-sat/start',
data=json.dumps(payload),
content_type='application/json'
)
assert response.status_code == 200
data = response.get_json()
assert data['status'] == 'already_running'
assert data['satellite'] == 'NOAA-19'
def test_start_capture_invalid_satellite(self, client):
"""POST /weather-sat/start with invalid satellite."""
with patch('routes.weather_sat.is_weather_sat_available', return_value=True), \
patch('routes.weather_sat.get_weather_sat_decoder') as mock_get:
mock_decoder = MagicMock()
mock_decoder.is_running = False
mock_get.return_value = mock_decoder
payload = {'satellite': 'FAKE-SAT-99'}
response = client.post(
'/weather-sat/start',
data=json.dumps(payload),
content_type='application/json'
)
assert response.status_code == 400
data = response.get_json()
assert data['status'] == 'error'
assert 'Invalid satellite' in data['message']
def test_start_capture_invalid_device(self, client):
"""POST /weather-sat/start with invalid device index."""
with patch('routes.weather_sat.is_weather_sat_available', return_value=True), \
patch('routes.weather_sat.get_weather_sat_decoder') as mock_get:
mock_decoder = MagicMock()
mock_decoder.is_running = False
mock_get.return_value = mock_decoder
payload = {'satellite': 'NOAA-18', 'device': -1}
response = client.post(
'/weather-sat/start',
data=json.dumps(payload),
content_type='application/json'
)
assert response.status_code == 400
data = response.get_json()
assert data['status'] == 'error'
def test_start_capture_invalid_gain(self, client):
"""POST /weather-sat/start with invalid gain."""
with patch('routes.weather_sat.is_weather_sat_available', return_value=True), \
patch('routes.weather_sat.get_weather_sat_decoder') as mock_get:
mock_decoder = MagicMock()
mock_decoder.is_running = False
mock_get.return_value = mock_decoder
payload = {'satellite': 'NOAA-18', 'gain': 999}
response = client.post(
'/weather-sat/start',
data=json.dumps(payload),
content_type='application/json'
)
assert response.status_code == 400
data = response.get_json()
assert data['status'] == 'error'
def test_start_capture_device_busy(self, client):
"""POST /weather-sat/start when SDR device is busy."""
with patch('routes.weather_sat.is_weather_sat_available', return_value=True), \
patch('routes.weather_sat.get_weather_sat_decoder') as mock_get, \
patch('app.claim_sdr_device', return_value='Device busy with pager') as mock_claim:
mock_decoder = MagicMock()
mock_decoder.is_running = False
mock_get.return_value = mock_decoder
payload = {'satellite': 'NOAA-18'}
response = client.post(
'/weather-sat/start',
data=json.dumps(payload),
content_type='application/json'
)
assert response.status_code == 409
data = response.get_json()
assert data['status'] == 'error'
assert data['error_type'] == 'DEVICE_BUSY'
assert 'Device busy' in data['message']
def test_start_capture_start_failure(self, client):
"""POST /weather-sat/start when decoder.start() fails."""
with patch('routes.weather_sat.is_weather_sat_available', return_value=True), \
patch('routes.weather_sat.get_weather_sat_decoder') as mock_get:
mock_decoder = MagicMock()
mock_decoder.is_running = False
mock_decoder.start.return_value = False
mock_get.return_value = mock_decoder
payload = {'satellite': 'NOAA-18'}
response = client.post(
'/weather-sat/start',
data=json.dumps(payload),
content_type='application/json'
)
assert response.status_code == 500
data = response.get_json()
assert data['status'] == 'error'
assert 'Failed to start capture' in data['message']
def test_test_decode_success(self, client):
"""POST /weather-sat/test-decode successfully starts file decode."""
with patch('routes.weather_sat.is_weather_sat_available', return_value=True), \
patch('routes.weather_sat.get_weather_sat_decoder') as mock_get, \
patch('pathlib.Path.is_file', return_value=True), \
patch('pathlib.Path.resolve') as mock_resolve:
# Mock path resolution to be under data/
mock_path = MagicMock()
mock_path.is_relative_to.return_value = True
mock_resolve.return_value = mock_path
mock_decoder = MagicMock()
mock_decoder.is_running = False
mock_decoder.start_from_file.return_value = True
mock_get.return_value = mock_decoder
payload = {
'satellite': 'NOAA-18',
'input_file': 'data/weather_sat/test.wav',
'sample_rate': 1000000,
}
response = client.post(
'/weather-sat/test-decode',
data=json.dumps(payload),
content_type='application/json'
)
assert response.status_code == 200
data = response.get_json()
assert data['status'] == 'started'
assert data['satellite'] == 'NOAA-18'
assert data['source'] == 'file'
def test_test_decode_invalid_path(self, client):
"""POST /weather-sat/test-decode with path outside data/."""
with patch('routes.weather_sat.is_weather_sat_available', return_value=True), \
patch('routes.weather_sat.get_weather_sat_decoder') as mock_get, \
patch('pathlib.Path.resolve') as mock_resolve:
# Mock path outside allowed directory
mock_path = MagicMock()
mock_path.is_relative_to.return_value = False
mock_resolve.return_value = mock_path
mock_decoder = MagicMock()
mock_decoder.is_running = False
mock_get.return_value = mock_decoder
payload = {
'satellite': 'NOAA-18',
'input_file': '/etc/passwd',
}
response = client.post(
'/weather-sat/test-decode',
data=json.dumps(payload),
content_type='application/json'
)
assert response.status_code == 403
data = response.get_json()
assert data['status'] == 'error'
assert 'data/ directory' in data['message']
def test_test_decode_file_not_found(self, client):
"""POST /weather-sat/test-decode with non-existent file."""
with patch('routes.weather_sat.is_weather_sat_available', return_value=True), \
patch('routes.weather_sat.get_weather_sat_decoder') as mock_get, \
patch('pathlib.Path.is_file', return_value=False), \
patch('pathlib.Path.resolve') as mock_resolve:
mock_path = MagicMock()
mock_path.is_relative_to.return_value = True
mock_resolve.return_value = mock_path
mock_decoder = MagicMock()
mock_decoder.is_running = False
mock_get.return_value = mock_decoder
payload = {
'satellite': 'NOAA-18',
'input_file': 'data/missing.wav',
}
response = client.post(
'/weather-sat/test-decode',
data=json.dumps(payload),
content_type='application/json'
)
assert response.status_code == 404
data = response.get_json()
assert data['status'] == 'error'
assert 'not found' in data['message'].lower()
def test_test_decode_invalid_sample_rate(self, client):
"""POST /weather-sat/test-decode with invalid sample rate."""
with patch('routes.weather_sat.is_weather_sat_available', return_value=True), \
patch('routes.weather_sat.get_weather_sat_decoder') as mock_get:
mock_decoder = MagicMock()
mock_decoder.is_running = False
mock_get.return_value = mock_decoder
payload = {
'satellite': 'NOAA-18',
'input_file': 'data/test.wav',
'sample_rate': 100, # Too low
}
response = client.post(
'/weather-sat/test-decode',
data=json.dumps(payload),
content_type='application/json'
)
assert response.status_code == 400
data = response.get_json()
assert data['status'] == 'error'
assert 'sample_rate' in data['message']
def test_stop_capture(self, client):
"""POST /weather-sat/stop stops capture."""
with patch('routes.weather_sat.get_weather_sat_decoder') as mock_get:
mock_decoder = MagicMock()
mock_decoder.device_index = 0
mock_get.return_value = mock_decoder
response = client.post('/weather-sat/stop')
assert response.status_code == 200
data = response.get_json()
assert data['status'] == 'stopped'
mock_decoder.stop.assert_called_once()
def test_list_images_empty(self, client):
"""GET /weather-sat/images with no images."""
with patch('routes.weather_sat.get_weather_sat_decoder') as mock_get:
mock_decoder = MagicMock()
mock_decoder.get_images.return_value = []
mock_get.return_value = mock_decoder
response = client.get('/weather-sat/images')
assert response.status_code == 200
data = response.get_json()
assert data['status'] == 'ok'
assert data['images'] == []
assert data['count'] == 0
def test_list_images_with_data(self, client):
"""GET /weather-sat/images with images."""
with patch('routes.weather_sat.get_weather_sat_decoder') as mock_get:
mock_decoder = MagicMock()
image = WeatherSatImage(
filename='NOAA-18_test.png',
path=Path('/tmp/test.png'),
satellite='NOAA-18',
mode='APT',
timestamp=datetime(2024, 1, 1, 12, 0, 0, tzinfo=timezone.utc),
frequency=137.9125,
size_bytes=12345,
product='RGB Composite',
)
mock_decoder.get_images.return_value = [image]
mock_get.return_value = mock_decoder
response = client.get('/weather-sat/images')
assert response.status_code == 200
data = response.get_json()
assert data['status'] == 'ok'
assert data['count'] == 1
assert data['images'][0]['filename'] == 'NOAA-18_test.png'
assert data['images'][0]['satellite'] == 'NOAA-18'
def test_list_images_with_filter(self, client):
"""GET /weather-sat/images with satellite filter."""
with patch('routes.weather_sat.get_weather_sat_decoder') as mock_get:
mock_decoder = MagicMock()
image1 = WeatherSatImage(
filename='NOAA-18_test.png',
path=Path('/tmp/test1.png'),
satellite='NOAA-18',
mode='APT',
timestamp=datetime.now(timezone.utc),
frequency=137.9125,
)
image2 = WeatherSatImage(
filename='NOAA-19_test.png',
path=Path('/tmp/test2.png'),
satellite='NOAA-19',
mode='APT',
timestamp=datetime.now(timezone.utc),
frequency=137.100,
)
mock_decoder.get_images.return_value = [image1, image2]
mock_get.return_value = mock_decoder
response = client.get('/weather-sat/images?satellite=NOAA-18')
assert response.status_code == 200
data = response.get_json()
assert data['count'] == 1
assert data['images'][0]['satellite'] == 'NOAA-18'
def test_list_images_with_limit(self, client):
"""GET /weather-sat/images with limit."""
with patch('routes.weather_sat.get_weather_sat_decoder') as mock_get:
mock_decoder = MagicMock()
images = [
WeatherSatImage(
filename=f'test{i}.png',
path=Path(f'/tmp/test{i}.png'),
satellite='NOAA-18',
mode='APT',
timestamp=datetime.now(timezone.utc),
frequency=137.9125,
)
for i in range(10)
]
mock_decoder.get_images.return_value = images
mock_get.return_value = mock_decoder
response = client.get('/weather-sat/images?limit=5')
assert response.status_code == 200
data = response.get_json()
assert data['count'] == 5
def test_get_image_success(self, client):
"""GET /weather-sat/images/<filename> serves image."""
with patch('routes.weather_sat.get_weather_sat_decoder') as mock_get, \
patch('routes.weather_sat.send_file') as mock_send, \
patch('pathlib.Path.exists', return_value=True):
mock_decoder = MagicMock()
mock_decoder._output_dir = Path('/tmp')
mock_get.return_value = mock_decoder
mock_send.return_value = MagicMock()
response = client.get('/weather-sat/images/test_image.png')
mock_send.assert_called_once()
call_args = mock_send.call_args
assert call_args[1]['mimetype'] == 'image/png'
def test_get_image_invalid_filename(self, client):
"""GET /weather-sat/images/<filename> with invalid filename."""
with patch('routes.weather_sat.get_weather_sat_decoder') as mock_get:
mock_decoder = MagicMock()
mock_get.return_value = mock_decoder
response = client.get('/weather-sat/images/../../../etc/passwd')
assert response.status_code == 400
data = response.get_json()
assert data['status'] == 'error'
assert 'Invalid filename' in data['message']
def test_get_image_wrong_extension(self, client):
"""GET /weather-sat/images/<filename> with wrong extension."""
with patch('routes.weather_sat.get_weather_sat_decoder') as mock_get:
mock_decoder = MagicMock()
mock_get.return_value = mock_decoder
response = client.get('/weather-sat/images/test.txt')
assert response.status_code == 400
data = response.get_json()
assert 'PNG/JPG' in data['message']
def test_get_image_not_found(self, client):
"""GET /weather-sat/images/<filename> for non-existent image."""
with patch('routes.weather_sat.get_weather_sat_decoder') as mock_get, \
patch('pathlib.Path.exists', return_value=False):
mock_decoder = MagicMock()
mock_decoder._output_dir = Path('/tmp')
mock_get.return_value = mock_decoder
response = client.get('/weather-sat/images/missing.png')
assert response.status_code == 404
def test_delete_image_success(self, client):
"""DELETE /weather-sat/images/<filename> deletes image."""
with patch('routes.weather_sat.get_weather_sat_decoder') as mock_get:
mock_decoder = MagicMock()
mock_decoder.delete_image.return_value = True
mock_get.return_value = mock_decoder
response = client.delete('/weather-sat/images/test.png')
assert response.status_code == 200
data = response.get_json()
assert data['status'] == 'deleted'
assert data['filename'] == 'test.png'
def test_delete_image_not_found(self, client):
"""DELETE /weather-sat/images/<filename> for non-existent image."""
with patch('routes.weather_sat.get_weather_sat_decoder') as mock_get:
mock_decoder = MagicMock()
mock_decoder.delete_image.return_value = False
mock_get.return_value = mock_decoder
response = client.delete('/weather-sat/images/missing.png')
assert response.status_code == 404
def test_delete_all_images(self, client):
"""DELETE /weather-sat/images deletes all images."""
with patch('routes.weather_sat.get_weather_sat_decoder') as mock_get:
mock_decoder = MagicMock()
mock_decoder.delete_all_images.return_value = 5
mock_get.return_value = mock_decoder
response = client.delete('/weather-sat/images')
assert response.status_code == 200
data = response.get_json()
assert data['status'] == 'ok'
assert data['deleted'] == 5
def test_stream_progress(self, client):
"""GET /weather-sat/stream returns SSE stream."""
response = client.get('/weather-sat/stream')
assert response.status_code == 200
assert response.mimetype == 'text/event-stream'
assert response.headers['Cache-Control'] == 'no-cache'
def test_get_passes_missing_params(self, client):
"""GET /weather-sat/passes without required params."""
response = client.get('/weather-sat/passes')
assert response.status_code == 400
data = response.get_json()
assert data['status'] == 'error'
assert 'latitude and longitude' in data['message']
def test_get_passes_invalid_coords(self, client):
"""GET /weather-sat/passes with invalid coordinates."""
response = client.get('/weather-sat/passes?latitude=999&longitude=0')
assert response.status_code == 400
data = response.get_json()
assert data['status'] == 'error'
def test_get_passes_success(self, client):
"""GET /weather-sat/passes successfully predicts passes."""
with patch('routes.weather_sat.predict_passes') as mock_predict:
mock_predict.return_value = [
{
'id': 'NOAA-18_202401011200',
'satellite': 'NOAA-18',
'name': 'NOAA 18',
'frequency': 137.9125,
'mode': 'APT',
'startTime': '2024-01-01 12:00 UTC',
'startTimeISO': '2024-01-01T12:00:00+00:00',
'endTimeISO': '2024-01-01T12:15:00+00:00',
'maxEl': 45.0,
'maxElAz': 180.0,
'riseAz': 160.0,
'setAz': 200.0,
'duration': 15.0,
'quality': 'good',
}
]
response = client.get('/weather-sat/passes?latitude=51.5&longitude=-0.1')
assert response.status_code == 200
data = response.get_json()
assert data['status'] == 'ok'
assert data['count'] == 1
assert data['passes'][0]['satellite'] == 'NOAA-18'
def test_get_passes_with_options(self, client):
"""GET /weather-sat/passes with trajectory and ground track."""
with patch('routes.weather_sat.predict_passes') as mock_predict:
mock_predict.return_value = []
response = client.get(
'/weather-sat/passes?latitude=51.5&longitude=-0.1&'
'hours=48&min_elevation=20&trajectory=true&ground_track=true'
)
assert response.status_code == 200
mock_predict.assert_called_once()
call_kwargs = mock_predict.call_args[1]
assert call_kwargs['lat'] == 51.5
assert call_kwargs['lon'] == -0.1
assert call_kwargs['hours'] == 48
assert call_kwargs['min_elevation'] == 20.0
assert call_kwargs['include_trajectory'] is True
assert call_kwargs['include_ground_track'] is True
def test_get_passes_import_error(self, client):
"""GET /weather-sat/passes when skyfield not installed."""
with patch('routes.weather_sat.predict_passes', side_effect=ImportError):
response = client.get('/weather-sat/passes?latitude=51.5&longitude=-0.1')
assert response.status_code == 503
data = response.get_json()
assert data['status'] == 'error'
assert 'skyfield' in data['message']
def test_get_passes_prediction_error(self, client):
"""GET /weather-sat/passes when prediction fails."""
with patch('routes.weather_sat.predict_passes', side_effect=Exception('TLE error')):
response = client.get('/weather-sat/passes?latitude=51.5&longitude=-0.1')
assert response.status_code == 500
data = response.get_json()
assert data['status'] == 'error'
class TestWeatherSatScheduler:
"""Tests for weather satellite scheduler endpoints."""
def test_enable_schedule_success(self, client):
"""POST /weather-sat/schedule/enable enables scheduler."""
with patch('routes.weather_sat.get_weather_sat_scheduler') as mock_get:
mock_scheduler = MagicMock()
mock_scheduler.enable.return_value = {
'enabled': True,
'observer': {'latitude': 51.5, 'longitude': -0.1},
'device': 0,
'gain': 40.0,
'bias_t': False,
'min_elevation': 15.0,
'scheduled_count': 3,
'total_passes': 3,
}
mock_get.return_value = mock_scheduler
payload = {
'latitude': 51.5,
'longitude': -0.1,
'min_elevation': 15,
'device': 0,
'gain': 40.0,
'bias_t': False,
}
response = client.post(
'/weather-sat/schedule/enable',
data=json.dumps(payload),
content_type='application/json'
)
assert response.status_code == 200
data = response.get_json()
assert data['status'] == 'ok'
assert data['enabled'] is True
def test_enable_schedule_missing_coords(self, client):
"""POST /weather-sat/schedule/enable without coordinates."""
payload = {'device': 0}
response = client.post(
'/weather-sat/schedule/enable',
data=json.dumps(payload),
content_type='application/json'
)
assert response.status_code == 400
data = response.get_json()
assert data['status'] == 'error'
assert 'latitude and longitude' in data['message']
def test_enable_schedule_invalid_coords(self, client):
"""POST /weather-sat/schedule/enable with invalid coordinates."""
payload = {'latitude': 999, 'longitude': 0}
response = client.post(
'/weather-sat/schedule/enable',
data=json.dumps(payload),
content_type='application/json'
)
assert response.status_code == 400
data = response.get_json()
assert data['status'] == 'error'
def test_disable_schedule(self, client):
"""POST /weather-sat/schedule/disable disables scheduler."""
with patch('routes.weather_sat.get_weather_sat_scheduler') as mock_get:
mock_scheduler = MagicMock()
mock_scheduler.disable.return_value = {'status': 'disabled'}
mock_get.return_value = mock_scheduler
response = client.post('/weather-sat/schedule/disable')
assert response.status_code == 200
data = response.get_json()
assert data['status'] == 'disabled'
def test_schedule_status(self, client):
"""GET /weather-sat/schedule/status returns scheduler status."""
with patch('routes.weather_sat.get_weather_sat_scheduler') as mock_get:
mock_scheduler = MagicMock()
mock_scheduler.get_status.return_value = {
'enabled': False,
'observer': {'latitude': 0, 'longitude': 0},
'device': 0,
'gain': 40.0,
'bias_t': False,
'min_elevation': 15.0,
'scheduled_count': 0,
'total_passes': 0,
}
mock_get.return_value = mock_scheduler
response = client.get('/weather-sat/schedule/status')
assert response.status_code == 200
data = response.get_json()
assert 'enabled' in data
def test_schedule_passes(self, client):
"""GET /weather-sat/schedule/passes lists scheduled passes."""
with patch('routes.weather_sat.get_weather_sat_scheduler') as mock_get:
mock_scheduler = MagicMock()
mock_scheduler.get_passes.return_value = [
{
'id': 'NOAA-18_202401011200',
'satellite': 'NOAA-18',
'status': 'scheduled',
}
]
mock_get.return_value = mock_scheduler
response = client.get('/weather-sat/schedule/passes')
assert response.status_code == 200
data = response.get_json()
assert data['status'] == 'ok'
assert data['count'] == 1
def test_skip_pass_success(self, client):
"""POST /weather-sat/schedule/skip/<id> skips a pass."""
with patch('routes.weather_sat.get_weather_sat_scheduler') as mock_get:
mock_scheduler = MagicMock()
mock_scheduler.skip_pass.return_value = True
mock_get.return_value = mock_scheduler
response = client.post('/weather-sat/schedule/skip/NOAA-18_202401011200')
assert response.status_code == 200
data = response.get_json()
assert data['status'] == 'skipped'
assert data['pass_id'] == 'NOAA-18_202401011200'
def test_skip_pass_not_found(self, client):
"""POST /weather-sat/schedule/skip/<id> for non-existent pass."""
with patch('routes.weather_sat.get_weather_sat_scheduler') as mock_get:
mock_scheduler = MagicMock()
mock_scheduler.skip_pass.return_value = False
mock_get.return_value = mock_scheduler
response = client.post('/weather-sat/schedule/skip/nonexistent')
assert response.status_code == 404
def test_skip_pass_invalid_id(self, client):
"""POST /weather-sat/schedule/skip/<id> with invalid ID."""
response = client.post('/weather-sat/schedule/skip/../../../etc/passwd')
assert response.status_code == 400
data = response.get_json()
assert data['status'] == 'error'
assert 'Invalid pass ID' in data['message']

779
tests/test_weather_sat_scheduler.py Normal file
View File

@@ -0,0 +1,779 @@
"""Tests for weather satellite auto-scheduler.
Covers WeatherSatScheduler class, pass scheduling, timer management,
and automatic capture execution.
"""
from __future__ import annotations
import threading
import time
from datetime import datetime, timezone, timedelta
from unittest.mock import patch, MagicMock, call
import pytest
from utils.weather_sat_scheduler import (
WeatherSatScheduler,
ScheduledPass,
get_weather_sat_scheduler,
)
class TestScheduledPass:
"""Tests for ScheduledPass class."""
def test_scheduled_pass_initialization(self):
"""ScheduledPass should initialize from pass data."""
pass_data = {
'id': 'NOAA-18_202401011200',
'satellite': 'NOAA-18',
'name': 'NOAA 18',
'frequency': 137.9125,
'mode': 'APT',
'startTimeISO': '2024-01-01T12:00:00+00:00',
'endTimeISO': '2024-01-01T12:15:00+00:00',
'maxEl': 45.0,
'duration': 15.0,
'quality': 'good',
}
sp = ScheduledPass(pass_data)
assert sp.id == 'NOAA-18_202401011200'
assert sp.satellite == 'NOAA-18'
assert sp.name == 'NOAA 18'
assert sp.frequency == 137.9125
assert sp.mode == 'APT'
assert sp.max_el == 45.0
assert sp.duration == 15.0
assert sp.quality == 'good'
assert sp.status == 'scheduled'
assert sp.skipped is False
def test_scheduled_pass_start_dt(self):
"""ScheduledPass.start_dt should parse ISO datetime."""
pass_data = {
'satellite': 'NOAA-18',
'name': 'NOAA 18',
'frequency': 137.9125,
'mode': 'APT',
'startTimeISO': '2024-01-01T12:00:00+00:00',
'endTimeISO': '2024-01-01T12:15:00+00:00',
'maxEl': 45.0,
'duration': 15.0,
'quality': 'good',
}
sp = ScheduledPass(pass_data)
assert sp.start_dt.year == 2024
assert sp.start_dt.month == 1
assert sp.start_dt.day == 1
assert sp.start_dt.hour == 12
assert sp.start_dt.tzinfo == timezone.utc
def test_scheduled_pass_end_dt(self):
"""ScheduledPass.end_dt should parse ISO datetime."""
pass_data = {
'satellite': 'NOAA-18',
'name': 'NOAA 18',
'frequency': 137.9125,
'mode': 'APT',
'startTimeISO': '2024-01-01T12:00:00+00:00',
'endTimeISO': '2024-01-01T12:15:00+00:00',
'maxEl': 45.0,
'duration': 15.0,
'quality': 'good',
}
sp = ScheduledPass(pass_data)
assert sp.end_dt.year == 2024
assert sp.end_dt.minute == 15
def test_scheduled_pass_to_dict(self):
"""ScheduledPass.to_dict() should serialize correctly."""
pass_data = {
'id': 'NOAA-18_202401011200',
'satellite': 'NOAA-18',
'name': 'NOAA 18',
'frequency': 137.9125,
'mode': 'APT',
'startTimeISO': '2024-01-01T12:00:00+00:00',
'endTimeISO': '2024-01-01T12:15:00+00:00',
'maxEl': 45.0,
'duration': 15.0,
'quality': 'good',
}
sp = ScheduledPass(pass_data)
sp.status = 'complete'
data = sp.to_dict()
assert data['id'] == 'NOAA-18_202401011200'
assert data['satellite'] == 'NOAA-18'
assert data['status'] == 'complete'
assert data['skipped'] is False
class TestWeatherSatScheduler:
"""Tests for WeatherSatScheduler class."""
def test_scheduler_initialization(self):
"""Scheduler should initialize with defaults."""
scheduler = WeatherSatScheduler()
assert scheduler.enabled is False
assert scheduler._lat == 0.0
assert scheduler._lon == 0.0
assert scheduler._min_elevation == 15.0
assert scheduler._device == 0
assert scheduler._gain == 40.0
assert scheduler._bias_t is False
assert scheduler._passes == []
def test_set_callbacks(self):
"""Scheduler should accept callbacks."""
scheduler = WeatherSatScheduler()
progress_cb = MagicMock()
event_cb = MagicMock()
scheduler.set_callbacks(progress_cb, event_cb)
assert scheduler._progress_callback == progress_cb
assert scheduler._event_callback == event_cb
@patch('utils.weather_sat_scheduler.WeatherSatScheduler._refresh_passes')
def test_enable(self, mock_refresh):
"""enable() should start scheduler."""
scheduler = WeatherSatScheduler()
result = scheduler.enable(
lat=51.5,
lon=-0.1,
min_elevation=20.0,
device=1,
gain=35.0,
bias_t=True,
)
assert scheduler._enabled is True
assert scheduler._lat == 51.5
assert scheduler._lon == -0.1
assert scheduler._min_elevation == 20.0
assert scheduler._device == 1
assert scheduler._gain == 35.0
assert scheduler._bias_t is True
mock_refresh.assert_called_once()
assert 'enabled' in result
def test_disable(self):
"""disable() should stop scheduler and cancel timers."""
scheduler = WeatherSatScheduler()
scheduler._enabled = True
# Add mock timer
mock_timer = MagicMock()
scheduler._refresh_timer = mock_timer
# Add pass with timer
pass_data = {
'satellite': 'NOAA-18',
'name': 'NOAA 18',
'frequency': 137.9125,
'mode': 'APT',
'startTimeISO': '2024-01-01T12:00:00+00:00',
'endTimeISO': '2024-01-01T12:15:00+00:00',
'maxEl': 45.0,
'duration': 15.0,
'quality': 'good',
}
sp = ScheduledPass(pass_data)
sp._timer = MagicMock()
sp._stop_timer = MagicMock()
scheduler._passes = [sp]
result = scheduler.disable()
assert scheduler._enabled is False
assert scheduler._passes == []
mock_timer.cancel.assert_called_once()
sp._timer.cancel.assert_called_once()
sp._stop_timer.cancel.assert_called_once()
assert result['status'] == 'disabled'
def test_skip_pass_success(self):
"""skip_pass() should skip a scheduled pass."""
scheduler = WeatherSatScheduler()
event_cb = MagicMock()
scheduler.set_callbacks(MagicMock(), event_cb)
pass_data = {
'id': 'NOAA-18_202401011200',
'satellite': 'NOAA-18',
'name': 'NOAA 18',
'frequency': 137.9125,
'mode': 'APT',
'startTimeISO': '2024-01-01T12:00:00+00:00',
'endTimeISO': '2024-01-01T12:15:00+00:00',
'maxEl': 45.0,
'duration': 15.0,
'quality': 'good',
}
sp = ScheduledPass(pass_data)
sp._timer = MagicMock()
scheduler._passes = [sp]
result = scheduler.skip_pass('NOAA-18_202401011200')
assert result is True
assert sp.status == 'skipped'
assert sp.skipped is True
sp._timer.cancel.assert_called_once()
event_cb.assert_called_once()
def test_skip_pass_not_found(self):
"""skip_pass() should return False for non-existent pass."""
scheduler = WeatherSatScheduler()
result = scheduler.skip_pass('NONEXISTENT')
assert result is False
def test_skip_pass_already_complete(self):
"""skip_pass() should not skip already complete passes."""
scheduler = WeatherSatScheduler()
pass_data = {
'id': 'NOAA-18_202401011200',
'satellite': 'NOAA-18',
'name': 'NOAA 18',
'frequency': 137.9125,
'mode': 'APT',
'startTimeISO': '2024-01-01T12:00:00+00:00',
'endTimeISO': '2024-01-01T12:15:00+00:00',
'maxEl': 45.0,
'duration': 15.0,
'quality': 'good',
}
sp = ScheduledPass(pass_data)
sp.status = 'complete'
scheduler._passes = [sp]
result = scheduler.skip_pass('NOAA-18_202401011200')
assert result is False
assert sp.status == 'complete'
def test_get_status(self):
"""get_status() should return scheduler state."""
scheduler = WeatherSatScheduler()
scheduler._enabled = True
scheduler._lat = 51.5
scheduler._lon = -0.1
scheduler._device = 0
scheduler._gain = 40.0
scheduler._bias_t = False
scheduler._min_elevation = 15.0
pass_data = {
'satellite': 'NOAA-18',
'name': 'NOAA 18',
'frequency': 137.9125,
'mode': 'APT',
'startTimeISO': '2024-01-01T12:00:00+00:00',
'endTimeISO': '2024-01-01T12:15:00+00:00',
'maxEl': 45.0,
'duration': 15.0,
'quality': 'good',
}
sp = ScheduledPass(pass_data)
scheduler._passes = [sp]
status = scheduler.get_status()
assert status['enabled'] is True
assert status['observer']['latitude'] == 51.5
assert status['observer']['longitude'] == -0.1
assert status['device'] == 0
assert status['gain'] == 40.0
assert status['bias_t'] is False
assert status['min_elevation'] == 15.0
assert status['scheduled_count'] == 1
assert status['total_passes'] == 1
def test_get_passes(self):
"""get_passes() should return list of scheduled passes."""
scheduler = WeatherSatScheduler()
pass_data = {
'id': 'NOAA-18_202401011200',
'satellite': 'NOAA-18',
'name': 'NOAA 18',
'frequency': 137.9125,
'mode': 'APT',
'startTimeISO': '2024-01-01T12:00:00+00:00',
'endTimeISO': '2024-01-01T12:15:00+00:00',
'maxEl': 45.0,
'duration': 15.0,
'quality': 'good',
}
sp = ScheduledPass(pass_data)
scheduler._passes = [sp]
passes = scheduler.get_passes()
assert len(passes) == 1
assert passes[0]['id'] == 'NOAA-18_202401011200'
@patch('utils.weather_sat_scheduler.predict_passes')
@patch('threading.Timer')
def test_refresh_passes(self, mock_timer, mock_predict):
"""_refresh_passes() should schedule future passes."""
now = datetime.now(timezone.utc)
future_pass = {
'id': 'NOAA-18_202401011200',
'satellite': 'NOAA-18',
'name': 'NOAA 18',
'frequency': 137.9125,
'mode': 'APT',
'startTimeISO': (now + timedelta(hours=2)).isoformat(),
'endTimeISO': (now + timedelta(hours=2, minutes=15)).isoformat(),
'maxEl': 45.0,
'duration': 15.0,
'quality': 'good',
}
mock_predict.return_value = [future_pass]
mock_timer_instance = MagicMock()
mock_timer.return_value = mock_timer_instance
scheduler = WeatherSatScheduler()
scheduler._enabled = True
scheduler._lat = 51.5
scheduler._lon = -0.1
scheduler._refresh_passes()
mock_predict.assert_called_once()
assert len(scheduler._passes) == 1
assert scheduler._passes[0].satellite == 'NOAA-18'
mock_timer_instance.start.assert_called()
@patch('utils.weather_sat_scheduler.predict_passes')
def test_refresh_passes_skip_past(self, mock_predict):
"""_refresh_passes() should skip passes that already started."""
now = datetime.now(timezone.utc)
past_pass = {
'id': 'NOAA-18_202401011200',
'satellite': 'NOAA-18',
'name': 'NOAA 18',
'frequency': 137.9125,
'mode': 'APT',
'startTimeISO': (now - timedelta(hours=1)).isoformat(),
'endTimeISO': (now - timedelta(hours=1) + timedelta(minutes=15)).isoformat(),
'maxEl': 45.0,
'duration': 15.0,
'quality': 'good',
}
mock_predict.return_value = [past_pass]
scheduler = WeatherSatScheduler()
scheduler._enabled = True
scheduler._lat = 51.5
scheduler._lon = -0.1
scheduler._refresh_passes()
# Should not schedule past passes
assert len(scheduler._passes) == 0
@patch('utils.weather_sat_scheduler.predict_passes')
def test_refresh_passes_disabled(self, mock_predict):
"""_refresh_passes() should do nothing when disabled."""
scheduler = WeatherSatScheduler()
scheduler._enabled = False
scheduler._refresh_passes()
mock_predict.assert_not_called()
@patch('utils.weather_sat_scheduler.predict_passes')
def test_refresh_passes_error_handling(self, mock_predict):
"""_refresh_passes() should handle prediction errors."""
mock_predict.side_effect = Exception('TLE error')
scheduler = WeatherSatScheduler()
scheduler._enabled = True
scheduler._lat = 51.5
scheduler._lon = -0.1
# Should not raise
scheduler._refresh_passes()
assert len(scheduler._passes) == 0
@patch('utils.weather_sat_scheduler.get_weather_sat_decoder')
def test_execute_capture_disabled(self, mock_get):
"""_execute_capture() should do nothing when disabled."""
scheduler = WeatherSatScheduler()
scheduler._enabled = False
pass_data = {
'satellite': 'NOAA-18',
'name': 'NOAA 18',
'frequency': 137.9125,
'mode': 'APT',
'startTimeISO': '2024-01-01T12:00:00+00:00',
'endTimeISO': '2024-01-01T12:15:00+00:00',
'maxEl': 45.0,
'duration': 15.0,
'quality': 'good',
}
sp = ScheduledPass(pass_data)
scheduler._execute_capture(sp)
mock_get.assert_not_called()
@patch('utils.weather_sat_scheduler.get_weather_sat_decoder')
def test_execute_capture_skipped(self, mock_get):
"""_execute_capture() should do nothing for skipped passes."""
scheduler = WeatherSatScheduler()
scheduler._enabled = True
pass_data = {
'satellite': 'NOAA-18',
'name': 'NOAA 18',
'frequency': 137.9125,
'mode': 'APT',
'startTimeISO': '2024-01-01T12:00:00+00:00',
'endTimeISO': '2024-01-01T12:15:00+00:00',
'maxEl': 45.0,
'duration': 15.0,
'quality': 'good',
}
sp = ScheduledPass(pass_data)
sp.skipped = True
scheduler._execute_capture(sp)
mock_get.assert_not_called()
@patch('utils.weather_sat_scheduler.get_weather_sat_decoder')
def test_execute_capture_decoder_busy(self, mock_get):
"""_execute_capture() should skip when decoder is busy."""
scheduler = WeatherSatScheduler()
scheduler._enabled = True
event_cb = MagicMock()
scheduler.set_callbacks(MagicMock(), event_cb)
mock_decoder = MagicMock()
mock_decoder.is_running = True
mock_get.return_value = mock_decoder
pass_data = {
'satellite': 'NOAA-18',
'name': 'NOAA 18',
'frequency': 137.9125,
'mode': 'APT',
'startTimeISO': '2024-01-01T12:00:00+00:00',
'endTimeISO': '2024-01-01T12:15:00+00:00',
'maxEl': 45.0,
'duration': 15.0,
'quality': 'good',
}
sp = ScheduledPass(pass_data)
scheduler._execute_capture(sp)
assert sp.status == 'skipped'
assert sp.skipped is True
event_cb.assert_called_once()
event_data = event_cb.call_args[0][0]
assert event_data['type'] == 'schedule_capture_skipped'
assert event_data['reason'] == 'sdr_busy'
@patch('utils.weather_sat_scheduler.get_weather_sat_decoder')
@patch('threading.Timer')
def test_execute_capture_success(self, mock_timer, mock_get):
"""_execute_capture() should start capture."""
scheduler = WeatherSatScheduler()
scheduler._enabled = True
scheduler._device = 0
scheduler._gain = 40.0
scheduler._bias_t = False
progress_cb = MagicMock()
event_cb = MagicMock()
scheduler.set_callbacks(progress_cb, event_cb)
mock_decoder = MagicMock()
mock_decoder.is_running = False
mock_decoder.start.return_value = True
mock_get.return_value = mock_decoder
mock_timer_instance = MagicMock()
mock_timer.return_value = mock_timer_instance
now = datetime.now(timezone.utc)
pass_data = {
'satellite': 'NOAA-18',
'name': 'NOAA 18',
'frequency': 137.9125,
'mode': 'APT',
'startTimeISO': (now + timedelta(seconds=10)).isoformat(),
'endTimeISO': (now + timedelta(minutes=15)).isoformat(),
'maxEl': 45.0,
'duration': 15.0,
'quality': 'good',
}
sp = ScheduledPass(pass_data)
scheduler._execute_capture(sp)
assert sp.status == 'capturing'
mock_decoder.set_callback.assert_called_once_with(progress_cb)
mock_decoder.start.assert_called_once_with(
satellite='NOAA-18',
device_index=0,
gain=40.0,
bias_t=False,
)
event_cb.assert_called_once()
event_data = event_cb.call_args[0][0]
assert event_data['type'] == 'schedule_capture_start'
@patch('utils.weather_sat_scheduler.get_weather_sat_decoder')
def test_execute_capture_start_failed(self, mock_get):
"""_execute_capture() should handle start failure."""
scheduler = WeatherSatScheduler()
scheduler._enabled = True
event_cb = MagicMock()
scheduler.set_callbacks(MagicMock(), event_cb)
mock_decoder = MagicMock()
mock_decoder.is_running = False
mock_decoder.start.return_value = False
mock_get.return_value = mock_decoder
pass_data = {
'satellite': 'NOAA-18',
'name': 'NOAA 18',
'frequency': 137.9125,
'mode': 'APT',
'startTimeISO': '2024-01-01T12:00:00+00:00',
'endTimeISO': '2024-01-01T12:15:00+00:00',
'maxEl': 45.0,
'duration': 15.0,
'quality': 'good',
}
sp = ScheduledPass(pass_data)
scheduler._execute_capture(sp)
assert sp.status == 'skipped'
event_cb.assert_called_once()
event_data = event_cb.call_args[0][0]
assert event_data['reason'] == 'start_failed'
@patch('utils.weather_sat_scheduler.get_weather_sat_decoder')
def test_stop_capture(self, mock_get):
"""_stop_capture() should stop decoder."""
scheduler = WeatherSatScheduler()
mock_decoder = MagicMock()
mock_decoder.is_running = True
mock_get.return_value = mock_decoder
pass_data = {
'satellite': 'NOAA-18',
'name': 'NOAA 18',
'frequency': 137.9125,
'mode': 'APT',
'startTimeISO': '2024-01-01T12:00:00+00:00',
'endTimeISO': '2024-01-01T12:15:00+00:00',
'maxEl': 45.0,
'duration': 15.0,
'quality': 'good',
}
sp = ScheduledPass(pass_data)
scheduler._stop_capture(sp)
mock_decoder.stop.assert_called_once()
def test_on_capture_complete(self):
"""_on_capture_complete() should mark pass complete and emit event."""
scheduler = WeatherSatScheduler()
event_cb = MagicMock()
scheduler.set_callbacks(MagicMock(), event_cb)
release_fn = MagicMock()
pass_data = {
'satellite': 'NOAA-18',
'name': 'NOAA 18',
'frequency': 137.9125,
'mode': 'APT',
'startTimeISO': '2024-01-01T12:00:00+00:00',
'endTimeISO': '2024-01-01T12:15:00+00:00',
'maxEl': 45.0,
'duration': 15.0,
'quality': 'good',
}
sp = ScheduledPass(pass_data)
scheduler._on_capture_complete(sp, release_fn)
assert sp.status == 'complete'
release_fn.assert_called_once()
event_cb.assert_called_once()
event_data = event_cb.call_args[0][0]
assert event_data['type'] == 'schedule_capture_complete'
def test_emit_event(self):
"""_emit_event() should call event callback."""
scheduler = WeatherSatScheduler()
event_cb = MagicMock()
scheduler.set_callbacks(MagicMock(), event_cb)
event = {'type': 'test_event', 'data': 'test'}
scheduler._emit_event(event)
event_cb.assert_called_once_with(event)
def test_emit_event_no_callback(self):
"""_emit_event() should handle missing callback."""
scheduler = WeatherSatScheduler()
event = {'type': 'test_event'}
scheduler._emit_event(event) # Should not raise
def test_emit_event_callback_exception(self):
"""_emit_event() should handle callback exceptions."""
scheduler = WeatherSatScheduler()
event_cb = MagicMock(side_effect=Exception('Callback error'))
scheduler.set_callbacks(MagicMock(), event_cb)
event = {'type': 'test_event'}
scheduler._emit_event(event) # Should not raise
class TestGlobalScheduler:
"""Tests for global scheduler singleton."""
def test_get_weather_sat_scheduler_singleton(self):
"""get_weather_sat_scheduler() should return singleton."""
import utils.weather_sat_scheduler as mod
old = mod._scheduler
mod._scheduler = None
try:
scheduler1 = get_weather_sat_scheduler()
scheduler2 = get_weather_sat_scheduler()
assert scheduler1 is scheduler2
finally:
mod._scheduler = old
def test_get_weather_sat_scheduler_thread_safe(self):
"""get_weather_sat_scheduler() should be thread-safe."""
import utils.weather_sat_scheduler as mod
old = mod._scheduler
mod._scheduler = None
schedulers = []
def create_scheduler():
schedulers.append(get_weather_sat_scheduler())
try:
threads = [threading.Thread(target=create_scheduler) for _ in range(10)]
for t in threads:
t.start()
for t in threads:
t.join()
# All should be the same instance
assert all(s is schedulers[0] for s in schedulers)
finally:
mod._scheduler = old
class TestSchedulerConfiguration:
"""Tests for scheduler configuration constants."""
def test_config_constants(self):
"""Scheduler should have configuration constants."""
from utils.weather_sat_scheduler import (
WEATHER_SAT_SCHEDULE_REFRESH_MINUTES,
WEATHER_SAT_CAPTURE_BUFFER_SECONDS,
)
assert isinstance(WEATHER_SAT_SCHEDULE_REFRESH_MINUTES, int)
assert isinstance(WEATHER_SAT_CAPTURE_BUFFER_SECONDS, int)
assert WEATHER_SAT_SCHEDULE_REFRESH_MINUTES > 0
assert WEATHER_SAT_CAPTURE_BUFFER_SECONDS >= 0
class TestSchedulerIntegration:
"""Integration tests for scheduler."""
@patch('utils.weather_sat_scheduler.predict_passes')
@patch('utils.weather_sat_scheduler.get_weather_sat_decoder')
@patch('threading.Timer')
def test_full_scheduling_cycle(self, mock_timer, mock_get_decoder, mock_predict):
"""Test complete scheduling cycle from enable to execute."""
now = datetime.now(timezone.utc)
future_pass = {
'id': 'NOAA-18_202401011200',
'satellite': 'NOAA-18',
'name': 'NOAA 18',
'frequency': 137.9125,
'mode': 'APT',
'startTimeISO': (now + timedelta(hours=2)).isoformat(),
'endTimeISO': (now + timedelta(hours=2, minutes=15)).isoformat(),
'maxEl': 45.0,
'duration': 15.0,
'quality': 'good',
}
mock_predict.return_value = [future_pass]
mock_timer_instance = MagicMock()
mock_timer.return_value = mock_timer_instance
mock_decoder = MagicMock()
mock_decoder.is_running = False
mock_decoder.start.return_value = True
mock_get_decoder.return_value = mock_decoder
scheduler = WeatherSatScheduler()
progress_cb = MagicMock()
event_cb = MagicMock()
scheduler.set_callbacks(progress_cb, event_cb)
# Enable scheduler
result = scheduler.enable(lat=51.5, lon=-0.1)
assert result['enabled'] is True
assert len(scheduler._passes) == 1
assert scheduler._passes[0].satellite == 'NOAA-18'
# Simulate timer firing (capture start)
scheduler._execute_capture(scheduler._passes[0])
assert scheduler._passes[0].status == 'capturing'
mock_decoder.start.assert_called_once()
# Simulate completion
release_fn = MagicMock()
scheduler._on_capture_complete(scheduler._passes[0], release_fn)
assert scheduler._passes[0].status == 'complete'
release_fn.assert_called_once()
# Disable scheduler
scheduler.disable()
assert scheduler.enabled is False
assert len(scheduler._passes) == 0