mirror of
https://github.com/smittix/intercept.git
synced 2026-04-25 07:10:00 -07:00
4e3f0ad800
- 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>
146 lines
4.2 KiB
Python
146 lines
4.2 KiB
Python
"""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')
|