Mirror/intercept
1
0
Fork 0
mirror of https://github.com/smittix/intercept.git synced 2026-07-06 08:38:14 -07:00
Code Issues Packages Projects Releases Wiki Activity
Files
96172ca593d51ef7abd2bc40f538cac114751e78
intercept/tests/test_agent_integration.py
T
James Smith 96172ca593 style: apply ruff-format to entire codebase 
First-time run of ruff-format via pre-commit hook normalises quote
style, trailing commas, and whitespace across 188 Python files.
No logic changes.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-07-05 14:48:11 +01:00

575 lines
21 KiB
Python
Raw Blame History

#!/usr/bin/env python3
"""
Integration tests for Intercept Agent with real tools.
These tests verify:
- Tool detection and availability
- Output parsing with sample/recorded data
- Live tool execution (optional, requires hardware)
Run with:
pytest tests/test_agent_integration.py -v
Run live tests (requires RTL-SDR hardware):
pytest tests/test_agent_integration.py -v -m live
Skip live tests:
pytest tests/test_agent_integration.py -v -m "not live"
"""
import json
import os
import shutil
import subprocess
import sys
import tempfile
import time
import pytest
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
# =============================================================================
# Sample Data for Parsing Tests
# =============================================================================
# Sample rtl_433 JSON outputs
RTL_433_SAMPLES = [
'{"time":"2024-01-15 10:30:00","model":"Acurite-Tower","id":12345,"channel":"A","battery_ok":1,"temperature_C":22.5,"humidity":45}',
'{"time":"2024-01-15 10:30:05","model":"Oregon-THGR122N","id":100,"channel":1,"battery_ok":1,"temperature_C":18.3,"humidity":62}',
'{"time":"2024-01-15 10:30:10","model":"LaCrosse-TX141W","id":55,"channel":2,"temperature_C":-5.2,"humidity":78}',
'{"time":"2024-01-15 10:30:15","model":"Ambient-F007TH","id":200,"channel":3,"temperature_C":25.0,"humidity":50,"battery_ok":1}',
]
# Sample SBS (BaseStation) format lines from dump1090
SBS_SAMPLES = [
"MSG,1,1,1,A1B2C3,1,2024/01/15,10:30:00.000,2024/01/15,10:30:00.000,UAL123,,,,,,,,,,0",
"MSG,3,1,1,A1B2C3,1,2024/01/15,10:30:01.000,2024/01/15,10:30:01.000,,35000,,,40.7128,-74.0060,,,0,0,0,0",
"MSG,4,1,1,A1B2C3,1,2024/01/15,10:30:02.000,2024/01/15,10:30:02.000,,,450,180,,,1500,,,,,",
"MSG,5,1,1,A1B2C3,1,2024/01/15,10:30:03.000,2024/01/15,10:30:03.000,UAL123,35000,,,,,,,,,",
"MSG,6,1,1,A1B2C3,1,2024/01/15,10:30:04.000,2024/01/15,10:30:04.000,,,,,,,,,,1200",
# Second aircraft
"MSG,1,1,1,D4E5F6,1,2024/01/15,10:30:05.000,2024/01/15,10:30:05.000,DAL456,,,,,,,,,,0",
"MSG,3,1,1,D4E5F6,1,2024/01/15,10:30:06.000,2024/01/15,10:30:06.000,,28000,,,40.8000,-73.9500,,,0,0,0,0",
]
# Sample airodump-ng CSV output (matches real airodump format - no blank line between header and data)
AIRODUMP_CSV_SAMPLE = """BSSID, First time seen, Last time seen, channel, Speed, Privacy, Cipher, Authentication, Power, # beacons, # IV, LAN IP, ID-length, ESSID, Key
00:11:22:33:44:55, 2024-01-15 10:00:00, 2024-01-15 10:30:00, 6, 54, WPA2, CCMP, PSK, -55, 100, 0, 0. 0. 0. 0, 8, HomeWiFi,
AA:BB:CC:DD:EE:FF, 2024-01-15 10:05:00, 2024-01-15 10:30:00, 11, 130, WPA2, CCMP, PSK, -70, 200, 0, 0. 0. 0. 0, 12, CoffeeShop,
11:22:33:44:55:66, 2024-01-15 10:10:00, 2024-01-15 10:30:00, 36, 867, WPA3, CCMP, SAE, -45, 150, 0, 0. 0. 0. 0, 7, Office5G,
Station MAC, First time seen, Last time seen, Power, # packets, BSSID, Probed ESSIDs
CA:FE:BA:BE:00:01, 2024-01-15 10:15:00, 2024-01-15 10:30:00, -60, 50, 00:11:22:33:44:55, HomeWiFi
DE:AD:BE:EF:00:02, 2024-01-15 10:20:00, 2024-01-15 10:30:00, -75, 25, AA:BB:CC:DD:EE:FF, CoffeeShop
"""
# =============================================================================
# Fixtures
# =============================================================================
@pytest.fixture
def agent():
"""Create a ModeManager instance for testing."""
from intercept_agent import ModeManager
return ModeManager()
@pytest.fixture
def temp_csv_file():
"""Create a temp airodump CSV file."""
with tempfile.NamedTemporaryFile(mode="w", suffix="-01.csv", delete=False) as f:
f.write(AIRODUMP_CSV_SAMPLE)
path = f.name
yield path[:-7] # Return base path without -01.csv suffix
# Cleanup
if os.path.exists(path):
os.unlink(path)
# =============================================================================
# Tool Detection Tests
# =============================================================================
class TestToolDetection:
"""Tests for tool availability detection."""
def test_rtl_433_available(self):
"""rtl_433 should be installed."""
assert shutil.which("rtl_433") is not None
def test_dump1090_available(self):
"""dump1090 should be installed."""
assert (
shutil.which("dump1090") is not None
or shutil.which("dump1090-fa") is not None
or shutil.which("readsb") is not None
)
def test_airodump_available(self):
"""airodump-ng should be installed."""
assert shutil.which("airodump-ng") is not None
def test_multimon_available(self):
"""multimon-ng should be installed."""
assert shutil.which("multimon-ng") is not None
def test_acarsdec_available(self):
"""acarsdec should be installed."""
assert shutil.which("acarsdec") is not None
def test_agent_detects_tools(self, agent):
"""Agent should detect available tools."""
caps = agent.detect_capabilities()
# These should all be True given the tools are installed
assert caps["modes"]["sensor"] is True
assert caps["modes"]["adsb"] is True
# wifi requires airmon-ng too
# bluetooth requires bluetoothctl
class TestRTLSDRDetection:
"""Tests for RTL-SDR hardware detection."""
def test_rtl_test_runs(self):
"""rtl_test should run (even if no device)."""
result = subprocess.run(["rtl_test", "-t"], capture_output=True, timeout=5)
# Will return 0 if device found, non-zero if not
# We just verify it runs without crashing
assert result.returncode in [0, 1, 255]
def test_agent_detects_sdr_devices(self, agent):
"""Agent should detect SDR devices."""
caps = agent.detect_capabilities()
# If RTL-SDR is connected, devices list should be non-empty
# This is hardware-dependent, so we just verify the key exists
assert "devices" in caps
@pytest.mark.live
def test_rtl_sdr_present(self):
"""Verify RTL-SDR device is present (for live tests)."""
result = subprocess.run(["rtl_test", "-t"], capture_output=True, timeout=5)
if b"Found 0 device" in result.stdout or b"No supported devices found" in result.stderr:
pytest.skip("No RTL-SDR device connected")
assert b"Found" in result.stdout
# =============================================================================
# Parsing Tests (No Hardware Required)
# =============================================================================
class TestRTL433Parsing:
"""Tests for rtl_433 JSON output parsing."""
def test_parse_acurite_sensor(self):
"""Parse Acurite temperature sensor data."""
data = json.loads(RTL_433_SAMPLES[0])
assert data["model"] == "Acurite-Tower"
assert data["id"] == 12345
assert data["temperature_C"] == 22.5
assert data["humidity"] == 45
assert data["battery_ok"] == 1
def test_parse_oregon_sensor(self):
"""Parse Oregon Scientific sensor data."""
data = json.loads(RTL_433_SAMPLES[1])
assert data["model"] == "Oregon-THGR122N"
assert data["temperature_C"] == 18.3
def test_parse_negative_temperature(self):
"""Parse sensor with negative temperature."""
data = json.loads(RTL_433_SAMPLES[2])
assert data["model"] == "LaCrosse-TX141W"
assert data["temperature_C"] == -5.2
def test_agent_sensor_data_format(self, agent):
"""Agent should format sensor data correctly for controller."""
# Simulate processing
sample = json.loads(RTL_433_SAMPLES[0])
sample["type"] = "sensor"
sample["received_at"] = "2024-01-15T10:30:00Z"
# Verify required fields for controller
assert "model" in sample
assert "temperature_C" in sample or "temperature_F" in sample
assert "received_at" in sample
class TestSBSParsing:
"""Tests for SBS (BaseStation) format parsing from dump1090."""
def test_parse_msg1_callsign(self, agent):
"""MSG,1 should extract callsign."""
line = SBS_SAMPLES[0]
agent._parse_sbs_line(line)
aircraft = agent.adsb_aircraft.get("A1B2C3")
assert aircraft is not None
assert aircraft["callsign"] == "UAL123"
def test_parse_msg3_position(self, agent):
"""MSG,3 should extract altitude and position."""
agent._parse_sbs_line(SBS_SAMPLES[0]) # First need MSG,1 for ICAO
agent._parse_sbs_line(SBS_SAMPLES[1])
aircraft = agent.adsb_aircraft.get("A1B2C3")
assert aircraft is not None
assert aircraft["altitude"] == 35000
assert abs(aircraft["lat"] - 40.7128) < 0.0001
assert abs(aircraft["lon"] - (-74.0060)) < 0.0001
def test_parse_msg4_velocity(self, agent):
"""MSG,4 should extract speed and heading."""
agent._parse_sbs_line(SBS_SAMPLES[0])
agent._parse_sbs_line(SBS_SAMPLES[2])
aircraft = agent.adsb_aircraft.get("A1B2C3")
assert aircraft is not None
assert aircraft["speed"] == 450
assert aircraft["heading"] == 180
assert aircraft["vertical_rate"] == 1500
def test_parse_msg6_squawk(self, agent):
"""MSG,6 should extract squawk code."""
agent._parse_sbs_line(SBS_SAMPLES[0])
agent._parse_sbs_line(SBS_SAMPLES[4])
aircraft = agent.adsb_aircraft.get("A1B2C3")
assert aircraft is not None
# Squawk may not be present if MSG,6 format doesn't have enough fields
# The sample line may need adjustment - check if squawk was parsed
if "squawk" in aircraft:
assert aircraft["squawk"] == "1200"
def test_parse_multiple_aircraft(self, agent):
"""Should track multiple aircraft simultaneously."""
for line in SBS_SAMPLES:
agent._parse_sbs_line(line)
assert "A1B2C3" in agent.adsb_aircraft
assert "D4E5F6" in agent.adsb_aircraft
assert agent.adsb_aircraft["D4E5F6"]["callsign"] == "DAL456"
def test_parse_malformed_sbs(self, agent):
"""Should handle malformed SBS lines gracefully."""
# Too few fields
agent._parse_sbs_line("MSG,1,1")
# Not MSG type
agent._parse_sbs_line("SEL,1,1,1,ABC123,1")
# Empty line
agent._parse_sbs_line("")
# Garbage
agent._parse_sbs_line("not,valid,sbs,data")
# Should not crash, aircraft dict should be empty
assert len(agent.adsb_aircraft) == 0
class TestAirodumpParsing:
"""Tests for airodump-ng CSV parsing using Intercept's parser."""
def test_intercept_parser_available(self):
"""Intercept's airodump parser should be importable."""
from utils.wifi.parsers.airodump import parse_airodump_csv
assert callable(parse_airodump_csv)
def test_parse_csv_networks_with_intercept_parser(self, temp_csv_file):
"""Intercept parser should parse network section of CSV."""
from utils.wifi.parsers.airodump import parse_airodump_csv
networks, clients = parse_airodump_csv(temp_csv_file + "-01.csv")
assert len(networks) >= 3
# Find HomeWiFi network by BSSID
home_wifi = next((n for n in networks if n.bssid == "00:11:22:33:44:55"), None)
assert home_wifi is not None
assert home_wifi.essid == "HomeWiFi"
assert home_wifi.channel == 6
assert home_wifi.rssi == -55
assert "WPA2" in home_wifi.security # Could be 'WPA2' or 'WPA/WPA2'
def test_parse_csv_clients_with_intercept_parser(self, temp_csv_file):
"""Intercept parser should parse client section of CSV."""
from utils.wifi.parsers.airodump import parse_airodump_csv
networks, clients = parse_airodump_csv(temp_csv_file + "-01.csv")
assert len(clients) >= 2
# Client should have MAC and associated BSSID
assert any(c.get("mac") == "CA:FE:BA:BE:00:01" for c in clients)
def test_agent_uses_intercept_parser(self, agent, temp_csv_file):
"""Agent should use Intercept's parser when available."""
networks, clients = agent._parse_airodump_csv(temp_csv_file + "-01.csv", None)
# Should return dict format
assert isinstance(networks, dict)
assert len(networks) >= 3
# Check a network entry
home_wifi = networks.get("00:11:22:33:44:55")
assert home_wifi is not None
assert home_wifi["essid"] == "HomeWiFi"
assert home_wifi["channel"] == 6
def test_parse_csv_clients(self, agent, temp_csv_file):
"""Agent should parse clients correctly."""
networks, clients = agent._parse_airodump_csv(temp_csv_file + "-01.csv", None)
assert len(clients) >= 2
# =============================================================================
# Live Tool Tests (Require Hardware)
# =============================================================================
@pytest.mark.live
class TestLiveRTL433:
"""Live tests with rtl_433 (requires RTL-SDR)."""
def test_rtl_433_runs(self):
"""rtl_433 should start and produce output."""
proc = subprocess.Popen(
["rtl_433", "-F", "json", "-T", "3"], # Run for 3 seconds
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
)
try:
stdout, stderr = proc.communicate(timeout=10)
# rtl_433 may or may not receive data in 3 seconds
# We just verify it starts without error
assert proc.returncode in [0, 1] # 1 = no data received, OK
except subprocess.TimeoutExpired:
proc.kill()
pytest.fail("rtl_433 did not complete in time")
def test_rtl_433_json_output(self):
"""rtl_433 JSON output should be parseable."""
proc = subprocess.Popen(["rtl_433", "-F", "json", "-T", "5"], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
try:
stdout, _ = proc.communicate(timeout=10)
# If we got any output, verify it's valid JSON
for line in stdout.decode("utf-8", errors="ignore").split("\n"):
line = line.strip()
if line:
try:
data = json.loads(line)
assert "model" in data or "time" in data
except json.JSONDecodeError:
pass # May be startup messages
except subprocess.TimeoutExpired:
proc.kill()
@pytest.mark.live
class TestLiveDump1090:
"""Live tests with dump1090 (requires RTL-SDR)."""
def test_dump1090_starts(self):
"""dump1090 should start successfully."""
dump1090_path = shutil.which("dump1090") or shutil.which("dump1090-fa")
if not dump1090_path:
pytest.skip("dump1090 not installed")
proc = subprocess.Popen([dump1090_path, "--net", "--quiet"], stdout=subprocess.DEVNULL, stderr=subprocess.PIPE)
try:
time.sleep(2)
if proc.poll() is not None:
stderr = proc.stderr.read().decode()
if "No supported RTLSDR devices found" in stderr:
pytest.skip("No RTL-SDR for ADS-B")
pytest.fail(f"dump1090 exited: {stderr}")
# Verify SBS port is open
import socket
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
result = sock.connect_ex(("localhost", 30003))
sock.close()
assert result == 0, "SBS port 30003 not open"
finally:
proc.terminate()
proc.wait()
@pytest.mark.live
class TestLiveAgentModes:
"""Live tests running agent modes (requires hardware)."""
def test_agent_sensor_mode(self, agent):
"""Agent should start and stop sensor mode."""
result = agent.start_mode("sensor", {})
if result.get("status") == "error":
if "not found" in result.get("message", ""):
pytest.skip("rtl_433 not found")
if "device" in result.get("message", "").lower():
pytest.skip("No RTL-SDR device")
assert result["status"] == "started"
assert "sensor" in agent.running_modes
# Let it run briefly
time.sleep(2)
# Check status
status = agent.get_mode_status("sensor")
assert status["running"] is True
# Stop
stop_result = agent.stop_mode("sensor")
assert stop_result["status"] == "stopped"
assert "sensor" not in agent.running_modes
def test_agent_adsb_mode(self, agent):
"""Agent should start and stop ADS-B mode."""
result = agent.start_mode("adsb", {})
if result.get("status") == "error":
if "not found" in result.get("message", ""):
pytest.skip("dump1090 not found")
if "device" in result.get("message", "").lower():
pytest.skip("No RTL-SDR device")
assert result["status"] == "started"
# Let it run briefly
time.sleep(3)
# Get data (may be empty if no aircraft)
data = agent.get_mode_data("adsb")
assert "data" in data
# Stop
agent.stop_mode("adsb")
# =============================================================================
# Controller Integration Tests
# =============================================================================
class TestAgentControllerFormat:
"""Tests that agent output matches controller expectations."""
def test_sensor_data_format(self, agent):
"""Sensor data should have required fields for controller."""
# Simulate parsed data
sample = {
"model": "Acurite-Tower",
"id": 12345,
"temperature_C": 22.5,
"humidity": 45,
"type": "sensor",
"received_at": "2024-01-15T10:30:00Z",
}
# Should be serializable
json_str = json.dumps(sample)
restored = json.loads(json_str)
assert restored["model"] == "Acurite-Tower"
def test_adsb_data_format(self, agent):
"""ADS-B data should have required fields for controller."""
# Simulate parsed aircraft
agent._parse_sbs_line(SBS_SAMPLES[0])
agent._parse_sbs_line(SBS_SAMPLES[1])
agent._parse_sbs_line(SBS_SAMPLES[2])
data = agent.get_mode_data("adsb")
# Should be list format
assert isinstance(data["data"], list)
if data["data"]:
aircraft = data["data"][0]
assert "icao" in aircraft
assert "last_seen" in aircraft
def test_push_payload_format(self, agent):
"""Push payload should match controller ingest format."""
# Simulate what agent sends to controller
payload = {
"agent_name": "test-sensor",
"scan_type": "adsb",
"interface": "rtlsdr0",
"payload": {"aircraft": [{"icao": "A1B2C3", "callsign": "UAL123", "altitude": 35000}]},
"received_at": "2024-01-15T10:30:00Z",
}
# Verify structure
assert "agent_name" in payload
assert "scan_type" in payload
assert "payload" in payload
# Should be JSON serializable
json_str = json.dumps(payload)
assert len(json_str) > 0
# =============================================================================
# GPS Integration Tests
# =============================================================================
class TestGPSIntegration:
"""Tests for GPS data in agent output."""
def test_data_includes_gps_field(self, agent):
"""Data should include GPS position if available."""
data = agent.get_mode_data("sensor")
# agent_gps field should exist (may be None if no GPS)
assert "agent_gps" in data or data.get("agent_gps") is None
def test_gps_position_format(self):
"""GPS position should have lat/lon fields."""
from intercept_agent import GPSManager
gps = GPSManager()
# Simulate position
class MockPosition:
latitude = 40.7128
longitude = -74.0060
altitude = 10.0
speed = 0.0
heading = 0.0
fix_quality = 2
gps._position = MockPosition()
pos = gps.position
assert pos is not None
assert "lat" in pos
assert "lon" in pos
assert pos["lat"] == 40.7128
assert pos["lon"] == -74.0060
# =============================================================================
# Run Tests
# =============================================================================
if __name__ == "__main__":
pytest.main([__file__, "-v", "-m", "not live"])
Reference in New Issue View Git Blame Copy Permalink