diff --git a/tests/test_weather_sat_routes.py b/tests/test_weather_sat_routes.py index 0106f7d..642d65d 100644 --- a/tests/test_weather_sat_routes.py +++ b/tests/test_weather_sat_routes.py @@ -11,9 +11,19 @@ from datetime import datetime, timezone from pathlib import Path from unittest.mock import MagicMock, patch +import pytest + from utils.weather_sat import WeatherSatImage +@pytest.fixture +def client(client): + """Authenticated client for weather-sat route tests.""" + with client.session_transaction() as sess: + sess['logged_in'] = True + return client + + class TestWeatherSatRoutes: """Tests for weather satellite routes.""" @@ -68,7 +78,8 @@ class TestWeatherSatRoutes: """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'): + patch('routes.weather_sat.queue.Queue'), \ + patch('app.claim_sdr_device', return_value=None): mock_decoder = MagicMock() mock_decoder.is_running = False @@ -96,12 +107,12 @@ class TestWeatherSatRoutes: 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, - ) + mock_decoder.start.assert_called_once() + call_kwargs = mock_decoder.start.call_args[1] + assert call_kwargs['satellite'] == 'NOAA-18' + assert call_kwargs['device_index'] == 0 + assert call_kwargs['gain'] == 40.0 + assert call_kwargs['bias_t'] is False def test_start_capture_no_satdump(self, client): """POST /weather-sat/start returns error when SatDump unavailable.""" @@ -290,7 +301,8 @@ class TestWeatherSatRoutes: 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: + patch('routes.weather_sat.get_weather_sat_decoder') as mock_get, \ + patch('app.claim_sdr_device', return_value=None): mock_decoder = MagicMock() mock_decoder.is_running = False @@ -409,7 +421,13 @@ class TestWeatherSatRoutes: 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: + 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 = MagicMock() + mock_path.is_relative_to.return_value = True + mock_resolve.return_value = mock_path mock_decoder = MagicMock() mock_decoder.is_running = False @@ -558,7 +576,7 @@ class TestWeatherSatRoutes: mock_decoder = MagicMock() mock_get.return_value = mock_decoder - response = client.get('/weather-sat/images/../../../etc/passwd') + response = client.get('/weather-sat/images/bad!file@name.png') assert response.status_code == 400 data = response.get_json() assert data['status'] == 'error' @@ -647,7 +665,7 @@ class TestWeatherSatRoutes: def test_get_passes_success(self, client): """GET /weather-sat/passes successfully predicts passes.""" - with patch('routes.weather_sat.predict_passes') as mock_predict: + with patch('utils.weather_sat_predict.predict_passes') as mock_predict: mock_predict.return_value = [ { 'id': 'NOAA-18_202401011200', @@ -676,7 +694,7 @@ class TestWeatherSatRoutes: 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: + with patch('utils.weather_sat_predict.predict_passes') as mock_predict: mock_predict.return_value = [] response = client.get( @@ -696,7 +714,7 @@ class TestWeatherSatRoutes: 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): + with patch('utils.weather_sat_predict.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() @@ -705,7 +723,7 @@ class TestWeatherSatRoutes: 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')): + with patch('utils.weather_sat_predict.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() @@ -717,7 +735,7 @@ class TestWeatherSatScheduler: 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: + with patch('utils.weather_sat_scheduler.get_weather_sat_scheduler') as mock_get: mock_scheduler = MagicMock() mock_scheduler.enable.return_value = { 'enabled': True, @@ -780,7 +798,7 @@ class TestWeatherSatScheduler: 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: + with patch('utils.weather_sat_scheduler.get_weather_sat_scheduler') as mock_get: mock_scheduler = MagicMock() mock_scheduler.disable.return_value = {'status': 'disabled'} mock_get.return_value = mock_scheduler @@ -792,7 +810,7 @@ class TestWeatherSatScheduler: 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: + with patch('utils.weather_sat_scheduler.get_weather_sat_scheduler') as mock_get: mock_scheduler = MagicMock() mock_scheduler.get_status.return_value = { 'enabled': False, @@ -813,7 +831,7 @@ class TestWeatherSatScheduler: 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: + with patch('utils.weather_sat_scheduler.get_weather_sat_scheduler') as mock_get: mock_scheduler = MagicMock() mock_scheduler.get_passes.return_value = [ { @@ -832,7 +850,7 @@ class TestWeatherSatScheduler: def test_skip_pass_success(self, client): """POST /weather-sat/schedule/skip/ skips a pass.""" - with patch('routes.weather_sat.get_weather_sat_scheduler') as mock_get: + with patch('utils.weather_sat_scheduler.get_weather_sat_scheduler') as mock_get: mock_scheduler = MagicMock() mock_scheduler.skip_pass.return_value = True mock_get.return_value = mock_scheduler @@ -845,7 +863,7 @@ class TestWeatherSatScheduler: def test_skip_pass_not_found(self, client): """POST /weather-sat/schedule/skip/ for non-existent pass.""" - with patch('routes.weather_sat.get_weather_sat_scheduler') as mock_get: + with patch('utils.weather_sat_scheduler.get_weather_sat_scheduler') as mock_get: mock_scheduler = MagicMock() mock_scheduler.skip_pass.return_value = False mock_get.return_value = mock_scheduler @@ -855,7 +873,7 @@ class TestWeatherSatScheduler: def test_skip_pass_invalid_id(self, client): """POST /weather-sat/schedule/skip/ with invalid ID.""" - response = client.post('/weather-sat/schedule/skip/../../../etc/passwd') + response = client.post('/weather-sat/schedule/skip/invalid!pass@id') assert response.status_code == 400 data = response.get_json() assert data['status'] == 'error' diff --git a/tests/test_weather_sat_scheduler.py b/tests/test_weather_sat_scheduler.py index 2fa8267..a82f27e 100644 --- a/tests/test_weather_sat_scheduler.py +++ b/tests/test_weather_sat_scheduler.py @@ -191,8 +191,10 @@ class TestWeatherSatScheduler: 'quality': 'good', } sp = ScheduledPass(pass_data) - sp._timer = MagicMock() - sp._stop_timer = MagicMock() + mock_pass_timer = MagicMock() + mock_stop_timer = MagicMock() + sp._timer = mock_pass_timer + sp._stop_timer = mock_stop_timer scheduler._passes = [sp] result = scheduler.disable() @@ -200,8 +202,8 @@ class TestWeatherSatScheduler: 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() + mock_pass_timer.cancel.assert_called_once() + mock_stop_timer.cancel.assert_called_once() assert result['status'] == 'disabled' def test_skip_pass_success(self): @@ -223,7 +225,8 @@ class TestWeatherSatScheduler: 'quality': 'good', } sp = ScheduledPass(pass_data) - sp._timer = MagicMock() + mock_pass_timer = MagicMock() + sp._timer = mock_pass_timer scheduler._passes = [sp] result = scheduler.skip_pass('NOAA-18_202401011200') @@ -231,7 +234,7 @@ class TestWeatherSatScheduler: assert result is True assert sp.status == 'skipped' assert sp.skipped is True - sp._timer.cancel.assert_called_once() + mock_pass_timer.cancel.assert_called_once() event_cb.assert_called_once() def test_skip_pass_not_found(self): @@ -531,9 +534,10 @@ class TestWeatherSatScheduler: assert event_data['type'] == 'schedule_capture_skipped' assert event_data['reason'] == 'sdr_busy' + @patch('app.claim_sdr_device', return_value=None) @patch('utils.weather_sat_scheduler.get_weather_sat_decoder') @patch('threading.Timer') - def test_execute_capture_success(self, mock_timer, mock_get): + def test_execute_capture_success(self, mock_timer, mock_get, mock_claim): """_execute_capture() should start capture.""" scheduler = WeatherSatScheduler() scheduler._enabled = True @@ -570,18 +574,18 @@ class TestWeatherSatScheduler: 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, - ) + call_kwargs = mock_decoder.start.call_args[1] + assert call_kwargs['satellite'] == 'NOAA-18' + assert call_kwargs['device_index'] == 0 + assert call_kwargs['gain'] == 40.0 + assert call_kwargs['bias_t'] is False event_cb.assert_called_once() event_data = event_cb.call_args[0][0] assert event_data['type'] == 'schedule_capture_start' + @patch('app.claim_sdr_device', return_value=None) @patch('utils.weather_sat_scheduler.get_weather_sat_decoder') - def test_execute_capture_start_failed(self, mock_get): + def test_execute_capture_start_failed(self, mock_get, mock_claim): """_execute_capture() should handle start failure.""" scheduler = WeatherSatScheduler() scheduler._enabled = True @@ -773,10 +777,11 @@ class TestUtcIsoParsing: class TestSchedulerIntegration: """Integration tests for scheduler.""" + @patch('app.claim_sdr_device', return_value=None) @patch('utils.weather_sat_predict.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): + def test_full_scheduling_cycle(self, mock_timer, mock_get_decoder, mock_predict, mock_claim): """Test complete scheduling cycle from enable to execute.""" now = datetime.now(timezone.utc) future_pass = {