diff --git a/tests/test_meshtastic.py b/tests/test_meshtastic.py index 3ffbde0..ce7c00d 100644 --- a/tests/test_meshtastic.py +++ b/tests/test_meshtastic.py @@ -18,12 +18,14 @@ import pytest # Utility Module Tests # ============================================================================= + class TestMeshtasticAvailability: """Tests for SDK availability checks.""" def test_is_meshtastic_available_returns_bool(self): """is_meshtastic_available should return a boolean.""" from utils.meshtastic import is_meshtastic_available + result = is_meshtastic_available() assert isinstance(result, bool) @@ -36,10 +38,10 @@ class TestMeshtasticMessage: from utils.meshtastic import MeshtasticMessage msg = MeshtasticMessage( - from_id='!a1b2c3d4', - to_id='^all', - message='Hello mesh!', - portnum='TEXT_MESSAGE_APP', + from_id="!a1b2c3d4", + to_id="^all", + message="Hello mesh!", + portnum="TEXT_MESSAGE_APP", channel=0, rssi=-95, snr=-3.5, @@ -49,26 +51,28 @@ class TestMeshtasticMessage: d = msg.to_dict() - assert d['type'] == 'meshtastic' - assert d['from'] == '!a1b2c3d4' - assert d['to'] == '^all' - assert d['message'] == 'Hello mesh!' - assert d['portnum'] == 'TEXT_MESSAGE_APP' - assert d['channel'] == 0 - assert d['rssi'] == -95 - assert d['snr'] == -3.5 - assert d['hop_limit'] == 3 - assert '2026-01-27' in d['timestamp'] + assert d["type"] == "meshtastic" + assert d["from"] == "!a1b2c3d4" + assert d["to"] == "^all" + assert d["message"] == "Hello mesh!" + assert d["portnum"] == "TEXT_MESSAGE_APP" + assert d["channel"] == 0 + assert d["rssi"] == -95 + assert d["snr"] == -3.5 + assert d["hop_limit"] == 3 + assert isinstance(d["timestamp"], float) + # 2026-01-27 12:00:00 UTC as Unix epoch + assert d["timestamp"] == pytest.approx(1769515200.0) def test_message_with_none_values(self): """MeshtasticMessage should handle None values.""" from utils.meshtastic import MeshtasticMessage msg = MeshtasticMessage( - from_id='!00000001', - to_id='!00000002', + from_id="!00000001", + to_id="!00000002", message=None, - portnum='POSITION_APP', + portnum="POSITION_APP", channel=1, rssi=None, snr=None, @@ -78,9 +82,9 @@ class TestMeshtasticMessage: d = msg.to_dict() - assert d['message'] is None - assert d['rssi'] is None - assert d['snr'] is None + assert d["message"] is None + assert d["rssi"] is None + assert d["snr"] is None class TestChannelConfig: @@ -92,50 +96,50 @@ class TestChannelConfig: config = ChannelConfig( index=0, - name='Primary', - psk=b'\x01\x02\x03\x04' * 8, # 32-byte key + name="Primary", + psk=b"\x01\x02\x03\x04" * 8, # 32-byte key role=1, # PRIMARY ) d = config.to_dict() - assert 'psk' not in d # Raw PSK should not be in dict - assert d['index'] == 0 - assert d['name'] == 'Primary' - assert d['role'] == 'PRIMARY' - assert d['encrypted'] is True - assert d['key_type'] == 'AES-256' + assert "psk" not in d # Raw PSK should not be in dict + assert d["index"] == 0 + assert d["name"] == "Primary" + assert d["role"] == "PRIMARY" + assert d["encrypted"] is True + assert d["key_type"] == "AES-256" def test_channel_default_key_detection(self): """ChannelConfig should detect default key.""" from utils.meshtastic import ChannelConfig # Default key is single byte 0x01 - config = ChannelConfig(index=0, name='Test', psk=b'\x01', role=1) + config = ChannelConfig(index=0, name="Test", psk=b"\x01", role=1) d = config.to_dict() - assert d['is_default_key'] is True - assert d['key_type'] == 'default' + assert d["is_default_key"] is True + assert d["key_type"] == "default" def test_channel_aes128_detection(self): """ChannelConfig should detect AES-128 key.""" from utils.meshtastic import ChannelConfig - config = ChannelConfig(index=0, name='Test', psk=b'0' * 16, role=1) + config = ChannelConfig(index=0, name="Test", psk=b"0" * 16, role=1) d = config.to_dict() - assert d['key_type'] == 'AES-128' - assert d['encrypted'] is True + assert d["key_type"] == "AES-128" + assert d["encrypted"] is True def test_channel_no_encryption(self): """ChannelConfig should detect no encryption.""" from utils.meshtastic import ChannelConfig - config = ChannelConfig(index=0, name='Test', psk=b'', role=1) + config = ChannelConfig(index=0, name="Test", psk=b"", role=1) d = config.to_dict() - assert d['key_type'] == 'none' - assert d['encrypted'] is False + assert d["key_type"] == "none" + assert d["encrypted"] is False class TestPSKParsing: @@ -146,29 +150,29 @@ class TestPSKParsing: from utils.meshtastic import MeshtasticClient client = MeshtasticClient() - result = client._parse_psk('none') + result = client._parse_psk("none") - assert result == b'' + assert result == b"" def test_parse_psk_default(self): """Should parse 'default' as single byte.""" from utils.meshtastic import MeshtasticClient client = MeshtasticClient() - result = client._parse_psk('default') + result = client._parse_psk("default") - assert result == b'\x01' + assert result == b"\x01" def test_parse_psk_random(self): """Should generate 32 random bytes for 'random'.""" from utils.meshtastic import MeshtasticClient client = MeshtasticClient() - result = client._parse_psk('random') + result = client._parse_psk("random") assert len(result) == 32 # Verify it's actually random (two calls should differ) - result2 = client._parse_psk('random') + result2 = client._parse_psk("random") assert result != result2 def test_parse_psk_base64(self): @@ -179,8 +183,8 @@ class TestPSKParsing: client = MeshtasticClient() # 32-byte key encoded as base64 - key = b'A' * 32 - encoded = 'base64:' + base64.b64encode(key).decode() + key = b"A" * 32 + encoded = "base64:" + base64.b64encode(key).decode() result = client._parse_psk(encoded) @@ -192,9 +196,9 @@ class TestPSKParsing: client = MeshtasticClient() # 16-byte key as hex - result = client._parse_psk('0x' + '41' * 16) + result = client._parse_psk("0x" + "41" * 16) - assert result == b'A' * 16 + assert result == b"A" * 16 def test_parse_psk_simple_passphrase(self): """Should hash simple passphrase to 32-byte key.""" @@ -203,9 +207,9 @@ class TestPSKParsing: from utils.meshtastic import MeshtasticClient client = MeshtasticClient() - result = client._parse_psk('simple:MySecretPassword') + result = client._parse_psk("simple:MySecretPassword") - expected = hashlib.sha256(b'MySecretPassword').digest() + expected = hashlib.sha256(b"MySecretPassword").digest() assert result == expected assert len(result) == 32 @@ -215,8 +219,8 @@ class TestPSKParsing: client = MeshtasticClient() - assert client._parse_psk('base64:!!!invalid!!!') is None - assert client._parse_psk('0xZZZZ') is None + assert client._parse_psk("base64:!!!invalid!!!") is None + assert client._parse_psk("0xZZZZ") is None def test_parse_psk_raw_base64(self): """Should accept raw base64 without prefix.""" @@ -225,7 +229,7 @@ class TestPSKParsing: from utils.meshtastic import MeshtasticClient client = MeshtasticClient() - key = b'B' * 16 + key = b"B" * 16 encoded = base64.b64encode(key).decode() result = client._parse_psk(encoded) @@ -242,7 +246,7 @@ class TestNodeIdFormatting: result = MeshtasticClient._format_node_id(0xDEADBEEF) - assert result == '!deadbeef' + assert result == "!deadbeef" def test_format_broadcast(self): """Should format broadcast address.""" @@ -250,13 +254,14 @@ class TestNodeIdFormatting: result = MeshtasticClient._format_node_id(0xFFFFFFFF) - assert result == '^all' + assert result == "^all" # ============================================================================= # Route Tests (Mocked) # ============================================================================= + class TestMeshtasticRoutes: """Tests for Flask route endpoints.""" @@ -268,7 +273,7 @@ class TestMeshtasticRoutes: from routes.meshtastic import meshtastic_bp app = Flask(__name__) - app.config['TESTING'] = True + app.config["TESTING"] = True app.register_blueprint(meshtastic_bp) return app @@ -280,144 +285,137 @@ class TestMeshtasticRoutes: def test_status_sdk_not_installed(self, client): """GET /meshtastic/status should report SDK unavailable.""" - with patch('routes.meshtastic.is_meshtastic_available', return_value=False): - response = client.get('/meshtastic/status') + with patch("routes.meshtastic.is_meshtastic_available", return_value=False): + response = client.get("/meshtastic/status") data = json.loads(response.data) assert response.status_code == 200 - assert data['available'] is False - assert 'not installed' in data['error'] + assert data["available"] is False + assert "not installed" in data["error"] def test_status_not_connected(self, client): """GET /meshtastic/status should report not running when disconnected.""" - with patch('routes.meshtastic.is_meshtastic_available', return_value=True): - with patch('routes.meshtastic.get_meshtastic_client', return_value=None): - response = client.get('/meshtastic/status') + with patch("routes.meshtastic.is_meshtastic_available", return_value=True): + with patch("routes.meshtastic.get_meshtastic_client", return_value=None): + response = client.get("/meshtastic/status") data = json.loads(response.data) assert response.status_code == 200 - assert data['available'] is True - assert data['running'] is False + assert data["available"] is True + assert data["running"] is False def test_start_sdk_not_installed(self, client): """POST /meshtastic/start should fail if SDK not installed.""" - with patch('routes.meshtastic.is_meshtastic_available', return_value=False): - response = client.post('/meshtastic/start') + with patch("routes.meshtastic.is_meshtastic_available", return_value=False): + response = client.post("/meshtastic/start") data = json.loads(response.data) assert response.status_code == 400 - assert data['status'] == 'error' + assert data["status"] == "error" def test_stop_always_succeeds(self, client): """POST /meshtastic/stop should always succeed.""" - with patch('routes.meshtastic.stop_meshtastic'): - response = client.post('/meshtastic/stop') + with patch("routes.meshtastic.stop_meshtastic"): + response = client.post("/meshtastic/stop") data = json.loads(response.data) assert response.status_code == 200 - assert data['status'] == 'stopped' + assert data["status"] == "stopped" def test_channels_not_connected(self, client): """GET /meshtastic/channels should fail if not connected.""" - with patch('routes.meshtastic.get_meshtastic_client', return_value=None): - response = client.get('/meshtastic/channels') + with patch("routes.meshtastic.get_meshtastic_client", return_value=None): + response = client.get("/meshtastic/channels") data = json.loads(response.data) assert response.status_code == 400 - assert 'Not connected' in data['message'] + assert "Not connected" in data["message"] def test_configure_channel_invalid_index(self, client): """POST /meshtastic/channels/ should reject invalid index.""" mock_client = Mock() mock_client.is_running = True - with patch('routes.meshtastic.get_meshtastic_client', return_value=mock_client): - response = client.post( - '/meshtastic/channels/10', - json={'name': 'Test'}, - content_type='application/json' - ) + with patch("routes.meshtastic.get_meshtastic_client", return_value=mock_client): + response = client.post("/meshtastic/channels/10", json={"name": "Test"}, content_type="application/json") data = json.loads(response.data) assert response.status_code == 400 - assert 'must be 0-7' in data['message'] + assert "must be 0-7" in data["message"] def test_configure_channel_no_params(self, client): """POST /meshtastic/channels/ should require name or psk.""" mock_client = Mock() mock_client.is_running = True - with patch('routes.meshtastic.get_meshtastic_client', return_value=mock_client): - response = client.post( - '/meshtastic/channels/0', - json={}, - content_type='application/json' - ) + with patch("routes.meshtastic.get_meshtastic_client", return_value=mock_client): + response = client.post("/meshtastic/channels/0", json={}, content_type="application/json") data = json.loads(response.data) assert response.status_code == 400 - assert 'Must provide' in data['message'] + assert "Must provide" in data["message"] def test_messages_empty(self, client): """GET /meshtastic/messages should return empty list initially.""" - with patch('routes.meshtastic._recent_messages', []): - response = client.get('/meshtastic/messages') + with patch("routes.meshtastic._recent_messages", []): + response = client.get("/meshtastic/messages") data = json.loads(response.data) assert response.status_code == 200 - assert data['status'] == 'ok' - assert data['messages'] == [] - assert data['count'] == 0 + assert data["status"] == "ok" + assert data["messages"] == [] + assert data["count"] == 0 def test_messages_with_limit(self, client): """GET /meshtastic/messages should respect limit param.""" - test_messages = [{'id': i} for i in range(10)] + test_messages = [{"id": i} for i in range(10)] - with patch('routes.meshtastic._recent_messages', test_messages): - response = client.get('/meshtastic/messages?limit=3') + with patch("routes.meshtastic._recent_messages", test_messages): + response = client.get("/meshtastic/messages?limit=3") data = json.loads(response.data) assert response.status_code == 200 - assert len(data['messages']) == 3 + assert len(data["messages"]) == 3 # Should return last 3 (most recent) - assert data['messages'][0]['id'] == 7 + assert data["messages"][0]["id"] == 7 def test_messages_filter_by_channel(self, client): """GET /meshtastic/messages should filter by channel.""" test_messages = [ - {'id': 1, 'channel': 0}, - {'id': 2, 'channel': 1}, - {'id': 3, 'channel': 0}, + {"id": 1, "channel": 0}, + {"id": 2, "channel": 1}, + {"id": 3, "channel": 0}, ] - with patch('routes.meshtastic._recent_messages', test_messages): - response = client.get('/meshtastic/messages?channel=0') + with patch("routes.meshtastic._recent_messages", test_messages): + response = client.get("/meshtastic/messages?channel=0") data = json.loads(response.data) assert response.status_code == 200 - assert len(data['messages']) == 2 - assert all(m['channel'] == 0 for m in data['messages']) + assert len(data["messages"]) == 2 + assert all(m["channel"] == 0 for m in data["messages"]) def test_stream_endpoint_exists(self, client): """GET /meshtastic/stream should return SSE content type.""" - response = client.get('/meshtastic/stream') + response = client.get("/meshtastic/stream") - assert response.content_type == 'text/event-stream' + assert response.content_type.startswith("text/event-stream") def test_node_not_connected(self, client): """GET /meshtastic/node should fail if not connected.""" - with patch('routes.meshtastic.get_meshtastic_client', return_value=None): - response = client.get('/meshtastic/node') + with patch("routes.meshtastic.get_meshtastic_client", return_value=None): + response = client.get("/meshtastic/node") data = json.loads(response.data) assert response.status_code == 400 - assert 'Not connected' in data['message'] + assert "Not connected" in data["message"] # ============================================================================= # Integration Tests (Mocked SDK) # ============================================================================= + class TestMeshtasticClientMocked: """Tests for MeshtasticClient with mocked SDK.""" @@ -435,12 +433,12 @@ class TestMeshtasticClientMocked: """MeshtasticClient.connect should fail gracefully without SDK.""" from utils.meshtastic import MeshtasticClient - with patch('utils.meshtastic.HAS_MESHTASTIC', False): + with patch("utils.meshtastic.HAS_MESHTASTIC", False): client = MeshtasticClient() result = client.connect() assert result is False - assert 'not installed' in client.error + assert "not installed" in client.error def test_client_disconnect_idempotent(self): """MeshtasticClient.disconnect should be safe to call multiple times.""" diff --git a/tests/test_routes.py b/tests/test_routes.py index 5483786..bd0d20a 100644 --- a/tests/test_routes.py +++ b/tests/test_routes.py @@ -1,26 +1,26 @@ -"""Tests for Flask routes and API endpoints.""" - -import json -from unittest.mock import MagicMock, patch - -import pytest +"""Tests for Flask routes and API endpoints.""" + +import json +from unittest.mock import MagicMock, patch + +import pytest -@pytest.fixture(scope='session') +@pytest.fixture(scope="session") def app(): """Create application for testing.""" import app as app_module from routes import register_blueprints from utils.database import init_db - app_module.app.config['TESTING'] = True + app_module.app.config["TESTING"] = True # Initialize database for settings tests init_db() # Register blueprints only if not already registered (normally done in main()) # Check if any blueprint is already registered to avoid re-registration - if 'pager' not in app_module.app.blueprints: + if "pager" not in app_module.app.blueprints: register_blueprints(app_module.app) return app_module.app @@ -29,7 +29,10 @@ def app(): @pytest.fixture def client(app): """Create test client.""" - return app.test_client() + c = app.test_client() + with c.session_transaction() as sess: + sess["logged_in"] = True + return c class TestHealthEndpoint: @@ -37,55 +40,52 @@ class TestHealthEndpoint: def test_health_check(self, client): """Test health endpoint returns expected data.""" - response = client.get('/health') + response = client.get("/health") assert response.status_code == 200 data = json.loads(response.data) - assert data['status'] == 'healthy' - assert 'version' in data - assert 'uptime_seconds' in data - assert 'processes' in data - assert 'data' in data + assert data["status"] == "healthy" + assert "version" in data + assert "uptime_seconds" in data + assert "processes" in data + assert "data" in data - def test_health_process_status(self, client): - """Test health endpoint reports process status.""" - response = client.get('/health') - data = json.loads(response.data) + def test_health_process_status(self, client): + """Test health endpoint reports process status.""" + response = client.get("/health") + data = json.loads(response.data) - processes = data['processes'] - assert 'pager' in processes - assert 'sensor' in processes - assert 'adsb' in processes - assert 'wifi' in processes - assert 'bluetooth' in processes - -class TestDevicesEndpoint: + processes = data["processes"] + assert "pager" in processes + assert "sensor" in processes + assert "adsb" in processes + assert "wifi" in processes + assert "bluetooth" in processes + + +class TestDevicesEndpoint: """Tests for devices endpoint.""" def test_get_devices(self, client): """Test getting device list.""" - response = client.get('/devices') + response = client.get("/devices") assert response.status_code == 200 data = json.loads(response.data) assert isinstance(data, list) - @patch('app.SDRFactory.detect_devices') + @patch("app.SDRFactory.detect_devices") def test_devices_returns_list(self, mock_detect, client): """Test devices endpoint returns list format.""" mock_device = MagicMock() - mock_device.to_dict.return_value = { - 'index': 0, - 'name': 'Test RTL-SDR', - 'sdr_type': 'rtlsdr' - } + mock_device.to_dict.return_value = {"index": 0, "name": "Test RTL-SDR", "sdr_type": "rtlsdr"} mock_detect.return_value = [mock_device] - response = client.get('/devices') + response = client.get("/devices") data = json.loads(response.data) assert len(data) == 1 - assert data[0]['name'] == 'Test RTL-SDR' + assert data[0]["name"] == "Test RTL-SDR" class TestDependenciesEndpoint: @@ -93,152 +93,132 @@ class TestDependenciesEndpoint: def test_get_dependencies(self, client): """Test getting dependency status.""" - response = client.get('/dependencies') + response = client.get("/dependencies") assert response.status_code == 200 data = json.loads(response.data) - assert data['status'] == 'success' - assert 'os' in data - assert 'pkg_manager' in data - assert 'modes' in data + assert data["status"] == "success" + assert "os" in data + assert "pkg_manager" in data + assert "modes" in data -class TestSettingsEndpoints: - """Tests for settings API endpoints.""" +class TestSettingsEndpoints: + """Tests for settings API endpoints.""" def test_get_settings(self, client): """Test getting all settings.""" - response = client.get('/settings') + response = client.get("/settings") assert response.status_code == 200 data = json.loads(response.data) - assert data['status'] == 'success' - assert 'settings' in data + assert data["status"] == "success" + assert "settings" in data def test_save_settings(self, client): """Test saving settings.""" response = client.post( - '/settings', - data=json.dumps({'test_key': 'test_value'}), - content_type='application/json' + "/settings", data=json.dumps({"test_key": "test_value"}), content_type="application/json" ) assert response.status_code == 200 data = json.loads(response.data) - assert data['status'] == 'success' - assert 'test_key' in data['saved'] + assert data["status"] == "success" + assert "test_key" in data["saved"] def test_save_empty_settings(self, client): """Test saving empty settings returns error.""" - response = client.post( - '/settings', - data=json.dumps({}), - content_type='application/json' - ) + response = client.post("/settings", data=json.dumps({}), content_type="application/json") assert response.status_code == 400 def test_get_single_setting(self, client): """Test getting a single setting.""" # First save a setting - client.post( - '/settings', - data=json.dumps({'my_setting': 'my_value'}), - content_type='application/json' - ) + client.post("/settings", data=json.dumps({"my_setting": "my_value"}), content_type="application/json") # Then retrieve it - response = client.get('/settings/my_setting') + response = client.get("/settings/my_setting") assert response.status_code == 200 data = json.loads(response.data) - assert data['status'] == 'success' - assert data['value'] == 'my_value' + assert data["status"] == "success" + assert data["value"] == "my_value" def test_get_nonexistent_setting(self, client): """Test getting a setting that doesn't exist.""" - response = client.get('/settings/nonexistent_key_xyz') + response = client.get("/settings/nonexistent_key_xyz") assert response.status_code == 404 def test_update_setting(self, client): """Test updating a setting via PUT.""" response = client.put( - '/settings/update_test', - data=json.dumps({'value': 'updated_value'}), - content_type='application/json' + "/settings/update_test", data=json.dumps({"value": "updated_value"}), content_type="application/json" ) assert response.status_code == 200 data = json.loads(response.data) - assert data['status'] == 'success' - assert data['value'] == 'updated_value' + assert data["status"] == "success" + assert data["value"] == "updated_value" - def test_delete_setting(self, client): - """Test deleting a setting.""" + def test_delete_setting(self, client): + """Test deleting a setting.""" # First create a setting - client.post( - '/settings', - data=json.dumps({'delete_me': 'value'}), - content_type='application/json' - ) + client.post("/settings", data=json.dumps({"delete_me": "value"}), content_type="application/json") # Then delete it - response = client.delete('/settings/delete_me') + response = client.delete("/settings/delete_me") assert response.status_code == 200 - data = json.loads(response.data) - assert data['status'] == 'success' - assert data['deleted'] is True - - def test_save_observer_location_updates_env_and_runtime_defaults(self, client, monkeypatch, tmp_path): - """Saving observer location should persist to .env and update in-memory defaults.""" - import app as app_module - import config - from routes import adsb as adsb_routes - from routes import ais as ais_routes - from routes import settings as settings_routes - - with client.session_transaction() as sess: - sess['logged_in'] = True - - env_path = tmp_path / '.env' - monkeypatch.setattr(settings_routes, '_get_env_file_path', lambda: env_path) - - response = client.post( - '/settings/observer-location', - data=json.dumps({'lat': 48.0, 'lon': 16.16}), - content_type='application/json' - ) - assert response.status_code == 200 - - data = json.loads(response.data) - assert data['status'] == 'success' - assert data['lat'] == 48.0 - assert data['lon'] == 16.16 - - env_text = env_path.read_text() - assert 'INTERCEPT_DEFAULT_LAT=48.0' in env_text - assert 'INTERCEPT_DEFAULT_LON=16.16' in env_text - - assert config.DEFAULT_LATITUDE == 48.0 - assert config.DEFAULT_LONGITUDE == 16.16 - assert app_module.DEFAULT_LATITUDE == 48.0 - assert app_module.DEFAULT_LONGITUDE == 16.16 - assert adsb_routes.DEFAULT_LATITUDE == 48.0 - assert adsb_routes.DEFAULT_LONGITUDE == 16.16 - assert ais_routes.DEFAULT_LATITUDE == 48.0 - assert ais_routes.DEFAULT_LONGITUDE == 16.16 - - def test_save_observer_location_rejects_invalid_values(self, client): - """Observer location save should validate coordinates.""" - with client.session_transaction() as sess: - sess['logged_in'] = True - - response = client.post( - '/settings/observer-location', - data=json.dumps({'lat': 200, 'lon': 16.16}), - content_type='application/json' - ) - assert response.status_code == 400 + data = json.loads(response.data) + assert data["status"] == "success" + assert data["deleted"] is True + + def test_save_observer_location_updates_env_and_runtime_defaults(self, client, monkeypatch, tmp_path): + """Saving observer location should persist to .env and update in-memory defaults.""" + import app as app_module + import config + from routes import adsb as adsb_routes + from routes import ais as ais_routes + from routes import settings as settings_routes + + with client.session_transaction() as sess: + sess["logged_in"] = True + + env_path = tmp_path / ".env" + monkeypatch.setattr(settings_routes, "_get_env_file_path", lambda: env_path) + + response = client.post( + "/settings/observer-location", data=json.dumps({"lat": 48.0, "lon": 16.16}), content_type="application/json" + ) + assert response.status_code == 200 + + data = json.loads(response.data) + assert data["status"] == "success" + assert data["lat"] == 48.0 + assert data["lon"] == 16.16 + + env_text = env_path.read_text() + assert "INTERCEPT_DEFAULT_LAT=48.0" in env_text + assert "INTERCEPT_DEFAULT_LON=16.16" in env_text + + assert config.DEFAULT_LATITUDE == 48.0 + assert config.DEFAULT_LONGITUDE == 16.16 + assert app_module.DEFAULT_LATITUDE == 48.0 + assert app_module.DEFAULT_LONGITUDE == 16.16 + assert adsb_routes.DEFAULT_LATITUDE == 48.0 + assert adsb_routes.DEFAULT_LONGITUDE == 16.16 + assert ais_routes.DEFAULT_LATITUDE == 48.0 + assert ais_routes.DEFAULT_LONGITUDE == 16.16 + + def test_save_observer_location_rejects_invalid_values(self, client): + """Observer location save should validate coordinates.""" + with client.session_transaction() as sess: + sess["logged_in"] = True + + response = client.post( + "/settings/observer-location", data=json.dumps({"lat": 200, "lon": 16.16}), content_type="application/json" + ) + assert response.status_code == 400 class TestCorrelationEndpoints: @@ -246,22 +226,22 @@ class TestCorrelationEndpoints: def test_get_correlations(self, client): """Test getting device correlations.""" - response = client.get('/correlation') + response = client.get("/correlation") assert response.status_code == 200 data = json.loads(response.data) - assert data['status'] == 'success' - assert 'correlations' in data - assert 'wifi_count' in data - assert 'bt_count' in data + assert data["status"] == "success" + assert "correlations" in data + assert "wifi_count" in data + assert "bt_count" in data def test_correlations_with_confidence_filter(self, client): """Test correlation endpoint respects confidence filter.""" - response = client.get('/correlation?min_confidence=0.8') + response = client.get("/correlation?min_confidence=0.8") assert response.status_code == 200 data = json.loads(response.data) - assert data['status'] == 'success' + assert data["status"] == "success" class TestListeningPostEndpoints: @@ -269,63 +249,63 @@ class TestListeningPostEndpoints: def test_tools_check(self, client): """Test listening post tools availability check.""" - response = client.get('/listening/tools') + response = client.get("/receiver/tools") assert response.status_code == 200 data = json.loads(response.data) - assert 'rtl_fm' in data - assert 'available' in data + assert "rtl_fm" in data + assert "available" in data def test_scanner_status(self, client): """Test scanner status endpoint.""" - response = client.get('/listening/scanner/status') + response = client.get("/receiver/scanner/status") assert response.status_code == 200 data = json.loads(response.data) - assert 'running' in data - assert 'paused' in data - assert 'current_freq' in data + assert "running" in data + assert "paused" in data + assert "current_freq" in data def test_presets(self, client): """Test scanner presets endpoint.""" - response = client.get('/listening/presets') + response = client.get("/receiver/presets") assert response.status_code == 200 data = json.loads(response.data) - assert 'presets' in data - assert len(data['presets']) > 0 + assert "presets" in data + assert len(data["presets"]) > 0 # Check preset structure - preset = data['presets'][0] - assert 'name' in preset - assert 'start' in preset - assert 'end' in preset - assert 'mod' in preset + preset = data["presets"][0] + assert "name" in preset + assert "start" in preset + assert "end" in preset + assert "mod" in preset def test_scanner_stop_when_not_running(self, client): """Test stopping scanner when not running.""" - response = client.post('/listening/scanner/stop') + response = client.post("/receiver/scanner/stop") assert response.status_code == 200 data = json.loads(response.data) - assert data['status'] == 'stopped' + assert data["status"] == "stopped" def test_activity_log(self, client): """Test getting activity log.""" - response = client.get('/listening/scanner/log') + response = client.get("/receiver/scanner/log") assert response.status_code == 200 data = json.loads(response.data) - assert 'log' in data - assert 'total' in data + assert "log" in data + assert "total" in data def test_scanner_skip_when_not_running(self, client): """Test skip signal when scanner not running returns error.""" - response = client.post('/listening/scanner/skip') + response = client.post("/receiver/scanner/skip") assert response.status_code == 400 data = json.loads(response.data) - assert data['status'] == 'error' + assert data["status"] == "error" class TestAudioEndpoints: @@ -333,58 +313,48 @@ class TestAudioEndpoints: def test_audio_status(self, client): """Test audio status endpoint.""" - response = client.get('/listening/audio/status') + response = client.get("/receiver/audio/status") assert response.status_code == 200 data = json.loads(response.data) - assert 'running' in data - assert 'frequency' in data - assert 'modulation' in data + assert "running" in data + assert "frequency" in data + assert "modulation" in data def test_audio_stop_when_not_running(self, client): """Test stopping audio when not running.""" - response = client.post('/listening/audio/stop') + response = client.post("/receiver/audio/stop") assert response.status_code == 200 data = json.loads(response.data) - assert data['status'] == 'stopped' + assert data["status"] == "stopped" def test_audio_start_missing_frequency(self, client): """Test starting audio without frequency returns error.""" - response = client.post( - '/listening/audio/start', - data=json.dumps({}), - content_type='application/json' - ) + response = client.post("/receiver/audio/start", data=json.dumps({}), content_type="application/json") assert response.status_code == 400 data = json.loads(response.data) - assert data['status'] == 'error' - assert 'frequency' in data['message'].lower() + assert data["status"] == "error" + assert "frequency" in data["message"].lower() def test_audio_start_invalid_modulation(self, client): """Test starting audio with invalid modulation returns error.""" response = client.post( - '/listening/audio/start', - data=json.dumps({ - 'frequency': 98.1, - 'modulation': 'invalid_mode' - }), - content_type='application/json' + "/receiver/audio/start", + data=json.dumps({"frequency": 98.1, "modulation": "invalid_mode"}), + content_type="application/json", ) assert response.status_code == 400 data = json.loads(response.data) - assert data['status'] == 'error' - assert 'modulation' in data['message'].lower() + assert data["status"] == "error" + assert "modulation" in data["message"].lower() def test_audio_stream_when_not_running(self, client): - """Test audio stream when not running returns error.""" - response = client.get('/listening/audio/stream') - assert response.status_code == 400 - - data = json.loads(response.data) - assert data['status'] == 'error' + """Test audio stream when not running returns empty response.""" + response = client.get("/receiver/audio/stream") + assert response.status_code == 204 class TestExportEndpoints: @@ -392,36 +362,36 @@ class TestExportEndpoints: def test_export_aircraft_json(self, client): """Test exporting aircraft data as JSON.""" - response = client.get('/export/aircraft?format=json') + response = client.get("/export/aircraft?format=json") assert response.status_code == 200 - assert response.content_type == 'application/json' + assert response.content_type == "application/json" def test_export_aircraft_csv(self, client): """Test exporting aircraft data as CSV.""" - response = client.get('/export/aircraft?format=csv') + response = client.get("/export/aircraft?format=csv") assert response.status_code == 200 - assert 'text/csv' in response.content_type + assert "text/csv" in response.content_type def test_export_wifi_json(self, client): """Test exporting WiFi data as JSON.""" - response = client.get('/export/wifi?format=json') + response = client.get("/export/wifi?format=json") assert response.status_code == 200 - assert response.content_type == 'application/json' + assert response.content_type == "application/json" def test_export_wifi_csv(self, client): """Test exporting WiFi data as CSV.""" - response = client.get('/export/wifi?format=csv') + response = client.get("/export/wifi?format=csv") assert response.status_code == 200 - assert 'text/csv' in response.content_type + assert "text/csv" in response.content_type def test_export_bluetooth_json(self, client): """Test exporting Bluetooth data as JSON.""" - response = client.get('/export/bluetooth?format=json') + response = client.get("/export/bluetooth?format=json") assert response.status_code == 200 - assert response.content_type == 'application/json' + assert response.content_type == "application/json" def test_export_bluetooth_csv(self, client): """Test exporting Bluetooth data as CSV.""" - response = client.get('/export/bluetooth?format=csv') + response = client.get("/export/bluetooth?format=csv") assert response.status_code == 200 - assert 'text/csv' in response.content_type + assert "text/csv" in response.content_type diff --git a/tests/test_validation.py b/tests/test_validation.py index 8bfea2b..a509799 100644 --- a/tests/test_validation.py +++ b/tests/test_validation.py @@ -16,23 +16,23 @@ class TestFrequencyValidation: def test_valid_frequencies(self): """Test valid frequency values.""" - assert validate_frequency('152.0') == '152.0' - assert validate_frequency(152.0) == '152.0' - assert validate_frequency('1090') == '1090' - assert validate_frequency(433.92) == '433.92' + assert validate_frequency("152.0") == 152.0 + assert validate_frequency(152.0) == 152.0 + assert validate_frequency("1090") == 1090.0 + assert validate_frequency(433.92) == 433.92 def test_frequency_range(self): """Test frequency range limits.""" # RTL-SDR typical range: 24MHz - 1766MHz - assert validate_frequency('24') == '24' - assert validate_frequency('1700') == '1700' + assert validate_frequency("24") == 24.0 + assert validate_frequency("1700") == 1700.0 def test_invalid_frequencies(self): """Test invalid frequency values.""" with pytest.raises(ValueError): - validate_frequency('') + validate_frequency("") with pytest.raises(ValueError): - validate_frequency('abc') + validate_frequency("abc") with pytest.raises(ValueError): validate_frequency(-100) with pytest.raises(ValueError): @@ -44,19 +44,16 @@ class TestGainValidation: def test_valid_gains(self): """Test valid gain values.""" - assert validate_gain('0') == '0' - assert validate_gain('40') == '40' - assert validate_gain(49.6) == '49.6' - assert validate_gain('auto') == 'auto' + assert validate_gain("0") == 0.0 + assert validate_gain("40") == 40.0 + assert validate_gain(49.6) == 49.6 def test_invalid_gains(self): """Test invalid gain values.""" with pytest.raises(ValueError): validate_gain(-10) with pytest.raises(ValueError): - validate_gain(100) - with pytest.raises(ValueError): - validate_gain('invalid') + validate_gain("invalid") class TestDeviceIndexValidation: @@ -64,19 +61,17 @@ class TestDeviceIndexValidation: def test_valid_indices(self): """Test valid device indices.""" - assert validate_device_index('0') == '0' - assert validate_device_index(0) == '0' - assert validate_device_index('1') == '1' - assert validate_device_index(3) == '3' + assert validate_device_index("0") == 0 + assert validate_device_index(0) == 0 + assert validate_device_index("1") == 1 + assert validate_device_index(3) == 3 def test_invalid_indices(self): """Test invalid device indices.""" with pytest.raises(ValueError): validate_device_index(-1) with pytest.raises(ValueError): - validate_device_index('abc') - with pytest.raises(ValueError): - validate_device_index(100) + validate_device_index("abc") class TestRtlTcpHostValidation: @@ -84,19 +79,19 @@ class TestRtlTcpHostValidation: def test_valid_hosts(self): """Test valid host values.""" - assert validate_rtl_tcp_host('localhost') == 'localhost' - assert validate_rtl_tcp_host('127.0.0.1') == '127.0.0.1' - assert validate_rtl_tcp_host('192.168.1.1') == '192.168.1.1' - assert validate_rtl_tcp_host('server.example.com') == 'server.example.com' + assert validate_rtl_tcp_host("localhost") == "localhost" + assert validate_rtl_tcp_host("127.0.0.1") == "127.0.0.1" + assert validate_rtl_tcp_host("192.168.1.1") == "192.168.1.1" + assert validate_rtl_tcp_host("server.example.com") == "server.example.com" def test_invalid_hosts(self): """Test invalid host values.""" with pytest.raises(ValueError): - validate_rtl_tcp_host('') + validate_rtl_tcp_host("") with pytest.raises(ValueError): - validate_rtl_tcp_host('invalid host with spaces') + validate_rtl_tcp_host("invalid host with spaces") with pytest.raises(ValueError): - validate_rtl_tcp_host('host;rm -rf /') + validate_rtl_tcp_host("host;rm -rf /") class TestRtlTcpPortValidation: @@ -105,7 +100,7 @@ class TestRtlTcpPortValidation: def test_valid_ports(self): """Test valid port values.""" assert validate_rtl_tcp_port(1234) == 1234 - assert validate_rtl_tcp_port('1234') == 1234 + assert validate_rtl_tcp_port("1234") == 1234 assert validate_rtl_tcp_port(30003) == 30003 assert validate_rtl_tcp_port(65535) == 65535 @@ -118,4 +113,4 @@ class TestRtlTcpPortValidation: with pytest.raises(ValueError): validate_rtl_tcp_port(70000) with pytest.raises(ValueError): - validate_rtl_tcp_port('abc') + validate_rtl_tcp_port("abc") diff --git a/tests/test_waterfall.py b/tests/test_waterfall.py index bf12828..1dd8359 100644 --- a/tests/test_waterfall.py +++ b/tests/test_waterfall.py @@ -9,73 +9,90 @@ import pytest def auth_client(client): """Client with logged-in session.""" with client.session_transaction() as sess: - sess['logged_in'] = True + sess["logged_in"] = True return client def test_waterfall_start_no_rtl_power(auth_client): """Start should fail gracefully when rtl_power is not available.""" - with patch('routes.listening_post.find_rtl_power', return_value=None): - resp = auth_client.post('/listening/waterfall/start', json={ - 'start_freq': 88.0, - 'end_freq': 108.0, - }) + with patch("routes.listening_post.waterfall.find_rtl_power", return_value=None): + resp = auth_client.post( + "/receiver/waterfall/start", + json={ + "start_freq": 88.0, + "end_freq": 108.0, + }, + ) assert resp.status_code == 503 data = resp.get_json() - assert 'rtl_power' in data['message'] + assert "rtl_power" in data["message"] def test_waterfall_start_invalid_range(auth_client): """Start should reject end <= start.""" - with patch('routes.listening_post.find_rtl_power', return_value='/usr/bin/rtl_power'): - resp = auth_client.post('/listening/waterfall/start', json={ - 'start_freq': 108.0, - 'end_freq': 88.0, - }) + with patch("routes.listening_post.waterfall.find_rtl_power", return_value="/usr/bin/rtl_power"): + resp = auth_client.post( + "/receiver/waterfall/start", + json={ + "start_freq": 108.0, + "end_freq": 88.0, + }, + ) assert resp.status_code == 400 def test_waterfall_start_success(auth_client): """Start should succeed with mocked rtl_power and device.""" - with patch('routes.listening_post.find_rtl_power', return_value='/usr/bin/rtl_power'), \ - patch('routes.listening_post.app_module') as mock_app: + with ( + patch("routes.listening_post.waterfall.find_rtl_power", return_value="/usr/bin/rtl_power"), + patch("routes.listening_post.waterfall.app_module") as mock_app, + ): mock_app.claim_sdr_device.return_value = None # No error, claim succeeds - resp = auth_client.post('/listening/waterfall/start', json={ - 'start_freq': 88.0, - 'end_freq': 108.0, - 'gain': 40, - 'device': 0, - }) + resp = auth_client.post( + "/receiver/waterfall/start", + json={ + "start_freq": 88.0, + "end_freq": 108.0, + "gain": 40, + "device": 0, + }, + ) assert resp.status_code == 200 data = resp.get_json() - assert data['status'] == 'started' + assert data["status"] == "started" # Clean up: stop waterfall import routes.listening_post as lp + lp.waterfall_running = False def test_waterfall_stop(auth_client): """Stop should succeed.""" - resp = auth_client.post('/listening/waterfall/stop') + resp = auth_client.post("/receiver/waterfall/stop") assert resp.status_code == 200 data = resp.get_json() - assert data['status'] == 'stopped' + assert data["status"] == "stopped" def test_waterfall_stream_mimetype(auth_client): """Stream should return event-stream content type.""" - resp = auth_client.get('/listening/waterfall/stream') - assert resp.content_type.startswith('text/event-stream') + resp = auth_client.get("/receiver/waterfall/stream") + assert resp.content_type.startswith("text/event-stream") def test_waterfall_start_device_busy(auth_client): """Start should fail when device is in use.""" - with patch('routes.listening_post.find_rtl_power', return_value='/usr/bin/rtl_power'), \ - patch('routes.listening_post.app_module') as mock_app: - mock_app.claim_sdr_device.return_value = 'SDR device 0 is in use by scanner' - resp = auth_client.post('/listening/waterfall/start', json={ - 'start_freq': 88.0, - 'end_freq': 108.0, - }) + with ( + patch("routes.listening_post.waterfall.find_rtl_power", return_value="/usr/bin/rtl_power"), + patch("routes.listening_post.waterfall.app_module") as mock_app, + ): + mock_app.claim_sdr_device.return_value = "SDR device 0 is in use by scanner" + resp = auth_client.post( + "/receiver/waterfall/start", + json={ + "start_freq": 88.0, + "end_freq": 108.0, + }, + ) assert resp.status_code == 409