Mirror/intercept
1
0
Fork 0
mirror of https://github.com/smittix/intercept.git synced 2026-04-26 07:40:01 -07:00
Code Issues Packages Projects Releases Wiki Activity

feat: ship waterfall receiver overhaul and platform mode updates

Browse Source
This commit is contained in:
Smittix
2026-02-22 23:22:37 +00:00
parent 5d4b61b4c3
commit 5f480caa3f
41 changed files with 7635 additions and 3516 deletions
Download Patch File Download Diff File Expand all files Collapse all files

202
tests/test_analytics.py
View File

@@ -1,202 +0,0 @@
"""Tests for analytics endpoints, export, and squawk detection."""
import json
import tempfile
from pathlib import Path
from unittest.mock import MagicMock, patch
import pytest
@pytest.fixture(scope='session')
def app():
"""Create application for testing."""
import app as app_module
import utils.database as db_mod
from routes import register_blueprints
app_module.app.config['TESTING'] = True
# Use temp directory for test database
tmp_dir = Path(tempfile.mkdtemp())
db_mod.DB_DIR = tmp_dir
db_mod.DB_PATH = tmp_dir / 'test_intercept.db'
# Reset thread-local connection so it picks up new path
if hasattr(db_mod._local, 'connection') and db_mod._local.connection:
db_mod._local.connection.close()
db_mod._local.connection = None
db_mod.init_db()
if 'pager' not in app_module.app.blueprints:
register_blueprints(app_module.app)
return app_module.app
@pytest.fixture
def client(app):
client = app.test_client()
# Set session login to bypass require_login before_request hook
with client.session_transaction() as sess:
sess['logged_in'] = True
return client
class TestAnalyticsSummary:
"""Tests for /analytics/summary endpoint."""
def test_summary_returns_json(self, client):
response = client.get('/analytics/summary')
assert response.status_code == 200
data = json.loads(response.data)
assert data['status'] == 'success'
assert 'counts' in data
assert 'health' in data
assert 'squawks' in data
def test_summary_counts_structure(self, client):
response = client.get('/analytics/summary')
data = json.loads(response.data)
counts = data['counts']
assert 'adsb' in counts
assert 'ais' in counts
assert 'wifi' in counts
assert 'bluetooth' in counts
assert 'dsc' in counts
# All should be integers
for val in counts.values():
assert isinstance(val, int)
def test_summary_health_structure(self, client):
response = client.get('/analytics/summary')
data = json.loads(response.data)
health = data['health']
# Should have process statuses
assert 'pager' in health
assert 'sensor' in health
assert 'adsb' in health
# Each should have a running flag
for mode_info in health.values():
if isinstance(mode_info, dict) and 'running' in mode_info:
assert isinstance(mode_info['running'], bool)
class TestAnalyticsExport:
"""Tests for /analytics/export/<mode> endpoint."""
def test_export_adsb_json(self, client):
response = client.get('/analytics/export/adsb?format=json')
assert response.status_code == 200
data = json.loads(response.data)
assert data['status'] == 'success'
assert data['mode'] == 'adsb'
assert 'data' in data
assert isinstance(data['data'], list)
def test_export_adsb_csv(self, client):
response = client.get('/analytics/export/adsb?format=csv')
assert response.status_code == 200
assert response.content_type.startswith('text/csv')
assert 'Content-Disposition' in response.headers
def test_export_invalid_mode(self, client):
response = client.get('/analytics/export/invalid_mode')
assert response.status_code == 400
data = json.loads(response.data)
assert data['status'] == 'error'
def test_export_wifi_json(self, client):
response = client.get('/analytics/export/wifi?format=json')
assert response.status_code == 200
data = json.loads(response.data)
assert data['status'] == 'success'
assert data['mode'] == 'wifi'
class TestAnalyticsSquawks:
"""Tests for squawk detection."""
def test_squawks_endpoint(self, client):
response = client.get('/analytics/squawks')
assert response.status_code == 200
data = json.loads(response.data)
assert data['status'] == 'success'
assert isinstance(data['squawks'], list)
def test_get_emergency_squawks_detects_7700(self):
from utils.analytics import get_emergency_squawks
# Mock the adsb_aircraft DataStore
mock_store = MagicMock()
mock_store.items.return_value = [
('ABC123', {'squawk': '7700', 'callsign': 'TEST01', 'altitude': 35000}),
('DEF456', {'squawk': '1200', 'callsign': 'TEST02'}),
]
with patch('utils.analytics.app_module') as mock_app:
mock_app.adsb_aircraft = mock_store
squawks = get_emergency_squawks()
assert len(squawks) == 1
assert squawks[0]['squawk'] == '7700'
assert squawks[0]['meaning'] == 'General Emergency'
assert squawks[0]['icao'] == 'ABC123'
class TestGeofenceCRUD:
"""Tests for geofence CRUD endpoints."""
def test_list_geofences(self, client):
response = client.get('/analytics/geofences')
assert response.status_code == 200
data = json.loads(response.data)
assert data['status'] == 'success'
assert isinstance(data['zones'], list)
def test_create_geofence(self, client):
response = client.post('/analytics/geofences',
data=json.dumps({
'name': 'Test Zone',
'lat': 51.5074,
'lon': -0.1278,
'radius_m': 500,
}),
content_type='application/json')
assert response.status_code == 200
data = json.loads(response.data)
assert data['status'] == 'success'
assert 'zone_id' in data
def test_create_geofence_missing_fields(self, client):
response = client.post('/analytics/geofences',
data=json.dumps({'name': 'No coords'}),
content_type='application/json')
assert response.status_code == 400
def test_create_geofence_invalid_coords(self, client):
response = client.post('/analytics/geofences',
data=json.dumps({
'name': 'Bad',
'lat': 100,
'lon': 0,
'radius_m': 100,
}),
content_type='application/json')
assert response.status_code == 400
def test_delete_geofence_not_found(self, client):
response = client.delete('/analytics/geofences/99999')
assert response.status_code == 404
class TestAnalyticsActivity:
"""Tests for /analytics/activity endpoint."""
def test_activity_returns_sparklines(self, client):
response = client.get('/analytics/activity')
assert response.status_code == 200
data = json.loads(response.data)
assert data['status'] == 'success'
assert 'sparklines' in data
assert isinstance(data['sparklines'], dict)

