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

Merge upstream/main and resolve acars, vdl2, dashboard conflicts

Browse Source 
Resolved conflicts:
- routes/acars.py: keep /messages and /clear endpoints for history reload
- routes/vdl2.py: keep /messages and /clear endpoints for history reload
- templates/adsb_dashboard.html: keep removal of hardcoded device-1
  defaults for ACARS/VDL2 selectors (users pick their own device)

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
mitchross
2026-02-27 14:47:57 -05:00
parent bbab991c1f 2723d91526
commit 922db620cc
41 changed files with 5065 additions and 1951 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_morse.py
+6 -6
View File
@@ -257,8 +257,8 @@ class TestMorseLifecycleRoutes:
released_devices = []
monkeypatch.setattr(app_module, 'claim_sdr_device', lambda idx, mode: None)
monkeypatch.setattr(app_module, 'release_sdr_device', lambda idx: released_devices.append(idx))
monkeypatch.setattr(app_module, 'claim_sdr_device', lambda idx, mode, sdr_type='rtlsdr': None)
monkeypatch.setattr(app_module, 'release_sdr_device', lambda idx, sdr_type='rtlsdr': released_devices.append(idx))
class DummyDevice:
sdr_type = morse_routes.SDRType.RTL_SDR
@@ -337,8 +337,8 @@ class TestMorseLifecycleRoutes:
released_devices = []
monkeypatch.setattr(app_module, 'claim_sdr_device', lambda idx, mode: None)
monkeypatch.setattr(app_module, 'release_sdr_device', lambda idx: released_devices.append(idx))
monkeypatch.setattr(app_module, 'claim_sdr_device', lambda idx, mode, sdr_type='rtlsdr': None)
monkeypatch.setattr(app_module, 'release_sdr_device', lambda idx, sdr_type='rtlsdr': released_devices.append(idx))
class DummyDevice:
sdr_type = morse_routes.SDRType.RTL_SDR
@@ -421,8 +421,8 @@ class TestMorseLifecycleRoutes:
released_devices = []
monkeypatch.setattr(app_module, 'claim_sdr_device', lambda idx, mode: None)
monkeypatch.setattr(app_module, 'release_sdr_device', lambda idx: released_devices.append(idx))
monkeypatch.setattr(app_module, 'claim_sdr_device', lambda idx, mode, sdr_type='rtlsdr': None)
monkeypatch.setattr(app_module, 'release_sdr_device', lambda idx, sdr_type='rtlsdr': released_devices.append(idx))
class DummyDevice:
def __init__(self, index: int):
tests/test_system.py
+141
View File
@@ -30,6 +30,32 @@ def test_metrics_returns_expected_keys(client):
assert 'uptime_human' in data['system']
def test_metrics_enhanced_keys(client):
"""GET /system/metrics returns enhanced metric keys."""
_login(client)
resp = client.get('/system/metrics')
assert resp.status_code == 200
data = resp.get_json()
# New enhanced keys
assert 'network' in data
assert 'disk_io' in data
assert 'boot_time' in data
assert 'battery' in data
assert 'fans' in data
assert 'power' in data
# CPU should have per_core and freq
if data['cpu'] is not None:
assert 'per_core' in data['cpu']
assert 'freq' in data['cpu']
# Network should have interfaces and connections
if data['network'] is not None:
assert 'interfaces' in data['network']
assert 'connections' in data['network']
assert 'io' in data['network']
def test_metrics_without_psutil(client):
"""Metrics degrade gracefully when psutil is unavailable."""
_login(client)
@@ -45,6 +71,11 @@ def test_metrics_without_psutil(client):
assert data['cpu'] is None
assert data['memory'] is None
assert data['disk'] is None
assert data['network'] is None
assert data['disk_io'] is None
assert data['battery'] is None
assert data['boot_time'] is None
assert data['power'] is None
finally:
mod._HAS_PSUTIL = orig
@@ -87,3 +118,113 @@ def test_stream_returns_sse_content_type(client):
resp = client.get('/system/stream')
assert resp.status_code == 200
assert 'text/event-stream' in resp.content_type
def test_location_returns_shape(client):
"""GET /system/location returns lat/lon/source shape."""
_login(client)
resp = client.get('/system/location')
assert resp.status_code == 200
data = resp.get_json()
assert 'lat' in data
assert 'lon' in data
assert 'source' in data
def test_location_from_gps(client):
"""Location endpoint returns GPS data when fix available."""
_login(client)
mock_pos = MagicMock()
mock_pos.fix_quality = 3
mock_pos.latitude = 51.5074
mock_pos.longitude = -0.1278
mock_pos.satellites = 12
mock_pos.epx = 2.5
mock_pos.epy = 3.1
mock_pos.altitude = 45.0
with patch('routes.system.get_current_position', return_value=mock_pos, create=True):
# Patch the import inside the function
import routes.system as mod
original = mod._get_observer_location
def _patched():
with patch('utils.gps.get_current_position', return_value=mock_pos):
return original()
mod._get_observer_location = _patched
try:
resp = client.get('/system/location')
finally:
mod._get_observer_location = original
assert resp.status_code == 200
data = resp.get_json()
assert data['source'] == 'gps'
assert data['lat'] == 51.5074
assert data['lon'] == -0.1278
assert data['gps']['fix_quality'] == 3
assert data['gps']['satellites'] == 12
assert data['gps']['accuracy'] == 3.1
assert data['gps']['altitude'] == 45.0
def test_location_falls_back_to_defaults(client):
"""Location endpoint returns constants defaults when GPS and config unavailable."""
_login(client)
resp = client.get('/system/location')
assert resp.status_code == 200
data = resp.get_json()
assert 'source' in data
# Should get location from config or default constants
assert data['lat'] is not None
assert data['lon'] is not None
assert data['source'] in ('config', 'default')
def test_weather_requires_location(client):
"""Weather endpoint returns error when no location available."""
_login(client)
# Without lat/lon params and no GPS state or config
resp = client.get('/system/weather')
assert resp.status_code == 200
data = resp.get_json()
# Either returns weather or error (depending on config)
assert 'error' in data or 'temp_c' in data
def test_weather_with_mocked_response(client):
"""Weather endpoint returns parsed weather data with mocked HTTP."""
_login(client)
mock_resp = MagicMock()
mock_resp.status_code = 200
mock_resp.json.return_value = {
'current_condition': [{
'temp_C': '22',
'temp_F': '72',
'weatherDesc': [{'value': 'Clear'}],
'humidity': '45',
'windspeedMiles': '8',
'winddir16Point': 'NW',
'FeelsLikeC': '20',
'visibility': '10',
'pressure': '1013',
}]
}
mock_resp.raise_for_status = MagicMock()
import routes.system as mod
# Clear cache
mod._weather_cache.clear()
mod._weather_cache_time = 0.0
with patch('routes.system._requests') as mock_requests:
mock_requests.get.return_value = mock_resp
resp = client.get('/system/weather?lat=40.7&lon=-74.0')
assert resp.status_code == 200
data = resp.get_json()
assert data['temp_c'] == '22'
assert data['condition'] == 'Clear'
assert data['humidity'] == '45'
assert data['wind_mph'] == '8'
tests/test_wefax.py
+192 -192
View File
@@ -54,72 +54,72 @@ class TestWeFaxStations:
from utils.wefax_stations import get_station
assert get_station('noj') is not None
def test_get_station_not_found(self):
"""get_station() should return None for unknown callsign."""
from utils.wefax_stations import get_station
assert get_station('XXXXX') is None
def test_resolve_tuning_frequency_auto_uses_carrier_for_known_station(self):
"""Known station frequencies default to carrier-list behavior in auto mode."""
from utils.wefax_stations import resolve_tuning_frequency_khz
tuned, reference, offset_applied = resolve_tuning_frequency_khz(
listed_frequency_khz=4298.0,
station_callsign='NOJ',
frequency_reference='auto',
)
assert math.isclose(tuned, 4296.1, abs_tol=1e-6)
assert reference == 'carrier'
assert offset_applied is True
def test_resolve_tuning_frequency_auto_preserves_unknown_station_input(self):
"""Ad-hoc frequencies (no station metadata) should be treated as dial."""
from utils.wefax_stations import resolve_tuning_frequency_khz
tuned, reference, offset_applied = resolve_tuning_frequency_khz(
listed_frequency_khz=4298.0,
station_callsign='',
frequency_reference='auto',
)
assert math.isclose(tuned, 4298.0, abs_tol=1e-6)
assert reference == 'dial'
assert offset_applied is False
def test_resolve_tuning_frequency_dial_override(self):
"""Explicit dial reference must bypass USB alignment."""
from utils.wefax_stations import resolve_tuning_frequency_khz
tuned, reference, offset_applied = resolve_tuning_frequency_khz(
listed_frequency_khz=4298.0,
station_callsign='NOJ',
frequency_reference='dial',
)
assert math.isclose(tuned, 4298.0, abs_tol=1e-6)
assert reference == 'dial'
assert offset_applied is False
def test_resolve_tuning_frequency_rejects_invalid_reference(self):
"""Invalid frequency reference values should raise a validation error."""
from utils.wefax_stations import resolve_tuning_frequency_khz
try:
resolve_tuning_frequency_khz(
listed_frequency_khz=4298.0,
station_callsign='NOJ',
frequency_reference='invalid',
)
assert False, "Expected ValueError for invalid frequency_reference"
except ValueError as exc:
assert 'frequency_reference' in str(exc)
def test_station_frequencies_have_khz(self):
"""Each frequency entry must have 'khz' and 'description'."""
from utils.wefax_stations import load_stations
for station in load_stations():
for freq in station['frequencies']:
def test_get_station_not_found(self):
"""get_station() should return None for unknown callsign."""
from utils.wefax_stations import get_station
assert get_station('XXXXX') is None
def test_resolve_tuning_frequency_auto_uses_carrier_for_known_station(self):
"""Known station frequencies default to carrier-list behavior in auto mode."""
from utils.wefax_stations import resolve_tuning_frequency_khz
tuned, reference, offset_applied = resolve_tuning_frequency_khz(
listed_frequency_khz=4298.0,
station_callsign='NOJ',
frequency_reference='auto',
)
assert math.isclose(tuned, 4296.1, abs_tol=1e-6)
assert reference == 'carrier'
assert offset_applied is True
def test_resolve_tuning_frequency_auto_preserves_unknown_station_input(self):
"""Ad-hoc frequencies (no station metadata) should be treated as dial."""
from utils.wefax_stations import resolve_tuning_frequency_khz
tuned, reference, offset_applied = resolve_tuning_frequency_khz(
listed_frequency_khz=4298.0,
station_callsign='',
frequency_reference='auto',
)
assert math.isclose(tuned, 4298.0, abs_tol=1e-6)
assert reference == 'dial'
assert offset_applied is False
def test_resolve_tuning_frequency_dial_override(self):
"""Explicit dial reference must bypass USB alignment."""
from utils.wefax_stations import resolve_tuning_frequency_khz
tuned, reference, offset_applied = resolve_tuning_frequency_khz(
listed_frequency_khz=4298.0,
station_callsign='NOJ',
frequency_reference='dial',
)
assert math.isclose(tuned, 4298.0, abs_tol=1e-6)
assert reference == 'dial'
assert offset_applied is False
def test_resolve_tuning_frequency_rejects_invalid_reference(self):
"""Invalid frequency reference values should raise a validation error."""
from utils.wefax_stations import resolve_tuning_frequency_khz
try:
resolve_tuning_frequency_khz(
listed_frequency_khz=4298.0,
station_callsign='NOJ',
frequency_reference='invalid',
)
assert False, "Expected ValueError for invalid frequency_reference"
except ValueError as exc:
assert 'frequency_reference' in str(exc)
def test_station_frequencies_have_khz(self):
"""Each frequency entry must have 'khz' and 'description'."""
from utils.wefax_stations import load_stations
for station in load_stations():
for freq in station['frequencies']:
assert 'khz' in freq, f"{station['callsign']} missing khz"
assert 'description' in freq, f"{station['callsign']} missing description"
assert isinstance(freq['khz'], (int, float))
@@ -281,7 +281,7 @@ class TestWeFaxDecoder:
# Route tests
# ---------------------------------------------------------------------------
class TestWeFaxRoutes:
class TestWeFaxRoutes:
"""WeFax route endpoint tests."""
def test_status(self, client):
@@ -390,11 +390,11 @@ class TestWeFaxRoutes:
data = response.get_json()
assert 'LPM' in data['message']
def test_start_success(self, client):
"""POST /wefax/start with valid params should succeed."""
_login_session(client)
mock_decoder = MagicMock()
mock_decoder.is_running = False
def test_start_success(self, client):
"""POST /wefax/start with valid params should succeed."""
_login_session(client)
mock_decoder = MagicMock()
mock_decoder.is_running = False
mock_decoder.start.return_value = True
with patch('routes.wefax.get_wefax_decoder', return_value=mock_decoder), \
@@ -411,46 +411,46 @@ class TestWeFaxRoutes:
content_type='application/json',
)
assert response.status_code == 200
data = response.get_json()
assert data['status'] == 'started'
assert data['frequency_khz'] == 4298
assert data['usb_offset_applied'] is True
assert math.isclose(data['tuned_frequency_khz'], 4296.1, abs_tol=1e-6)
assert data['frequency_reference'] == 'carrier'
assert data['station'] == 'NOJ'
mock_decoder.start.assert_called_once()
start_kwargs = mock_decoder.start.call_args.kwargs
assert math.isclose(start_kwargs['frequency_khz'], 4296.1, abs_tol=1e-6)
def test_start_respects_dial_reference_override(self, client):
"""POST /wefax/start with dial reference should not apply USB offset."""
_login_session(client)
mock_decoder = MagicMock()
mock_decoder.is_running = False
mock_decoder.start.return_value = True
with patch('routes.wefax.get_wefax_decoder', return_value=mock_decoder), \
patch('routes.wefax.app_module.claim_sdr_device', return_value=None):
response = client.post(
'/wefax/start',
data=json.dumps({
'frequency_khz': 4298,
'station': 'NOJ',
'device': 0,
'frequency_reference': 'dial',
}),
content_type='application/json',
)
assert response.status_code == 200
data = response.get_json()
assert data['status'] == 'started'
assert data['usb_offset_applied'] is False
assert math.isclose(data['tuned_frequency_khz'], 4298.0, abs_tol=1e-6)
assert data['frequency_reference'] == 'dial'
start_kwargs = mock_decoder.start.call_args.kwargs
assert math.isclose(start_kwargs['frequency_khz'], 4298.0, abs_tol=1e-6)
assert response.status_code == 200
data = response.get_json()
assert data['status'] == 'started'
assert data['frequency_khz'] == 4298
assert data['usb_offset_applied'] is True
assert math.isclose(data['tuned_frequency_khz'], 4296.1, abs_tol=1e-6)
assert data['frequency_reference'] == 'carrier'
assert data['station'] == 'NOJ'
mock_decoder.start.assert_called_once()
start_kwargs = mock_decoder.start.call_args.kwargs
assert math.isclose(start_kwargs['frequency_khz'], 4296.1, abs_tol=1e-6)
def test_start_respects_dial_reference_override(self, client):
"""POST /wefax/start with dial reference should not apply USB offset."""
_login_session(client)
mock_decoder = MagicMock()
mock_decoder.is_running = False
mock_decoder.start.return_value = True
with patch('routes.wefax.get_wefax_decoder', return_value=mock_decoder), \
patch('routes.wefax.app_module.claim_sdr_device', return_value=None):
response = client.post(
'/wefax/start',
data=json.dumps({
'frequency_khz': 4298,
'station': 'NOJ',
'device': 0,
'frequency_reference': 'dial',
}),
content_type='application/json',
)
assert response.status_code == 200
data = response.get_json()
assert data['status'] == 'started'
assert data['usb_offset_applied'] is False
assert math.isclose(data['tuned_frequency_khz'], 4298.0, abs_tol=1e-6)
assert data['frequency_reference'] == 'dial'
start_kwargs = mock_decoder.start.call_args.kwargs
assert math.isclose(start_kwargs['frequency_khz'], 4298.0, abs_tol=1e-6)
def test_start_device_busy(self, client):
"""POST /wefax/start should return 409 when device is busy."""
@@ -509,83 +509,83 @@ 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_schedule_enable_applies_usb_alignment(self, client):
"""Scheduler should receive tuned USB dial frequency in auto mode."""
_login_session(client)
mock_scheduler = MagicMock()
mock_scheduler.enable.return_value = {
'enabled': True,
'scheduled_count': 2,
'total_broadcasts': 2,
}
with patch('utils.wefax_scheduler.get_wefax_scheduler', return_value=mock_scheduler):
response = client.post(
'/wefax/schedule/enable',
data=json.dumps({
'station': 'NOJ',
'frequency_khz': 4298,
'device': 0,
}),
content_type='application/json',
)
assert response.status_code == 200
data = response.get_json()
assert data['status'] == 'ok'
assert data['usb_offset_applied'] is True
assert math.isclose(data['tuned_frequency_khz'], 4296.1, abs_tol=1e-6)
enable_kwargs = mock_scheduler.enable.call_args.kwargs
assert math.isclose(enable_kwargs['frequency_khz'], 4296.1, abs_tol=1e-6)
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
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_schedule_enable_applies_usb_alignment(self, client):
"""Scheduler should receive tuned USB dial frequency in auto mode."""
_login_session(client)
mock_scheduler = MagicMock()
mock_scheduler.enable.return_value = {
'enabled': True,
'scheduled_count': 2,
'total_broadcasts': 2,
}
with patch('utils.wefax_scheduler.get_wefax_scheduler', return_value=mock_scheduler):
response = client.post(
'/wefax/schedule/enable',
data=json.dumps({
'station': 'NOJ',
'frequency_khz': 4298,
'device': 0,
}),
content_type='application/json',
)
assert response.status_code == 200
data = response.get_json()
assert data['status'] == 'ok'
assert data['usb_offset_applied'] is True
assert math.isclose(data['tuned_frequency_khz'], 4296.1, abs_tol=1e-6)
enable_kwargs = mock_scheduler.enable.call_args.kwargs
assert math.isclose(enable_kwargs['frequency_khz'], 4296.1, abs_tol=1e-6)
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, 'rtlsdr')
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