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

fix: SDR device registry collision with multiple SDR types

Browse Source 
The registry used plain int keys (device index), so HackRF at index 0
and RTL-SDR at index 0 would collide. Changed to composite string keys
("sdr_type:index") so each SDR type+index pair is tracked independently.
Updated all route callers, frontend device selectors, and session restore.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Smittix
2026-02-27 09:06:41 +00:00
parent 304e02c48b
commit 62be720c0a
17 changed files with 907 additions and 786 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_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