46
tests/test_sdr_detection.py Normal file
View File

@@ -0,0 +1,46 @@
"""Tests for RTL-SDR detection parsing."""
from unittest.mock import MagicMock, patch
from utils.sdr.base import SDRType
from utils.sdr.detection import detect_rtlsdr_devices
@patch('utils.sdr.detection._check_tool', return_value=True)
@patch('utils.sdr.detection.subprocess.run')
def test_detect_rtlsdr_devices_filters_empty_serial_entries(mock_run, _mock_check_tool):
"""Ignore malformed rtl_test rows that have an empty SN field."""
mock_result = MagicMock()
mock_result.stdout = ""
mock_result.stderr = (
"Found 3 device(s):\n"
" 0: ??C?, , SN:\n"
" 1: ??C?, , SN:\n"
" 2: RTLSDRBlog, Blog V4, SN: 1\n"
)
mock_run.return_value = mock_result
devices = detect_rtlsdr_devices()
assert len(devices) == 1
assert devices[0].sdr_type == SDRType.RTL_SDR
assert devices[0].index == 2
assert devices[0].name == "RTLSDRBlog, Blog V4"
assert devices[0].serial == "1"
@patch('utils.sdr.detection._check_tool', return_value=True)
@patch('utils.sdr.detection.subprocess.run')
def test_detect_rtlsdr_devices_uses_replace_decode_mode(mock_run, _mock_check_tool):
"""Run rtl_test with tolerant decoding for malformed output bytes."""
mock_result = MagicMock()
mock_result.stdout = ""
mock_result.stderr = "Found 0 device(s):"
mock_run.return_value = mock_result
detect_rtlsdr_devices()
_, kwargs = mock_run.call_args
assert kwargs["text"] is True
assert kwargs["encoding"] == "utf-8"
assert kwargs["errors"] == "replace"

54
tests/test_waterfall_websocket.py Normal file
View File

@@ -0,0 +1,54 @@
"""Tests for waterfall WebSocket configuration helpers."""
from routes.waterfall_websocket import (
_parse_center_freq_mhz,
_parse_span_mhz,
_pick_sample_rate,
)
from utils.sdr import SDRType
from utils.sdr.base import SDRCapabilities
def _caps(sample_rates):
return SDRCapabilities(
sdr_type=SDRType.RTL_SDR,
freq_min_mhz=24.0,
freq_max_mhz=1766.0,
gain_min=0.0,
gain_max=49.6,
sample_rates=sample_rates,
supports_bias_t=True,
supports_ppm=True,
tx_capable=False,
)
def test_parse_center_prefers_center_freq_mhz():
assert _parse_center_freq_mhz({'center_freq_mhz': 162.55, 'center_freq': 144000000}) == 162.55
def test_parse_center_supports_center_freq_hz():
assert _parse_center_freq_mhz({'center_freq_hz': 915000000}) == 915.0
def test_parse_center_supports_legacy_hz_payload():
assert _parse_center_freq_mhz({'center_freq': 109000000}) == 109.0
def test_parse_center_supports_legacy_mhz_payload():
assert _parse_center_freq_mhz({'center_freq': 433.92}) == 433.92
def test_parse_span_from_hz_and_mhz():
assert _parse_span_mhz({'span_hz': 2400000}) == 2.4
assert _parse_span_mhz({'span_mhz': 10.0}) == 10.0
def test_pick_sample_rate_chooses_nearest_declared_rate():
caps = _caps([250000, 1024000, 1800000, 2048000, 2400000])
assert _pick_sample_rate(700000, caps, SDRType.RTL_SDR) == 1024000
def test_pick_sample_rate_falls_back_to_max_bandwidth():
caps = _caps([])
assert _pick_sample_rate(10_000_000, caps, SDRType.RTL_SDR) == 2_400_000