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

fix: Make ADS-B history optional and add tests

Browse Source 
- Use Docker Compose profiles to make Postgres optional
- Default `docker compose up` runs without history/Postgres
- Use `docker compose --profile history up` to enable history
- Add 11 unit tests for AdsbHistoryWriter and AdsbSnapshotWriter

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
Smittix
2026-01-25 13:18:33 +00:00
parent 88537c1119
commit d297f87115
2 changed files with 319 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files

47
docker-compose.yml
View File

@@ -1,12 +1,16 @@
# INTERCEPT - Signal Intelligence Platform
# Docker Compose configuration for easy deployment
#
# Basic usage:
# docker compose up -d
#
# With ADS-B history (Postgres):
# docker compose --profile history up -d
services:
intercept:
build: .
container_name: intercept
depends_on:
- adsb_db
ports:
- "5050:5050"
# Privileged mode required for USB SDR device access
@@ -20,6 +24,40 @@ services:
# - ./data:/app/data
# Optional: mount logs directory
# - ./logs:/app/logs
environment:
- INTERCEPT_HOST=0.0.0.0
- INTERCEPT_PORT=5050
- INTERCEPT_LOG_LEVEL=INFO
# ADS-B history is disabled by default
# To enable, use: docker compose --profile history up -d
# - INTERCEPT_ADSB_HISTORY_ENABLED=true
# - INTERCEPT_ADSB_DB_HOST=adsb_db
# - INTERCEPT_ADSB_DB_PORT=5432
# - INTERCEPT_ADSB_DB_NAME=intercept_adsb
# - INTERCEPT_ADSB_DB_USER=intercept
# - INTERCEPT_ADSB_DB_PASSWORD=intercept
# Network mode for WiFi scanning (requires host network)
# network_mode: host
restart: unless-stopped
healthcheck:
test: ["CMD", "curl", "-sf", "http://localhost:5050/health"]
interval: 30s
timeout: 10s
retries: 3
start_period: 10s
# ADS-B history with Postgres persistence
# Enable with: docker compose --profile history up -d
intercept-history:
build: .
container_name: intercept
profiles:
- history
depends_on:
- adsb_db
ports:
- "5050:5050"
privileged: true
environment:
- INTERCEPT_HOST=0.0.0.0
- INTERCEPT_PORT=5050
@@ -30,8 +68,6 @@ services:
- INTERCEPT_ADSB_DB_NAME=intercept_adsb
- INTERCEPT_ADSB_DB_USER=intercept
- INTERCEPT_ADSB_DB_PASSWORD=intercept
# Network mode for WiFi scanning (requires host network)
# network_mode: host
restart: unless-stopped
healthcheck:
test: ["CMD", "curl", "-sf", "http://localhost:5050/health"]
@@ -43,12 +79,13 @@ services:
adsb_db:
image: postgres:16-alpine
container_name: intercept-adsb-db
profiles:
- history
environment:
- POSTGRES_DB=intercept_adsb
- POSTGRES_USER=intercept
- POSTGRES_PASSWORD=intercept
volumes:
# Move this path to the USB drive later for larger retention
- ./pgdata:/var/lib/postgresql/data
restart: unless-stopped
healthcheck:

277
tests/test_adsb_history.py Normal file
View File

