mirror of
https://github.com/smittix/intercept.git
synced 2026-07-05 16:18:12 -07:00
Merge upstream/main and resolve adsb_dashboard.html conflict
Take upstream's crosshair animation system and updated selectAircraft(icao, source) signature. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
mitchross
@@ -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)
|
||||
@@ -0,0 +1,47 @@
|
||||
"""Tests for pager scope waveform payload generation."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import io
|
||||
import queue
|
||||
import struct
|
||||
import threading
|
||||
|
||||
from routes.pager import _encode_scope_waveform, audio_relay_thread
|
||||
|
||||
|
||||
def test_encode_scope_waveform_respects_window_and_range():
|
||||
samples = (-32768, -16384, 0, 16384, 32767)
|
||||
waveform = _encode_scope_waveform(samples, window_size=4)
|
||||
|
||||
assert len(waveform) == 4
|
||||
assert waveform[0] == -64
|
||||
assert waveform[1] == 0
|
||||
assert waveform[2] == 64
|
||||
assert waveform[3] == 127
|
||||
assert max(waveform) <= 127
|
||||
assert min(waveform) >= -127
|
||||
|
||||
|
||||
def test_audio_relay_thread_emits_scope_waveform(monkeypatch):
|
||||
base_samples = (0, 32767, -32768, 16384) * 512
|
||||
pcm = struct.pack(f"<{len(base_samples)}h", *base_samples)
|
||||
|
||||
rtl_stdout = io.BytesIO(pcm)
|
||||
multimon_stdin = io.BytesIO()
|
||||
output_queue: queue.Queue = queue.Queue()
|
||||
stop_event = threading.Event()
|
||||
|
||||
ticks = iter([0.0, 0.2, 0.2, 0.2])
|
||||
monkeypatch.setattr("routes.pager.time.monotonic", lambda: next(ticks, 0.2))
|
||||
|
||||
audio_relay_thread(rtl_stdout, multimon_stdin, output_queue, stop_event)
|
||||
|
||||
scope_event = output_queue.get_nowait()
|
||||
assert scope_event["type"] == "scope"
|
||||
assert scope_event["rms"] > 0
|
||||
assert scope_event["peak"] > 0
|
||||
assert "waveform" in scope_event
|
||||
assert len(scope_event["waveform"]) > 0
|
||||
assert max(scope_event["waveform"]) <= 127
|
||||
assert min(scope_event["waveform"]) >= -127
|
||||
@@ -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"
|
||||
@@ -0,0 +1,21 @@
|
||||
"""Tests for synthesized 433 MHz scope waveform payload."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from routes.sensor import _build_scope_waveform
|
||||
|
||||
|
||||
def test_build_scope_waveform_has_expected_shape_and_bounds():
|
||||
waveform = _build_scope_waveform(rssi=-8.5, snr=11.2, noise=-26.0, points=96)
|
||||
|
||||
assert len(waveform) == 96
|
||||
assert max(waveform) <= 127
|
||||
assert min(waveform) >= -127
|
||||
assert any(sample != 0 for sample in waveform)
|
||||
|
||||
|
||||
def test_build_scope_waveform_changes_with_signal_profile():
|
||||
low_snr = _build_scope_waveform(rssi=-14.0, snr=2.0, noise=-12.0, points=64)
|
||||
high_snr = _build_scope_waveform(rssi=-14.0, snr=20.0, noise=-12.0, points=64)
|
||||
|
||||
assert low_snr != high_snr
|
||||
@@ -0,0 +1,99 @@
|
||||
"""Tests for the SigID Wiki lookup API endpoint."""
|
||||
|
||||
from unittest.mock import patch
|
||||
|
||||
import pytest
|
||||
|
||||
import routes.signalid as signalid_module
|
||||
|
||||
|
||||
@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_sigidwiki_lookup_missing_frequency(auth_client):
|
||||
"""frequency_mhz is required."""
|
||||
resp = auth_client.post('/signalid/sigidwiki', json={})
|
||||
assert resp.status_code == 400
|
||||
data = resp.get_json()
|
||||
assert data['status'] == 'error'
|
||||
|
||||
|
||||
def test_sigidwiki_lookup_invalid_frequency(auth_client):
|
||||
"""frequency_mhz must be numeric and positive."""
|
||||
resp = auth_client.post('/signalid/sigidwiki', json={'frequency_mhz': 'abc'})
|
||||
assert resp.status_code == 400
|
||||
|
||||
resp = auth_client.post('/signalid/sigidwiki', json={'frequency_mhz': -1})
|
||||
assert resp.status_code == 400
|
||||
|
||||
|
||||
def test_sigidwiki_lookup_success(auth_client):
|
||||
"""Endpoint returns normalized SigID lookup structure."""
|
||||
signalid_module._cache.clear()
|
||||
fake_lookup = {
|
||||
'matches': [
|
||||
{
|
||||
'title': 'POCSAG',
|
||||
'url': 'https://www.sigidwiki.com/wiki/POCSAG',
|
||||
'frequencies_mhz': [929.6625],
|
||||
'modes': ['NFM'],
|
||||
'modulations': ['FSK'],
|
||||
'distance_hz': 0,
|
||||
'source': 'SigID Wiki',
|
||||
}
|
||||
],
|
||||
'search_used': False,
|
||||
'exact_queries': ['[[Category:Signal]][[Frequencies::929.6625 MHz]]|?Frequencies|?Mode|?Modulation|limit=10'],
|
||||
}
|
||||
|
||||
with patch('routes.signalid._lookup_sigidwiki_matches', return_value=fake_lookup) as lookup_mock:
|
||||
resp = auth_client.post('/signalid/sigidwiki', json={
|
||||
'frequency_mhz': 929.6625,
|
||||
'modulation': 'fm',
|
||||
'limit': 5,
|
||||
})
|
||||
|
||||
assert lookup_mock.call_count == 1
|
||||
assert resp.status_code == 200
|
||||
data = resp.get_json()
|
||||
assert data['status'] == 'ok'
|
||||
assert data['source'] == 'sigidwiki'
|
||||
assert data['cached'] is False
|
||||
assert data['match_count'] == 1
|
||||
assert data['matches'][0]['title'] == 'POCSAG'
|
||||
|
||||
|
||||
def test_sigidwiki_lookup_cached_response(auth_client):
|
||||
"""Second identical lookup should be served from cache."""
|
||||
signalid_module._cache.clear()
|
||||
fake_lookup = {
|
||||
'matches': [{'title': 'Test Signal', 'url': 'https://www.sigidwiki.com/wiki/Test_Signal'}],
|
||||
'search_used': True,
|
||||
'exact_queries': [],
|
||||
}
|
||||
|
||||
payload = {'frequency_mhz': 433.92, 'modulation': 'nfm', 'limit': 5}
|
||||
with patch('routes.signalid._lookup_sigidwiki_matches', return_value=fake_lookup) as lookup_mock:
|
||||
first = auth_client.post('/signalid/sigidwiki', json=payload)
|
||||
second = auth_client.post('/signalid/sigidwiki', json=payload)
|
||||
|
||||
assert lookup_mock.call_count == 1
|
||||
assert first.status_code == 200
|
||||
assert second.status_code == 200
|
||||
assert first.get_json()['cached'] is False
|
||||
assert second.get_json()['cached'] is True
|
||||
|
||||
|
||||
def test_sigidwiki_lookup_backend_failure(auth_client):
|
||||
"""Unexpected lookup failures should return 502."""
|
||||
signalid_module._cache.clear()
|
||||
with patch('routes.signalid._lookup_sigidwiki_matches', side_effect=RuntimeError('boom')):
|
||||
resp = auth_client.post('/signalid/sigidwiki', json={'frequency_mhz': 433.92})
|
||||
assert resp.status_code == 502
|
||||
data = resp.get_json()
|
||||
assert data['status'] == 'error'
|
||||
@@ -0,0 +1,25 @@
|
||||
"""Tests for SSTV scope waveform encoding."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import numpy as np
|
||||
|
||||
from utils.sstv.sstv_decoder import _encode_scope_waveform
|
||||
|
||||
|
||||
def test_encode_scope_waveform_respects_window_and_bounds():
|
||||
samples = np.array([-32768, -16384, 0, 16384, 32767], dtype=np.int16)
|
||||
waveform = _encode_scope_waveform(samples, window_size=4)
|
||||
|
||||
assert len(waveform) == 4
|
||||
assert waveform[0] == -64
|
||||
assert waveform[1] == 0
|
||||
assert waveform[2] == 64
|
||||
assert waveform[3] == 127
|
||||
assert max(waveform) <= 127
|
||||
assert min(waveform) >= -127
|
||||
|
||||
|
||||
def test_encode_scope_waveform_empty_input():
|
||||
waveform = _encode_scope_waveform(np.array([], dtype=np.int16))
|
||||
assert waveform == []
|
||||
@@ -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
|
||||
Reference in New Issue
Block a user