mirror of
https://github.com/smittix/intercept.git
synced 2026-07-05 16:18:12 -07:00
Add DMR digital voice, WebSDR, and listening post enhancements
- DMR/P25 digital voice decoder mode with DSD-FME integration - WebSDR mode with KiwiSDR audio proxy and websocket-client support - Listening post waterfall/spectrogram visualization and audio streaming - Dockerfile updates for mbelib and DSD-FME build dependencies - New tests for DMR, WebSDR, KiwiSDR, waterfall, and signal guess API - Chart.js date adapter for time-scale axes Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Smittix
+4
-2
@@ -5,11 +5,13 @@ from app import app as flask_app
|
||||
from routes import register_blueprints
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
@pytest.fixture(scope='session')
|
||||
def app():
|
||||
"""Create application for testing."""
|
||||
register_blueprints(flask_app)
|
||||
flask_app.config['TESTING'] = True
|
||||
# Register blueprints only if not already registered
|
||||
if 'pager' not in flask_app.blueprints:
|
||||
register_blueprints(flask_app)
|
||||
return flask_app
|
||||
|
||||
|
||||
|
||||
@@ -0,0 +1,145 @@
|
||||
"""Tests for the DMR / Digital Voice decoding module."""
|
||||
|
||||
from unittest.mock import patch, MagicMock
|
||||
import pytest
|
||||
from routes.dmr import parse_dsd_output
|
||||
|
||||
|
||||
# ============================================
|
||||
# parse_dsd_output() tests
|
||||
# ============================================
|
||||
|
||||
def test_parse_sync_dmr():
|
||||
"""Should parse DMR sync line."""
|
||||
result = parse_dsd_output('Sync: +DMR (data)')
|
||||
assert result is not None
|
||||
assert result['type'] == 'sync'
|
||||
assert 'DMR' in result['protocol']
|
||||
|
||||
|
||||
def test_parse_sync_p25():
|
||||
"""Should parse P25 sync line."""
|
||||
result = parse_dsd_output('Sync: +P25 Phase 1')
|
||||
assert result is not None
|
||||
assert result['type'] == 'sync'
|
||||
assert 'P25' in result['protocol']
|
||||
|
||||
|
||||
def test_parse_talkgroup_and_source():
|
||||
"""Should parse talkgroup and source ID."""
|
||||
result = parse_dsd_output('TG: 12345 Src: 67890')
|
||||
assert result is not None
|
||||
assert result['type'] == 'call'
|
||||
assert result['talkgroup'] == 12345
|
||||
assert result['source_id'] == 67890
|
||||
|
||||
|
||||
def test_parse_slot():
|
||||
"""Should parse slot info."""
|
||||
result = parse_dsd_output('Slot 1')
|
||||
assert result is not None
|
||||
assert result['type'] == 'slot'
|
||||
assert result['slot'] == 1
|
||||
|
||||
|
||||
def test_parse_voice():
|
||||
"""Should parse voice frame info."""
|
||||
result = parse_dsd_output('Voice Frame 1')
|
||||
assert result is not None
|
||||
assert result['type'] == 'voice'
|
||||
|
||||
|
||||
def test_parse_nac():
|
||||
"""Should parse P25 NAC."""
|
||||
result = parse_dsd_output('NAC: 293')
|
||||
assert result is not None
|
||||
assert result['type'] == 'nac'
|
||||
assert result['nac'] == '293'
|
||||
|
||||
|
||||
def test_parse_empty_line():
|
||||
"""Empty lines should return None."""
|
||||
assert parse_dsd_output('') is None
|
||||
assert parse_dsd_output(' ') is None
|
||||
|
||||
|
||||
def test_parse_unrecognized():
|
||||
"""Unrecognized lines should return None."""
|
||||
assert parse_dsd_output('some random text') is None
|
||||
|
||||
|
||||
# ============================================
|
||||
# Endpoint tests
|
||||
# ============================================
|
||||
|
||||
@pytest.fixture
|
||||
def auth_client(client):
|
||||
"""Client with logged-in session."""
|
||||
with client.session_transaction() as sess:
|
||||
sess['logged_in'] = True
|
||||
return client
|
||||
|
||||
|
||||
def test_dmr_tools(auth_client):
|
||||
"""Tools endpoint should return availability info."""
|
||||
resp = auth_client.get('/dmr/tools')
|
||||
assert resp.status_code == 200
|
||||
data = resp.get_json()
|
||||
assert 'dsd' in data
|
||||
assert 'rtl_fm' in data
|
||||
assert 'protocols' in data
|
||||
|
||||
|
||||
def test_dmr_status(auth_client):
|
||||
"""Status endpoint should work."""
|
||||
resp = auth_client.get('/dmr/status')
|
||||
assert resp.status_code == 200
|
||||
data = resp.get_json()
|
||||
assert 'running' in data
|
||||
|
||||
|
||||
def test_dmr_start_no_dsd(auth_client):
|
||||
"""Start should fail gracefully when dsd is not installed."""
|
||||
with patch('routes.dmr.find_dsd', return_value=None):
|
||||
resp = auth_client.post('/dmr/start', json={
|
||||
'frequency': 462.5625,
|
||||
'protocol': 'auto',
|
||||
})
|
||||
assert resp.status_code == 503
|
||||
data = resp.get_json()
|
||||
assert 'dsd' in data['message']
|
||||
|
||||
|
||||
def test_dmr_start_no_rtl_fm(auth_client):
|
||||
"""Start should fail when rtl_fm is missing."""
|
||||
with patch('routes.dmr.find_dsd', return_value='/usr/bin/dsd'), \
|
||||
patch('routes.dmr.find_rtl_fm', return_value=None):
|
||||
resp = auth_client.post('/dmr/start', json={
|
||||
'frequency': 462.5625,
|
||||
})
|
||||
assert resp.status_code == 503
|
||||
|
||||
|
||||
def test_dmr_start_invalid_protocol(auth_client):
|
||||
"""Start should reject invalid protocol."""
|
||||
with patch('routes.dmr.find_dsd', return_value='/usr/bin/dsd'), \
|
||||
patch('routes.dmr.find_rtl_fm', return_value='/usr/bin/rtl_fm'):
|
||||
resp = auth_client.post('/dmr/start', json={
|
||||
'frequency': 462.5625,
|
||||
'protocol': 'invalid',
|
||||
})
|
||||
assert resp.status_code == 400
|
||||
|
||||
|
||||
def test_dmr_stop(auth_client):
|
||||
"""Stop should succeed."""
|
||||
resp = auth_client.post('/dmr/stop')
|
||||
assert resp.status_code == 200
|
||||
data = resp.get_json()
|
||||
assert data['status'] == 'stopped'
|
||||
|
||||
|
||||
def test_dmr_stream_mimetype(auth_client):
|
||||
"""Stream should return event-stream content type."""
|
||||
resp = auth_client.get('/dmr/stream')
|
||||
assert resp.content_type.startswith('text/event-stream')
|
||||
@@ -0,0 +1,321 @@
|
||||
"""Tests for the KiwiSDR WebSocket audio client."""
|
||||
|
||||
import struct
|
||||
from unittest.mock import patch, MagicMock
|
||||
|
||||
import pytest
|
||||
|
||||
from utils.kiwisdr import (
|
||||
KiwiSDRClient,
|
||||
KIWI_SAMPLE_RATE,
|
||||
KIWI_SND_HEADER_SIZE,
|
||||
KIWI_DEFAULT_PORT,
|
||||
MODE_FILTERS,
|
||||
VALID_MODES,
|
||||
parse_host_port,
|
||||
)
|
||||
|
||||
|
||||
# ============================================
|
||||
# parse_host_port tests
|
||||
# ============================================
|
||||
|
||||
def test_parse_host_port_basic():
|
||||
"""Should parse host:port from a simple URL."""
|
||||
assert parse_host_port('http://kiwi.example.com:8073') == ('kiwi.example.com', 8073)
|
||||
|
||||
|
||||
def test_parse_host_port_no_port():
|
||||
"""Should default to 8073 when port is missing."""
|
||||
assert parse_host_port('http://kiwi.example.com') == ('kiwi.example.com', KIWI_DEFAULT_PORT)
|
||||
|
||||
|
||||
def test_parse_host_port_https():
|
||||
"""Should strip https:// prefix."""
|
||||
assert parse_host_port('https://secure.kiwi.com:9090') == ('secure.kiwi.com', 9090)
|
||||
|
||||
|
||||
def test_parse_host_port_ws():
|
||||
"""Should strip ws:// prefix."""
|
||||
assert parse_host_port('ws://kiwi.local:8074') == ('kiwi.local', 8074)
|
||||
|
||||
|
||||
def test_parse_host_port_with_path():
|
||||
"""Should strip trailing path from URL."""
|
||||
assert parse_host_port('http://kiwi.com:8073/some/path') == ('kiwi.com', 8073)
|
||||
|
||||
|
||||
def test_parse_host_port_bare_host():
|
||||
"""Should handle bare hostname without protocol."""
|
||||
assert parse_host_port('kiwi.local') == ('kiwi.local', KIWI_DEFAULT_PORT)
|
||||
|
||||
|
||||
def test_parse_host_port_bare_host_with_port():
|
||||
"""Should handle bare hostname with port."""
|
||||
assert parse_host_port('kiwi.local:8074') == ('kiwi.local', 8074)
|
||||
|
||||
|
||||
def test_parse_host_port_empty():
|
||||
"""Should handle empty/None input."""
|
||||
assert parse_host_port('') == ('', KIWI_DEFAULT_PORT)
|
||||
|
||||
|
||||
def test_parse_host_port_invalid_port():
|
||||
"""Should default port for non-numeric port."""
|
||||
assert parse_host_port('http://kiwi.com:abc') == ('kiwi.com', KIWI_DEFAULT_PORT)
|
||||
|
||||
|
||||
# ============================================
|
||||
# SND frame parsing tests
|
||||
# ============================================
|
||||
|
||||
def _make_snd_frame(smeter_raw: int, pcm_samples: list[int]) -> bytes:
|
||||
"""Build a mock KiwiSDR SND binary frame."""
|
||||
header = b'SND' # 3 bytes: magic
|
||||
header += b'\x00' # 1 byte: flags
|
||||
header += struct.pack('>I', 42) # 4 bytes: sequence number
|
||||
header += struct.pack('>h', smeter_raw) # 2 bytes: S-meter
|
||||
# PCM data: 16-bit signed LE
|
||||
pcm = b''.join(struct.pack('<h', s) for s in pcm_samples)
|
||||
return header + pcm
|
||||
|
||||
|
||||
def test_parse_snd_frame_smeter():
|
||||
"""Should extract S-meter value from SND frame."""
|
||||
client = KiwiSDRClient(host='test', port=8073)
|
||||
audio_data = []
|
||||
|
||||
def on_audio(pcm, smeter):
|
||||
audio_data.append((pcm, smeter))
|
||||
|
||||
client._on_audio = on_audio
|
||||
frame = _make_snd_frame(-730, [100, -100, 200]) # -73.0 dBm = S9
|
||||
client._parse_snd_frame(frame)
|
||||
|
||||
assert client.last_smeter == -730
|
||||
assert len(audio_data) == 1
|
||||
assert audio_data[0][1] == -730
|
||||
|
||||
|
||||
def test_parse_snd_frame_pcm_data():
|
||||
"""Should forward PCM data from SND frame."""
|
||||
client = KiwiSDRClient(host='test', port=8073)
|
||||
received_pcm = []
|
||||
|
||||
def on_audio(pcm, smeter):
|
||||
received_pcm.append(pcm)
|
||||
|
||||
client._on_audio = on_audio
|
||||
samples = [1000, -2000, 3000, -4000]
|
||||
frame = _make_snd_frame(0, samples)
|
||||
client._parse_snd_frame(frame)
|
||||
|
||||
assert len(received_pcm) == 1
|
||||
# PCM data is 8 bytes (4 samples * 2 bytes each)
|
||||
assert len(received_pcm[0]) == len(samples) * 2
|
||||
|
||||
|
||||
def test_parse_snd_frame_short():
|
||||
"""Should ignore frames shorter than header size."""
|
||||
client = KiwiSDRClient(host='test', port=8073)
|
||||
client._on_audio = MagicMock()
|
||||
client._parse_snd_frame(b'SND\x00') # Too short
|
||||
client._on_audio.assert_not_called()
|
||||
|
||||
|
||||
def test_parse_snd_frame_wrong_magic():
|
||||
"""Should ignore frames with wrong header magic."""
|
||||
client = KiwiSDRClient(host='test', port=8073)
|
||||
client._on_audio = MagicMock()
|
||||
frame = b'XXX' + b'\x00' * 7 + b'\x00' * 10 # Wrong magic
|
||||
client._parse_snd_frame(frame)
|
||||
client._on_audio.assert_not_called()
|
||||
|
||||
|
||||
# ============================================
|
||||
# Client state tests
|
||||
# ============================================
|
||||
|
||||
def test_client_initial_state():
|
||||
"""New client should start disconnected."""
|
||||
client = KiwiSDRClient(host='kiwi.local', port=8073)
|
||||
assert client.connected is False
|
||||
assert client.host == 'kiwi.local'
|
||||
assert client.port == 8073
|
||||
assert client.frequency_khz == 0
|
||||
assert client.mode == 'am'
|
||||
|
||||
|
||||
def test_client_tune_when_disconnected():
|
||||
"""Tune should fail when not connected."""
|
||||
client = KiwiSDRClient(host='test', port=8073)
|
||||
assert client.tune(7000, 'usb') is False
|
||||
|
||||
|
||||
def test_client_disconnect_when_not_connected():
|
||||
"""Disconnect should not raise when already disconnected."""
|
||||
client = KiwiSDRClient(host='test', port=8073)
|
||||
client.disconnect() # Should not raise
|
||||
assert client.connected is False
|
||||
|
||||
|
||||
@patch('utils.kiwisdr.WEBSOCKET_CLIENT_AVAILABLE', False)
|
||||
def test_client_connect_no_websocket():
|
||||
"""Connect should fail if websocket-client not available."""
|
||||
client = KiwiSDRClient(host='test', port=8073)
|
||||
assert client.connect(7000, 'am') is False
|
||||
|
||||
|
||||
# ============================================
|
||||
# Constants tests
|
||||
# ============================================
|
||||
|
||||
def test_sample_rate():
|
||||
"""Sample rate should be 12 kHz."""
|
||||
assert KIWI_SAMPLE_RATE == 12000
|
||||
|
||||
|
||||
def test_snd_header_size():
|
||||
"""SND header should be 10 bytes."""
|
||||
assert KIWI_SND_HEADER_SIZE == 10
|
||||
|
||||
|
||||
def test_valid_modes():
|
||||
"""All expected modes should be in VALID_MODES."""
|
||||
assert 'am' in VALID_MODES
|
||||
assert 'usb' in VALID_MODES
|
||||
assert 'lsb' in VALID_MODES
|
||||
assert 'cw' in VALID_MODES
|
||||
|
||||
|
||||
def test_mode_filters_defined():
|
||||
"""All valid modes should have filter definitions."""
|
||||
for mode in VALID_MODES:
|
||||
assert mode in MODE_FILTERS
|
||||
low, high = MODE_FILTERS[mode]
|
||||
assert low < high
|
||||
|
||||
|
||||
def test_mode_filter_am_symmetric():
|
||||
"""AM filter should be symmetric."""
|
||||
low, high = MODE_FILTERS['am']
|
||||
assert low == -high
|
||||
|
||||
|
||||
def test_mode_filter_usb_positive():
|
||||
"""USB filter should be in positive passband."""
|
||||
low, high = MODE_FILTERS['usb']
|
||||
assert low > 0
|
||||
assert high > low
|
||||
|
||||
|
||||
def test_mode_filter_lsb_negative():
|
||||
"""LSB filter should be in negative passband."""
|
||||
low, high = MODE_FILTERS['lsb']
|
||||
assert low < 0
|
||||
assert high < 0
|
||||
|
||||
|
||||
# ============================================
|
||||
# Connection tests with mocked WebSocket
|
||||
# ============================================
|
||||
|
||||
@patch('utils.kiwisdr.WEBSOCKET_CLIENT_AVAILABLE', True)
|
||||
@patch('utils.kiwisdr.websocket')
|
||||
def test_client_connect_success(mock_ws_module):
|
||||
"""Connect should establish a WebSocket connection."""
|
||||
mock_ws = MagicMock()
|
||||
mock_ws_module.WebSocket.return_value = mock_ws
|
||||
|
||||
client = KiwiSDRClient(host='kiwi.local', port=8073)
|
||||
result = client.connect(7000, 'am')
|
||||
|
||||
assert result is True
|
||||
assert client.connected is True
|
||||
assert client.frequency_khz == 7000
|
||||
assert client.mode == 'am'
|
||||
|
||||
# Verify WebSocket was created and connected
|
||||
mock_ws_module.WebSocket.assert_called_once()
|
||||
mock_ws.connect.assert_called_once()
|
||||
|
||||
# Verify protocol messages were sent
|
||||
calls = [str(c) for c in mock_ws.send.call_args_list]
|
||||
auth_sent = any('SET auth' in c for c in calls)
|
||||
compression_sent = any('SET compression=0' in c for c in calls)
|
||||
mod_sent = any('SET mod=am' in c and 'freq=7000' in c for c in calls)
|
||||
assert auth_sent, "Auth message not sent"
|
||||
assert compression_sent, "Compression message not sent"
|
||||
assert mod_sent, "Tune message not sent"
|
||||
|
||||
# Cleanup
|
||||
client.disconnect()
|
||||
|
||||
|
||||
@patch('utils.kiwisdr.WEBSOCKET_CLIENT_AVAILABLE', True)
|
||||
@patch('utils.kiwisdr.websocket')
|
||||
def test_client_connect_failure(mock_ws_module):
|
||||
"""Connect should handle connection failures."""
|
||||
mock_ws = MagicMock()
|
||||
mock_ws.connect.side_effect = ConnectionRefusedError("Connection refused")
|
||||
mock_ws_module.WebSocket.return_value = mock_ws
|
||||
|
||||
client = KiwiSDRClient(host='unreachable.local', port=8073)
|
||||
result = client.connect(7000, 'am')
|
||||
|
||||
assert result is False
|
||||
assert client.connected is False
|
||||
|
||||
|
||||
@patch('utils.kiwisdr.WEBSOCKET_CLIENT_AVAILABLE', True)
|
||||
@patch('utils.kiwisdr.websocket')
|
||||
def test_client_tune_success(mock_ws_module):
|
||||
"""Tune should send the correct SET mod command."""
|
||||
mock_ws = MagicMock()
|
||||
mock_ws_module.WebSocket.return_value = mock_ws
|
||||
|
||||
client = KiwiSDRClient(host='kiwi.local', port=8073)
|
||||
client.connect(7000, 'am')
|
||||
|
||||
mock_ws.send.reset_mock()
|
||||
result = client.tune(14000, 'usb')
|
||||
|
||||
assert result is True
|
||||
assert client.frequency_khz == 14000
|
||||
assert client.mode == 'usb'
|
||||
|
||||
tune_calls = [str(c) for c in mock_ws.send.call_args_list]
|
||||
assert any('SET mod=usb' in c and 'freq=14000' in c for c in tune_calls)
|
||||
|
||||
client.disconnect()
|
||||
|
||||
|
||||
@patch('utils.kiwisdr.WEBSOCKET_CLIENT_AVAILABLE', True)
|
||||
@patch('utils.kiwisdr.websocket')
|
||||
def test_client_invalid_mode_fallback(mock_ws_module):
|
||||
"""Connect with invalid mode should fall back to AM."""
|
||||
mock_ws = MagicMock()
|
||||
mock_ws_module.WebSocket.return_value = mock_ws
|
||||
|
||||
client = KiwiSDRClient(host='kiwi.local', port=8073)
|
||||
client.connect(7000, 'invalid_mode')
|
||||
|
||||
assert client.mode == 'am'
|
||||
client.disconnect()
|
||||
|
||||
|
||||
@patch('utils.kiwisdr.WEBSOCKET_CLIENT_AVAILABLE', True)
|
||||
@patch('utils.kiwisdr.websocket')
|
||||
def test_client_ws_url_format(mock_ws_module):
|
||||
"""WebSocket URL should follow KiwiSDR format."""
|
||||
mock_ws = MagicMock()
|
||||
mock_ws_module.WebSocket.return_value = mock_ws
|
||||
|
||||
client = KiwiSDRClient(host='test.kiwi.com', port=8074)
|
||||
client.connect(7000, 'am')
|
||||
|
||||
ws_url = mock_ws.connect.call_args[0][0]
|
||||
assert ws_url.startswith('ws://test.kiwi.com:8074/')
|
||||
assert ws_url.endswith('/SND')
|
||||
|
||||
client.disconnect()
|
||||
@@ -0,0 +1,100 @@
|
||||
"""Tests for the Signal Identification (guess) API endpoint."""
|
||||
|
||||
import pytest
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def auth_client(client):
|
||||
"""Client with logged-in session."""
|
||||
with client.session_transaction() as sess:
|
||||
sess['logged_in'] = True
|
||||
return client
|
||||
|
||||
|
||||
def test_signal_guess_fm_broadcast(auth_client):
|
||||
"""FM broadcast frequency should return a known signal type."""
|
||||
resp = auth_client.post('/listening/signal/guess', json={
|
||||
'frequency_mhz': 98.1,
|
||||
'modulation': 'wfm',
|
||||
})
|
||||
assert resp.status_code == 200
|
||||
data = resp.get_json()
|
||||
assert data['status'] == 'ok'
|
||||
assert data['primary_label']
|
||||
assert data['confidence'] in ('HIGH', 'MEDIUM', 'LOW')
|
||||
|
||||
|
||||
def test_signal_guess_airband(auth_client):
|
||||
"""Airband frequency should be identified."""
|
||||
resp = auth_client.post('/listening/signal/guess', json={
|
||||
'frequency_mhz': 121.5,
|
||||
'modulation': 'am',
|
||||
})
|
||||
assert resp.status_code == 200
|
||||
data = resp.get_json()
|
||||
assert data['status'] == 'ok'
|
||||
assert data['primary_label']
|
||||
|
||||
|
||||
def test_signal_guess_ism_band(auth_client):
|
||||
"""ISM band frequency (433.92 MHz) should be identified."""
|
||||
resp = auth_client.post('/listening/signal/guess', json={
|
||||
'frequency_mhz': 433.92,
|
||||
})
|
||||
assert resp.status_code == 200
|
||||
data = resp.get_json()
|
||||
assert data['status'] == 'ok'
|
||||
assert data['primary_label']
|
||||
assert data['confidence'] in ('HIGH', 'MEDIUM', 'LOW')
|
||||
|
||||
|
||||
def test_signal_guess_missing_frequency(auth_client):
|
||||
"""Missing frequency should return 400."""
|
||||
resp = auth_client.post('/listening/signal/guess', json={})
|
||||
assert resp.status_code == 400
|
||||
data = resp.get_json()
|
||||
assert data['status'] == 'error'
|
||||
|
||||
|
||||
def test_signal_guess_invalid_frequency(auth_client):
|
||||
"""Invalid frequency value should return 400."""
|
||||
resp = auth_client.post('/listening/signal/guess', json={
|
||||
'frequency_mhz': 'abc',
|
||||
})
|
||||
assert resp.status_code == 400
|
||||
|
||||
|
||||
def test_signal_guess_negative_frequency(auth_client):
|
||||
"""Negative frequency should return 400."""
|
||||
resp = auth_client.post('/listening/signal/guess', json={
|
||||
'frequency_mhz': -5.0,
|
||||
})
|
||||
assert resp.status_code == 400
|
||||
|
||||
|
||||
def test_signal_guess_with_region(auth_client):
|
||||
"""Specifying region should work."""
|
||||
resp = auth_client.post('/listening/signal/guess', json={
|
||||
'frequency_mhz': 462.5625,
|
||||
'region': 'US',
|
||||
})
|
||||
assert resp.status_code == 200
|
||||
data = resp.get_json()
|
||||
assert data['status'] == 'ok'
|
||||
|
||||
|
||||
def test_signal_guess_response_structure(auth_client):
|
||||
"""Response should have all expected fields."""
|
||||
resp = auth_client.post('/listening/signal/guess', json={
|
||||
'frequency_mhz': 146.52,
|
||||
'modulation': 'fm',
|
||||
})
|
||||
assert resp.status_code == 200
|
||||
data = resp.get_json()
|
||||
assert 'primary_label' in data
|
||||
assert 'confidence' in data
|
||||
assert 'alternatives' in data
|
||||
assert 'explanation' in data
|
||||
assert 'tags' in data
|
||||
assert isinstance(data['alternatives'], list)
|
||||
assert isinstance(data['tags'], list)
|
||||
@@ -0,0 +1,80 @@
|
||||
"""Tests for the Waterfall / Spectrogram endpoints."""
|
||||
|
||||
from unittest.mock import patch, MagicMock
|
||||
import pytest
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def auth_client(client):
|
||||
"""Client with logged-in session."""
|
||||
with client.session_transaction() as sess:
|
||||
sess['logged_in'] = True
|
||||
return client
|
||||
|
||||
|
||||
def test_waterfall_start_no_rtl_power(auth_client):
|
||||
"""Start should fail gracefully when rtl_power is not available."""
|
||||
with patch('routes.listening_post.find_rtl_power', return_value=None):
|
||||
resp = auth_client.post('/listening/waterfall/start', json={
|
||||
'start_freq': 88.0,
|
||||
'end_freq': 108.0,
|
||||
})
|
||||
assert resp.status_code == 503
|
||||
data = resp.get_json()
|
||||
assert 'rtl_power' in data['message']
|
||||
|
||||
|
||||
def test_waterfall_start_invalid_range(auth_client):
|
||||
"""Start should reject end <= start."""
|
||||
with patch('routes.listening_post.find_rtl_power', return_value='/usr/bin/rtl_power'):
|
||||
resp = auth_client.post('/listening/waterfall/start', json={
|
||||
'start_freq': 108.0,
|
||||
'end_freq': 88.0,
|
||||
})
|
||||
assert resp.status_code == 400
|
||||
|
||||
|
||||
def test_waterfall_start_success(auth_client):
|
||||
"""Start should succeed with mocked rtl_power and device."""
|
||||
with patch('routes.listening_post.find_rtl_power', return_value='/usr/bin/rtl_power'), \
|
||||
patch('routes.listening_post.app_module') as mock_app:
|
||||
mock_app.claim_sdr_device.return_value = None # No error, claim succeeds
|
||||
resp = auth_client.post('/listening/waterfall/start', json={
|
||||
'start_freq': 88.0,
|
||||
'end_freq': 108.0,
|
||||
'gain': 40,
|
||||
'device': 0,
|
||||
})
|
||||
assert resp.status_code == 200
|
||||
data = resp.get_json()
|
||||
assert data['status'] == 'started'
|
||||
|
||||
# Clean up: stop waterfall
|
||||
import routes.listening_post as lp
|
||||
lp.waterfall_running = False
|
||||
|
||||
|
||||
def test_waterfall_stop(auth_client):
|
||||
"""Stop should succeed."""
|
||||
resp = auth_client.post('/listening/waterfall/stop')
|
||||
assert resp.status_code == 200
|
||||
data = resp.get_json()
|
||||
assert data['status'] == 'stopped'
|
||||
|
||||
|
||||
def test_waterfall_stream_mimetype(auth_client):
|
||||
"""Stream should return event-stream content type."""
|
||||
resp = auth_client.get('/listening/waterfall/stream')
|
||||
assert resp.content_type.startswith('text/event-stream')
|
||||
|
||||
|
||||
def test_waterfall_start_device_busy(auth_client):
|
||||
"""Start should fail when device is in use."""
|
||||
with patch('routes.listening_post.find_rtl_power', return_value='/usr/bin/rtl_power'), \
|
||||
patch('routes.listening_post.app_module') as mock_app:
|
||||
mock_app.claim_sdr_device.return_value = 'SDR device 0 is in use by scanner'
|
||||
resp = auth_client.post('/listening/waterfall/start', json={
|
||||
'start_freq': 88.0,
|
||||
'end_freq': 108.0,
|
||||
})
|
||||
assert resp.status_code == 409
|
||||
@@ -0,0 +1,170 @@
|
||||
"""Tests for the HF/Shortwave WebSDR integration."""
|
||||
|
||||
from unittest.mock import patch, MagicMock
|
||||
import pytest
|
||||
from routes.websdr import _parse_gps_coord, _haversine
|
||||
from utils.kiwisdr import parse_host_port
|
||||
|
||||
|
||||
# ============================================
|
||||
# Helper function tests
|
||||
# ============================================
|
||||
|
||||
def test_parse_gps_coord_float():
|
||||
"""Should parse a simple float string."""
|
||||
assert _parse_gps_coord('51.5074') == pytest.approx(51.5074)
|
||||
|
||||
|
||||
def test_parse_gps_coord_negative():
|
||||
"""Should parse a negative coordinate."""
|
||||
assert _parse_gps_coord('-33.87') == pytest.approx(-33.87)
|
||||
|
||||
|
||||
def test_parse_gps_coord_parentheses():
|
||||
"""Should handle parentheses in coordinate string."""
|
||||
assert _parse_gps_coord('(-33.87)') == pytest.approx(-33.87)
|
||||
|
||||
|
||||
def test_parse_gps_coord_empty():
|
||||
"""Should return None for empty string."""
|
||||
assert _parse_gps_coord('') is None
|
||||
assert _parse_gps_coord(None) is None
|
||||
|
||||
|
||||
def test_parse_gps_coord_invalid():
|
||||
"""Should return None for invalid string."""
|
||||
assert _parse_gps_coord('abc') is None
|
||||
|
||||
|
||||
def test_haversine_same_point():
|
||||
"""Distance between same point should be 0."""
|
||||
assert _haversine(51.5, -0.1, 51.5, -0.1) == pytest.approx(0.0, abs=0.01)
|
||||
|
||||
|
||||
def test_haversine_known_distance():
|
||||
"""Test with known city pair (London to Paris ~343 km)."""
|
||||
dist = _haversine(51.5074, -0.1278, 48.8566, 2.3522)
|
||||
assert 340 < dist < 350
|
||||
|
||||
|
||||
# ============================================
|
||||
# Endpoint tests
|
||||
# ============================================
|
||||
|
||||
@pytest.fixture
|
||||
def auth_client(client):
|
||||
"""Client with logged-in session."""
|
||||
with client.session_transaction() as sess:
|
||||
sess['logged_in'] = True
|
||||
return client
|
||||
|
||||
|
||||
def test_websdr_status(auth_client):
|
||||
"""Status endpoint should return cache info."""
|
||||
resp = auth_client.get('/websdr/status')
|
||||
assert resp.status_code == 200
|
||||
data = resp.get_json()
|
||||
assert data['status'] == 'ok'
|
||||
assert 'cached_receivers' in data
|
||||
|
||||
|
||||
def test_websdr_receivers_empty_cache(auth_client):
|
||||
"""Receivers endpoint should work even with empty cache."""
|
||||
with patch('routes.websdr.get_receivers', return_value=[]):
|
||||
resp = auth_client.get('/websdr/receivers')
|
||||
assert resp.status_code == 200
|
||||
data = resp.get_json()
|
||||
assert data['status'] == 'success'
|
||||
assert data['receivers'] == []
|
||||
|
||||
|
||||
def test_websdr_receivers_with_data(auth_client):
|
||||
"""Receivers endpoint should return filtered data."""
|
||||
mock_receivers = [
|
||||
{'name': 'Test RX', 'url': 'http://test.com', 'lat': 51.5, 'lon': -0.1,
|
||||
'users': 1, 'users_max': 4, 'available': True, 'freq_lo': 0, 'freq_hi': 30000,
|
||||
'antenna': 'Dipole', 'bands': 'HF'},
|
||||
{'name': 'Full RX', 'url': 'http://full.com', 'lat': 48.8, 'lon': 2.3,
|
||||
'users': 4, 'users_max': 4, 'available': False, 'freq_lo': 0, 'freq_hi': 30000,
|
||||
'antenna': 'Loop', 'bands': 'HF'},
|
||||
]
|
||||
with patch('routes.websdr.get_receivers', return_value=mock_receivers):
|
||||
# Filter available only
|
||||
resp = auth_client.get('/websdr/receivers?available=true')
|
||||
assert resp.status_code == 200
|
||||
data = resp.get_json()
|
||||
assert len(data['receivers']) == 1
|
||||
assert data['receivers'][0]['name'] == 'Test RX'
|
||||
|
||||
|
||||
def test_websdr_nearest_missing_params(auth_client):
|
||||
"""Nearest endpoint should require lat/lon."""
|
||||
resp = auth_client.get('/websdr/receivers/nearest')
|
||||
assert resp.status_code == 400
|
||||
|
||||
|
||||
def test_websdr_nearest_with_coords(auth_client):
|
||||
"""Nearest endpoint should sort by distance."""
|
||||
mock_receivers = [
|
||||
{'name': 'Far RX', 'url': 'http://far.com', 'lat': -33.87, 'lon': 151.21,
|
||||
'users': 0, 'users_max': 4, 'available': True, 'freq_lo': 0, 'freq_hi': 30000,
|
||||
'antenna': 'Dipole', 'bands': 'HF'},
|
||||
{'name': 'Near RX', 'url': 'http://near.com', 'lat': 51.0, 'lon': -0.5,
|
||||
'users': 0, 'users_max': 4, 'available': True, 'freq_lo': 0, 'freq_hi': 30000,
|
||||
'antenna': 'Loop', 'bands': 'HF'},
|
||||
]
|
||||
with patch('routes.websdr.get_receivers', return_value=mock_receivers):
|
||||
resp = auth_client.get('/websdr/receivers/nearest?lat=51.5&lon=-0.1')
|
||||
assert resp.status_code == 200
|
||||
data = resp.get_json()
|
||||
assert data['status'] == 'success'
|
||||
assert len(data['receivers']) == 2
|
||||
# Near should be first
|
||||
assert data['receivers'][0]['name'] == 'Near RX'
|
||||
|
||||
|
||||
def test_websdr_spy_station_receivers(auth_client):
|
||||
"""Spy station cross-reference should find matching receivers."""
|
||||
mock_receivers = [
|
||||
{'name': 'HF RX', 'url': 'http://hf.com', 'lat': 51.5, 'lon': -0.1,
|
||||
'users': 0, 'users_max': 4, 'available': True, 'freq_lo': 0, 'freq_hi': 30000,
|
||||
'antenna': 'Dipole', 'bands': 'HF'},
|
||||
]
|
||||
with patch('routes.websdr.get_receivers', return_value=mock_receivers):
|
||||
# e06 is one of the spy stations
|
||||
resp = auth_client.get('/websdr/spy-station/e06/receivers')
|
||||
assert resp.status_code == 200
|
||||
data = resp.get_json()
|
||||
assert data['status'] == 'success'
|
||||
assert 'station' in data
|
||||
|
||||
|
||||
def test_websdr_spy_station_not_found(auth_client):
|
||||
"""Non-existent station should return 404."""
|
||||
resp = auth_client.get('/websdr/spy-station/nonexistent/receivers')
|
||||
assert resp.status_code == 404
|
||||
|
||||
|
||||
# ============================================
|
||||
# parse_host_port tests (integration)
|
||||
# ============================================
|
||||
|
||||
def test_parse_host_port_http_url():
|
||||
"""Should parse standard KiwiSDR URL."""
|
||||
host, port = parse_host_port('http://kiwi.example.com:8073')
|
||||
assert host == 'kiwi.example.com'
|
||||
assert port == 8073
|
||||
|
||||
|
||||
def test_parse_host_port_no_protocol():
|
||||
"""Should handle bare hostname."""
|
||||
host, port = parse_host_port('my-kiwi.local:8074')
|
||||
assert host == 'my-kiwi.local'
|
||||
assert port == 8074
|
||||
|
||||
|
||||
def test_parse_host_port_with_trailing_slash():
|
||||
"""Should handle URL with trailing path."""
|
||||
host, port = parse_host_port('http://kiwi.com:8073/')
|
||||
assert host == 'kiwi.com'
|
||||
assert port == 8073
|
||||
Reference in New Issue
Block a user