@@ -0,0 +1,277 @@
"""Tests for ADS-B history persistence utilities."""
import queue
import threading
import time
from datetime import datetime, timezone
from unittest.mock import MagicMock, patch, PropertyMock
import pytest
class TestAdsbHistoryWriterUnit:
"""Unit tests for AdsbHistoryWriter (no database)."""
@pytest.fixture
def mock_config(self):
"""Mock config with history disabled."""
with patch.multiple(
'utils.adsb_history',
ADSB_HISTORY_ENABLED=False,
ADSB_DB_HOST='localhost',
ADSB_DB_PORT=5432,
ADSB_DB_NAME='test_db',
ADSB_DB_USER='test',
ADSB_DB_PASSWORD='test',
ADSB_HISTORY_BATCH_SIZE=100,
ADSB_HISTORY_FLUSH_INTERVAL=1.0,
ADSB_HISTORY_QUEUE_SIZE=1000,
):
yield
@pytest.fixture
def mock_config_enabled(self):
"""Mock config with history enabled."""
with patch.multiple(
'utils.adsb_history',
ADSB_HISTORY_ENABLED=True,
ADSB_DB_HOST='localhost',
ADSB_DB_PORT=5432,
ADSB_DB_NAME='test_db',
ADSB_DB_USER='test',
ADSB_DB_PASSWORD='test',
ADSB_HISTORY_BATCH_SIZE=100,
ADSB_HISTORY_FLUSH_INTERVAL=1.0,
ADSB_HISTORY_QUEUE_SIZE=1000,
):
yield
def test_writer_disabled_by_default(self, mock_config):
"""Test writer does nothing when disabled."""
from utils.adsb_history import AdsbHistoryWriter
writer = AdsbHistoryWriter()
writer.enabled = False
# Should not start thread
writer.start()
assert writer._thread is None
# Should not queue records
writer.enqueue({'icao': 'ABC123'})
assert writer._queue.empty()
def test_enqueue_adds_received_at(self, mock_config_enabled):
"""Test enqueue adds received_at timestamp if missing."""
from utils.adsb_history import AdsbHistoryWriter
writer = AdsbHistoryWriter()
writer.enabled = True
record = {'icao': 'ABC123'}
writer.enqueue(record)
# Record should have received_at added
assert 'received_at' in record
assert isinstance(record['received_at'], datetime)
def test_enqueue_preserves_existing_received_at(self, mock_config_enabled):
"""Test enqueue preserves existing received_at."""
from utils.adsb_history import AdsbHistoryWriter
writer = AdsbHistoryWriter()
writer.enabled = True
original_time = datetime(2025, 1, 1, 12, 0, 0, tzinfo=timezone.utc)
record = {'icao': 'ABC123', 'received_at': original_time}
writer.enqueue(record)
assert record['received_at'] == original_time
def test_enqueue_drops_when_queue_full(self, mock_config_enabled):
"""Test enqueue drops records when queue is full."""
from utils.adsb_history import AdsbHistoryWriter
writer = AdsbHistoryWriter()
writer.enabled = True
writer._queue = queue.Queue(maxsize=2)
# Fill the queue
writer.enqueue({'icao': 'A'})
writer.enqueue({'icao': 'B'})
# This should be dropped
writer.enqueue({'icao': 'C'})
assert writer._dropped == 1
assert writer._queue.qsize() == 2
class TestAdsbSnapshotWriterUnit:
"""Unit tests for AdsbSnapshotWriter (no database)."""
@pytest.fixture
def mock_config_enabled(self):
"""Mock config with history enabled."""
with patch.multiple(
'utils.adsb_history',
ADSB_HISTORY_ENABLED=True,
ADSB_DB_HOST='localhost',
ADSB_DB_PORT=5432,
ADSB_DB_NAME='test_db',
ADSB_DB_USER='test',
ADSB_DB_PASSWORD='test',
ADSB_HISTORY_BATCH_SIZE=100,
ADSB_HISTORY_FLUSH_INTERVAL=1.0,
ADSB_HISTORY_QUEUE_SIZE=1000,
):
yield
def test_snapshot_enqueue_adds_captured_at(self, mock_config_enabled):
"""Test enqueue adds captured_at timestamp if missing."""
from utils.adsb_history import AdsbSnapshotWriter
writer = AdsbSnapshotWriter()
writer.enabled = True
record = {'icao': 'ABC123'}
writer.enqueue(record)
assert 'captured_at' in record
assert isinstance(record['captured_at'], datetime)
class TestMakeDsn:
"""Tests for DSN generation."""
def test_make_dsn_format(self):
"""Test DSN string format."""
with patch.multiple(
'utils.adsb_history',
ADSB_DB_HOST='testhost',
ADSB_DB_PORT=5433,
ADSB_DB_NAME='testdb',
ADSB_DB_USER='testuser',
ADSB_DB_PASSWORD='testpass',
):
from utils.adsb_history import _make_dsn
dsn = _make_dsn()
assert 'host=testhost' in dsn
assert 'port=5433' in dsn
assert 'dbname=testdb' in dsn
assert 'user=testuser' in dsn
assert 'password=testpass' in dsn
class TestEnsureAdsbSchema:
"""Tests for schema creation."""
def test_ensure_schema_creates_tables(self):
"""Test schema creation SQL is executed."""
from utils.adsb_history import _ensure_adsb_schema
mock_conn = MagicMock()
mock_cursor = MagicMock()
mock_conn.cursor.return_value.__enter__ = MagicMock(return_value=mock_cursor)
mock_conn.cursor.return_value.__exit__ = MagicMock(return_value=False)
_ensure_adsb_schema(mock_conn)
# Should execute CREATE TABLE statements
assert mock_cursor.execute.call_count >= 3 # 3 tables + indexes
# Should commit
mock_conn.commit.assert_called_once()
def test_ensure_schema_creates_indexes(self):
"""Test schema creates required indexes."""
from utils.adsb_history import _ensure_adsb_schema
mock_conn = MagicMock()
mock_cursor = MagicMock()
mock_conn.cursor.return_value.__enter__ = MagicMock(return_value=mock_cursor)
mock_conn.cursor.return_value.__exit__ = MagicMock(return_value=False)
_ensure_adsb_schema(mock_conn)
# Get all executed SQL
executed_sql = [str(call) for call in mock_cursor.execute.call_args_list]
sql_text = ' '.join(executed_sql)
# Should create indexes
assert 'CREATE INDEX' in sql_text or 'idx_adsb' in sql_text
class TestMessageFields:
"""Tests for message field constants."""
def test_message_fields_exist(self):
"""Test required message fields are defined."""
from utils.adsb_history import _MESSAGE_FIELDS
required_fields = [
'received_at', 'icao', 'callsign', 'altitude',
'speed', 'heading', 'lat', 'lon', 'squawk'
]
for field in required_fields:
assert field in _MESSAGE_FIELDS
def test_snapshot_fields_exist(self):
"""Test required snapshot fields are defined."""
from utils.adsb_history import _SNAPSHOT_FIELDS
required_fields = [
'captured_at', 'icao', 'callsign', 'altitude',
'lat', 'lon', 'snapshot'
]
for field in required_fields:
assert field in _SNAPSHOT_FIELDS
class TestWriterThreadSafety:
"""Tests for thread safety of writers."""
def test_multiple_enqueue_thread_safe(self):
"""Test multiple threads can enqueue safely."""
with patch.multiple(
'utils.adsb_history',
ADSB_HISTORY_ENABLED=True,
ADSB_HISTORY_QUEUE_SIZE=10000,
ADSB_HISTORY_BATCH_SIZE=100,
ADSB_HISTORY_FLUSH_INTERVAL=1.0,
ADSB_DB_HOST='localhost',
ADSB_DB_PORT=5432,
ADSB_DB_NAME='test',
ADSB_DB_USER='test',
ADSB_DB_PASSWORD='test',
):
from utils.adsb_history import AdsbHistoryWriter
writer = AdsbHistoryWriter()
writer.enabled = True
errors = []
def enqueue_many(n):
try:
for i in range(n):
writer.enqueue({'icao': f'TEST{i}', 'altitude': i * 100})
except Exception as e:
errors.append(e)
threads = [
threading.Thread(target=enqueue_many, args=(100,))
for _ in range(5)
]
for t in threads:
t.start()
for t in threads:
t.join()
assert len(errors) == 0
# Should have queued 500 records (5 threads * 100 each)
assert writer._queue.qsize() == 500