Mirror/intercept
1
0
Fork 0
mirror of https://github.com/smittix/intercept.git synced 2026-07-08 09:38:12 -07:00
Code Issues Packages Projects Releases Wiki Activity

Apply pending weather-sat and wefax updates

Browse Source
This commit is contained in:
Smittix
2026-02-24 21:46:58 +00:00
parent b90f1def60
commit 99c4304308
10 changed files with 712 additions and 206 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_weather_sat_regressions.py
+120
View File
@@ -0,0 +1,120 @@
"""Targeted regression tests for recent weather-satellite hardening fixes."""
from __future__ import annotations
import re
from unittest.mock import MagicMock, patch
import pytest
from utils.weather_sat import WeatherSatDecoder
@pytest.fixture
def authed_client(client):
"""Return a logged-in test client for authenticated weather-sat routes."""
with client.session_transaction() as session:
session['logged_in'] = True
return client
class TestWeatherSatRouteReleaseGuards:
"""Regression tests for safe SDR release behavior in weather-sat routes."""
def test_stop_does_not_release_device_owned_by_other_mode(self, authed_client):
"""POST /weather-sat/stop should not release a foreign-owned SDR device."""
mock_decoder = MagicMock()
mock_decoder.device_index = 2
with patch('routes.weather_sat.get_weather_sat_decoder', return_value=mock_decoder), \
patch('app.get_sdr_device_status', return_value={2: 'wifi'}), \
patch('app.release_sdr_device') as mock_release:
response = authed_client.post('/weather-sat/stop')
assert response.status_code == 200
assert response.get_json()['status'] == 'stopped'
mock_decoder.stop.assert_called_once()
mock_release.assert_not_called()
def test_stop_releases_device_owned_by_weather_sat(self, authed_client):
"""POST /weather-sat/stop should release SDR when weather-sat owns it."""
mock_decoder = MagicMock()
mock_decoder.device_index = 2
with patch('routes.weather_sat.get_weather_sat_decoder', return_value=mock_decoder), \
patch('app.get_sdr_device_status', return_value={2: 'weather_sat'}), \
patch('app.release_sdr_device') as mock_release:
response = authed_client.post('/weather-sat/stop')
assert response.status_code == 200
assert response.get_json()['status'] == 'stopped'
mock_decoder.stop.assert_called_once()
mock_release.assert_called_once_with(2)
def test_stop_skips_release_for_offline_decode_index(self, authed_client):
"""POST /weather-sat/stop should not release when decoder index is -1."""
mock_decoder = MagicMock()
mock_decoder.device_index = -1
with patch('routes.weather_sat.get_weather_sat_decoder', return_value=mock_decoder), \
patch('app.release_sdr_device') as mock_release:
response = authed_client.post('/weather-sat/stop')
assert response.status_code == 200
assert response.get_json()['status'] == 'stopped'
mock_decoder.stop.assert_called_once()
mock_release.assert_not_called()
class TestWeatherSatDecoderRegressions:
"""Regression tests for decoder filename and offline-device handling."""
def test_scan_output_dir_preserves_extension_and_sanitizes_filename(self, tmp_path):
"""Copied image names should stay safe and preserve JPG/JPEG extensions."""
output_dir = tmp_path / 'weather_sat_out'
capture_dir = tmp_path / 'capture'
capture_dir.mkdir(parents=True)
source_image = capture_dir / 'channel 3 (raw).jpeg'
source_image.write_bytes(b'\xff\xd8\xff' + b'\x00' * 2048)
with patch('shutil.which', return_value='/usr/bin/satdump'):
decoder = WeatherSatDecoder(output_dir=output_dir)
decoder._capture_output_dir = capture_dir
decoder._current_satellite = 'METEOR-M2-4'
decoder._current_mode = 'LRPT'
decoder._current_frequency = 137.9
decoder._scan_output_dir(set())
assert len(decoder._images) == 1
image = decoder._images[0]
assert image.filename.endswith('.jpeg')
assert re.fullmatch(r'[A-Za-z0-9_.-]+', image.filename)
assert (output_dir / image.filename).is_file()
def test_start_from_file_keeps_device_index_unclaimed(self, tmp_path):
"""Offline file decode should not claim or persist an SDR device index."""
with patch('shutil.which', return_value='/usr/bin/satdump'), \
patch('pathlib.Path.is_file', return_value=True), \
patch('pathlib.Path.resolve') as mock_resolve, \
patch.object(WeatherSatDecoder, '_start_satdump_offline') as mock_start:
resolved = MagicMock()
resolved.is_relative_to.return_value = True
mock_resolve.return_value = resolved
decoder = WeatherSatDecoder(output_dir=tmp_path / 'weather_sat_out')
success = decoder.start_from_file(
satellite='METEOR-M2-3',
input_file='data/weather_sat/samples/sample.wav',
sample_rate=1_000_000,
)
assert success is True
assert decoder.device_index == -1
mock_start.assert_called_once()
decoder.stop()
assert decoder.device_index == -1
tests/test_wefax.py
+52 -10
View File
@@ -225,7 +225,7 @@ class TestWeFaxDecoder:
# Route tests
# ---------------------------------------------------------------------------
class TestWeFaxRoutes:
class TestWeFaxRoutes:
"""WeFax route endpoint tests."""
def test_status(self, client):
@@ -419,12 +419,54 @@ class TestWeFaxRoutes:
assert response.status_code == 400
def test_delete_image_wrong_extension(self, client):
"""DELETE /wefax/images/<filename> should reject non-PNG."""
_login_session(client)
mock_decoder = MagicMock()
with patch('routes.wefax.get_wefax_decoder', return_value=mock_decoder):
response = client.delete('/wefax/images/test.jpg')
assert response.status_code == 400
def test_delete_image_wrong_extension(self, client):
"""DELETE /wefax/images/<filename> should reject non-PNG."""
_login_session(client)
mock_decoder = MagicMock()
with patch('routes.wefax.get_wefax_decoder', return_value=mock_decoder):
response = client.delete('/wefax/images/test.jpg')
assert response.status_code == 400
class TestWeFaxProgressCallback:
"""Regression tests for WeFax route-level progress callback behavior."""
def test_terminal_progress_releases_active_device(self):
"""Terminal decoder events must release any manually claimed SDR."""
import routes.wefax as wefax_routes
original_device = wefax_routes.wefax_active_device
try:
wefax_routes.wefax_active_device = 3
with patch('routes.wefax.app_module.release_sdr_device') as mock_release:
wefax_routes._progress_callback({
'type': 'wefax_progress',
'status': 'error',
'message': 'decode failed',
})
mock_release.assert_called_once_with(3)
assert wefax_routes.wefax_active_device is None
finally:
wefax_routes.wefax_active_device = original_device
def test_non_terminal_progress_does_not_release_active_device(self):
"""Non-terminal progress updates must not release SDR ownership."""
import routes.wefax as wefax_routes
original_device = wefax_routes.wefax_active_device
try:
wefax_routes.wefax_active_device = 4
with patch('routes.wefax.app_module.release_sdr_device') as mock_release:
wefax_routes._progress_callback({
'type': 'wefax_progress',
'status': 'receiving',
'line_count': 120,
})
mock_release.assert_not_called()
assert wefax_routes.wefax_active_device == 4
finally:
wefax_routes.wefax_active_device = original_device
tests/test_wefax_scheduler.py
+159
View File
@@ -0,0 +1,159 @@
"""Tests for WeFax auto-scheduler behavior and regressions."""
from __future__ import annotations
from datetime import datetime, timedelta, timezone
from unittest.mock import MagicMock, patch
from utils.wefax_scheduler import ScheduledBroadcast, WeFaxScheduler
class TestWeFaxScheduler:
"""WeFaxScheduler regression tests."""
@patch('threading.Timer')
def test_refresh_reschedules_same_utc_slot_next_day(self, mock_timer):
"""Completed broadcasts must not block the next day's same UTC slot."""
scheduler = WeFaxScheduler()
scheduler._enabled = True
scheduler._station = 'USCG Kodiak'
scheduler._callsign = 'NOJ'
scheduler._frequency_khz = 4298.0
now = datetime.now(timezone.utc)
utc_time = (now - timedelta(hours=2)).strftime('%H:%M')
today = now.date().isoformat()
prior = ScheduledBroadcast(
station='USCG Kodiak',
callsign='NOJ',
frequency_khz=4298.0,
utc_time=utc_time,
duration_min=20,
content='Chart',
occurrence_date=today,
)
prior.status = 'complete'
scheduler._broadcasts = [prior]
mock_timer.return_value = MagicMock()
with patch('utils.wefax_scheduler.get_station', return_value={
'name': 'USCG Kodiak',
'schedule': [{
'utc': utc_time,
'duration_min': 20,
'content': 'Chart',
}],
}):
scheduler._refresh_schedule()
capture_calls = [
c for c in mock_timer.call_args_list
if len(c.args) >= 2 and getattr(c.args[1], '__name__', '') == '_execute_capture'
]
assert capture_calls, "Expected a capture timer for the next-day occurrence"
scheduled = [b for b in scheduler._broadcasts if b.status == 'scheduled']
assert len(scheduled) == 1
assert scheduled[0].occurrence_date != today
def test_execute_capture_stops_immediately_if_window_elapsed(self):
"""If stop delay computes to <= 0, capture should close out immediately."""
scheduler = WeFaxScheduler()
scheduler._enabled = True
scheduler._callsign = 'NOJ'
scheduler._frequency_khz = 4298.0
scheduler._device = 0
scheduler._gain = 40.0
scheduler._ioc = 576
scheduler._lpm = 120
scheduler._direct_sampling = True
now = datetime.now(timezone.utc)
sb = ScheduledBroadcast(
station='USCG Kodiak',
callsign='NOJ',
frequency_khz=4298.0,
utc_time=now.strftime('%H:%M'),
duration_min=0,
content='Late chart',
occurrence_date=now.date().isoformat(),
)
sb.status = 'scheduled'
mock_decoder = MagicMock()
mock_decoder.is_running = False
mock_decoder.start.return_value = True
with patch('utils.wefax_scheduler.get_wefax_decoder', return_value=mock_decoder), \
patch('utils.wefax_scheduler.WEFAX_CAPTURE_BUFFER_SECONDS', 0), \
patch('app.claim_sdr_device', return_value=None), \
patch.object(scheduler, '_stop_capture') as mock_stop_capture:
scheduler._execute_capture_inner(sb)
mock_stop_capture.assert_called_once()
@patch('threading.Timer')
def test_terminal_progress_releases_scheduler_device_early(self, mock_timer):
"""Scheduler captures must release SDR as soon as terminal progress arrives."""
scheduler = WeFaxScheduler()
scheduler._enabled = True
scheduler._callsign = 'NOJ'
scheduler._frequency_khz = 4298.0
scheduler._device = 0
scheduler._gain = 40.0
scheduler._ioc = 576
scheduler._lpm = 120
scheduler._direct_sampling = True
sb = ScheduledBroadcast(
station='USCG Kodiak',
callsign='NOJ',
frequency_khz=4298.0,
utc_time='12:00',
duration_min=20,
content='Chart',
occurrence_date='2026-01-01',
)
sb.status = 'scheduled'
mock_decoder = MagicMock()
mock_decoder.is_running = False
mock_decoder.start.return_value = True
mock_timer.return_value = MagicMock()
with patch('utils.wefax_scheduler.get_wefax_decoder', return_value=mock_decoder), \
patch('app.claim_sdr_device', return_value=None), \
patch('app.release_sdr_device') as mock_release:
scheduler._execute_capture_inner(sb)
progress_cb = mock_decoder.set_callback.call_args[0][0]
progress_cb({
'type': 'wefax_progress',
'status': 'error',
'message': 'rtl_fm failed',
})
mock_release.assert_called_once_with(0)
assert sb.status == 'skipped'
def test_stop_capture_non_capturing_only_releases(self):
"""_stop_capture should be idempotent when capture already ended."""
scheduler = WeFaxScheduler()
sb = ScheduledBroadcast(
station='USCG Kodiak',
callsign='NOJ',
frequency_khz=4298.0,
utc_time='12:00',
duration_min=20,
content='Chart',
occurrence_date='2026-01-01',
)
sb.status = 'complete'
release_fn = MagicMock()
with patch('utils.wefax_scheduler.get_wefax_decoder') as mock_get_decoder:
scheduler._stop_capture(sb, release_fn)
release_fn.assert_called_once()
mock_get_decoder.assert_not_called()