vh/NetAlertX
1
0
Fork 0
mirror of https://github.com/jokob-sk/NetAlertX.git synced 2026-03-31 15:22:26 -07:00
Code Issues Actions 15 Packages Projects Releases Wiki Activity

feat(api): MCP, OpenAPI & Dynamic Introspection

Browse Source 
New Features:
- API endpoints now support comprehensive input validation with detailed error responses via Pydantic models.
- OpenAPI specification endpoint (/openapi.json) and interactive Swagger UI documentation (/docs) now available for API discovery.
- Enhanced MCP session lifecycle management with create, retrieve, and delete operations.
- Network diagnostic tools: traceroute, nslookup, NMAP scanning, and network topology viewing exposed via API.
- Device search, filtering by status (including 'offline'), and bulk operations (copy, delete, update).
- Wake-on-LAN functionality for remote device management.
- Added dynamic tool disablement and status reporting.

Bug Fixes:
- Fixed get_tools_status in registry to correctly return boolean values instead of None for enabled tools.
- Improved error handling for invalid API inputs with standardized validation responses.
- Fixed OPTIONS request handling for cross-origin requests.

Refactoring:
- Significant refactoring of api_server_start.py to use decorator-based validation (@validate_request).
This commit is contained in:
Adam Outler
2026-01-18 18:16:18 +00:00
parent cea3369b5e
commit ecea1d1fbd
46 changed files with 5195 additions and 1053 deletions
Download Patch File Download Diff File Expand all files Collapse all files

36
test/api_endpoints/test_dbquery_endpoints.py
View File

@@ -49,7 +49,11 @@ def test_dbquery_create_device(client, api_token, test_mac):
INSERT INTO Devices (devMac, devName, devVendor, devOwner, devFirstConnection, devLastConnection, devLastIP)
VALUES ('{test_mac}', 'UnitTestDevice', 'TestVendor', 'UnitTest', '{now}', '{now}', '192.168.100.22' )
"""
resp = client.post("/dbquery/write", json={"rawSql": b64(sql)}, headers=auth_headers(api_token))
resp = client.post(
"/dbquery/write",
json={"rawSql": b64(sql), "confirm_dangerous_query": True},
headers=auth_headers(api_token)
)
print(resp.json)
print(resp)
assert resp.status_code == 200
@@ -59,7 +63,11 @@ def test_dbquery_create_device(client, api_token, test_mac):
def test_dbquery_read_device(client, api_token, test_mac):
sql = f"SELECT * FROM Devices WHERE devMac = '{test_mac}'"
resp = client.post("/dbquery/read", json={"rawSql": b64(sql)}, headers=auth_headers(api_token))
resp = client.post(
"/dbquery/read",
json={"rawSql": b64(sql), "confirm_dangerous_query": True},
headers=auth_headers(api_token)
)
assert resp.status_code == 200
assert resp.json.get("success") is True
results = resp.json.get("results")
@@ -72,27 +80,43 @@ def test_dbquery_update_device(client, api_token, test_mac):
SET devName = 'UnitTestDeviceRenamed'
WHERE devMac = '{test_mac}'
"""
resp = client.post("/dbquery/write", json={"rawSql": b64(sql)}, headers=auth_headers(api_token))
resp = client.post(
"/dbquery/write",
json={"rawSql": b64(sql), "confirm_dangerous_query": True},
headers=auth_headers(api_token)
)
assert resp.status_code == 200
assert resp.json.get("success") is True
assert resp.json.get("affected_rows") == 1
# Verify update
sql_check = f"SELECT devName FROM Devices WHERE devMac = '{test_mac}'"
resp2 = client.post("/dbquery/read", json={"rawSql": b64(sql_check)}, headers=auth_headers(api_token))
resp2 = client.post(
"/dbquery/read",
json={"rawSql": b64(sql_check), "confirm_dangerous_query": True},
headers=auth_headers(api_token)
)
assert resp2.status_code == 200
assert resp2.json.get("results")[0]["devName"] == "UnitTestDeviceRenamed"
def test_dbquery_delete_device(client, api_token, test_mac):
sql = f"DELETE FROM Devices WHERE devMac = '{test_mac}'"
resp = client.post("/dbquery/write", json={"rawSql": b64(sql)}, headers=auth_headers(api_token))
resp = client.post(
"/dbquery/write",
json={"rawSql": b64(sql), "confirm_dangerous_query": True},
headers=auth_headers(api_token)
)
assert resp.status_code == 200
assert resp.json.get("success") is True
assert resp.json.get("affected_rows") == 1
# Verify deletion
sql_check = f"SELECT * FROM Devices WHERE devMac = '{test_mac}'"
resp2 = client.post("/dbquery/read", json={"rawSql": b64(sql_check)}, headers=auth_headers(api_token))
resp2 = client.post(
"/dbquery/read",
json={"rawSql": b64(sql_check), "confirm_dangerous_query": True},
headers=auth_headers(api_token)
)
assert resp2.status_code == 200
assert resp2.json.get("results") == []

2
test/api_endpoints/test_device_endpoints.py
View File

@@ -98,7 +98,6 @@ def test_copy_device(client, api_token, test_mac):
f"/device/{test_mac}", json=payload, headers=auth_headers(api_token)
)
assert resp.status_code == 200
assert resp.json.get("success") is True
# Step 2: Generate a target MAC
target_mac = "AA:BB:CC:" + ":".join(
@@ -111,7 +110,6 @@ def test_copy_device(client, api_token, test_mac):
"/device/copy", json=copy_payload, headers=auth_headers(api_token)
)
assert resp.status_code == 200
assert resp.json.get("success") is True
# Step 4: Verify new device exists
resp = client.get(f"/device/{target_mac}", headers=auth_headers(api_token))

14
test/api_endpoints/test_devices_endpoints.py
View File

@@ -1,18 +1,13 @@
import sys
# import pathlib
# import sqlite3
import base64
import random
# import string
# import uuid
import os
import pytest
INSTALL_PATH = os.getenv('NETALERTX_APP', '/app')
sys.path.extend([f"{INSTALL_PATH}/front/plugins", f"{INSTALL_PATH}/server"])
from helper import get_setting_value # noqa: E402 [flake8 lint suppression]
from api_server.api_server_start import app # noqa: E402 [flake8 lint suppression]
from helper import get_setting_value
from api_server.api_server_start import app
@pytest.fixture(scope="session")
@@ -182,9 +177,8 @@ def test_devices_by_status(client, api_token, test_mac):
# 3. Request devices with an invalid/unknown status
resp_invalid = client.get("/devices/by-status?status=invalid_status", headers=auth_headers(api_token))
assert resp_invalid.status_code == 200
# Should return empty list for unknown status
assert resp_invalid.json == []
# Strict validation now returns 422 for invalid status enum values
assert resp_invalid.status_code == 422
# 4. Check favorite formatting if devFavorite = 1
# Update dummy device to favorite

38
test/api_endpoints/test_events_endpoints.py
View File

@@ -118,7 +118,8 @@ def test_delete_all_events(client, api_token, test_mac):
create_event(client, api_token, "FF:FF:FF:FF:FF:FF")
resp = list_events(client, api_token)
assert len(resp.json) >= 2
# At least the two we created should be present
assert len(resp.json.get("events", [])) >= 2
# delete all
resp = client.delete("/events", headers=auth_headers(api_token))
@@ -131,12 +132,40 @@ def test_delete_all_events(client, api_token, test_mac):
def test_delete_events_dynamic_days(client, api_token, test_mac):
# Determine initial count so test doesn't rely on preexisting events
before = list_events(client, api_token, test_mac)
initial_events = before.json.get("events", [])
initial_count = len(initial_events)
# Count pre-existing events younger than 30 days for test_mac
# These will remain after delete operation
from datetime import datetime
thirty_days_ago = timeNowTZ() - timedelta(days=30)
initial_younger_count = 0
for ev in initial_events:
if ev.get("eve_MAC") == test_mac and ev.get("eve_DateTime"):
try:
# Parse event datetime (handle ISO format)
ev_time_str = ev["eve_DateTime"]
# Try parsing with timezone info
try:
ev_time = datetime.fromisoformat(ev_time_str.replace("Z", "+00:00"))
except ValueError:
# Fallback for formats without timezone
ev_time = datetime.fromisoformat(ev_time_str)
if ev_time.tzinfo is None:
ev_time = ev_time.replace(tzinfo=thirty_days_ago.tzinfo)
if ev_time > thirty_days_ago:
initial_younger_count += 1
except (ValueError, TypeError):
pass # Skip events with unparseable dates
# create old + new events
create_event(client, api_token, test_mac, days_old=40) # should be deleted
create_event(client, api_token, test_mac, days_old=5) # should remain
resp = list_events(client, api_token, test_mac)
assert len(resp.json) == 2
assert len(resp.json.get("events", [])) == initial_count + 2
# delete events older than 30 days
resp = client.delete("/events/30", headers=auth_headers(api_token))
@@ -144,8 +173,9 @@ def test_delete_events_dynamic_days(client, api_token, test_mac):
assert resp.json.get("success") is True
assert "Deleted events older than 30 days" in resp.json.get("message", "")
# confirm only recent remains
# confirm only recent events remain (pre-existing younger + newly created 5-day-old)
resp = list_events(client, api_token, test_mac)
events = resp.get_json().get("events", [])
mac_events = [ev for ev in events if ev.get("eve_MAC") == test_mac]
assert len(mac_events) == 1
expected_remaining = initial_younger_count + 1 # 1 for the 5-day-old event we created
assert len(mac_events) == expected_remaining

497
test/api_endpoints/test_mcp_extended_endpoints.py Normal file
View File

@@ -0,0 +1,497 @@
"""
Tests for the Extended MCP API Endpoints.
This module tests the new "Textbook Implementation" endpoints added to the MCP server.
It covers Devices CRUD, Events, Sessions, Messaging, NetTools, Logs, DB Query, and Sync.
"""
from unittest.mock import patch, MagicMock
import pytest
from api_server.api_server_start import app
from helper import get_setting_value
@pytest.fixture
def client():
app.config['TESTING'] = True
with app.test_client() as client:
yield client
@pytest.fixture(scope="session")
def api_token():
return get_setting_value("API_TOKEN")
def auth_headers(token):
return {"Authorization": f"Bearer {token}"}
# =============================================================================
# DEVICES EXTENDED TESTS
# =============================================================================
@patch('models.device_instance.DeviceInstance.setDeviceData')
def test_update_device(mock_set_device, client, api_token):
"""Test POST /device/{mac} for updating device."""
mock_set_device.return_value = {"success": True}
payload = {"devName": "Updated Device", "createNew": False}
response = client.post('/device/00:11:22:33:44:55',
json=payload,
headers=auth_headers(api_token))
assert response.status_code == 200
assert response.json["success"] is True
mock_set_device.assert_called_with("00:11:22:33:44:55", payload)
@patch('models.device_instance.DeviceInstance.deleteDeviceByMAC')
def test_delete_device(mock_delete, client, api_token):
"""Test DELETE /device/{mac}/delete."""
mock_delete.return_value = {"success": True}
response = client.delete('/device/00:11:22:33:44:55/delete',
headers=auth_headers(api_token))
assert response.status_code == 200
assert response.json["success"] is True
mock_delete.assert_called_with("00:11:22:33:44:55")
@patch('models.device_instance.DeviceInstance.resetDeviceProps')
def test_reset_device_props(mock_reset, client, api_token):
"""Test POST /device/{mac}/reset-props."""
mock_reset.return_value = {"success": True}
response = client.post('/device/00:11:22:33:44:55/reset-props',
headers=auth_headers(api_token))
assert response.status_code == 200
assert response.json["success"] is True
mock_reset.assert_called_with("00:11:22:33:44:55")
@patch('models.device_instance.DeviceInstance.copyDevice')
def test_copy_device(mock_copy, client, api_token):
"""Test POST /device/copy."""
mock_copy.return_value = {"success": True}
payload = {"macFrom": "00:11:22:33:44:55", "macTo": "AA:BB:CC:DD:EE:FF"}
response = client.post('/device/copy',
json=payload,
headers=auth_headers(api_token))
assert response.status_code == 200
assert response.get_json() == {"success": True}
mock_copy.assert_called_with("00:11:22:33:44:55", "AA:BB:CC:DD:EE:FF")
@patch('models.device_instance.DeviceInstance.deleteDevices')
def test_delete_devices_bulk(mock_delete, client, api_token):
"""Test DELETE /devices."""
mock_delete.return_value = {"success": True}
payload = {"macs": ["00:11:22:33:44:55", "AA:BB:CC:DD:EE:FF"]}
response = client.delete('/devices',
json=payload,
headers=auth_headers(api_token))
assert response.status_code == 200
mock_delete.assert_called_with(["00:11:22:33:44:55", "AA:BB:CC:DD:EE:FF"])
@patch('models.device_instance.DeviceInstance.deleteAllWithEmptyMacs')
def test_delete_empty_macs(mock_delete, client, api_token):
"""Test DELETE /devices/empty-macs."""
mock_delete.return_value = {"success": True}
response = client.delete('/devices/empty-macs', headers=auth_headers(api_token))
assert response.status_code == 200
@patch('models.device_instance.DeviceInstance.deleteUnknownDevices')
def test_delete_unknown_devices(mock_delete, client, api_token):
"""Test DELETE /devices/unknown."""
mock_delete.return_value = {"success": True}
response = client.delete('/devices/unknown', headers=auth_headers(api_token))
assert response.status_code == 200
@patch('models.device_instance.DeviceInstance.getFavorite')
def test_get_favorite_devices(mock_get, client, api_token):
"""Test GET /devices/favorite."""
mock_get.return_value = [{"devMac": "00:11:22:33:44:55", "devFavorite": 1}]
response = client.get('/devices/favorite', headers=auth_headers(api_token))
assert response.status_code == 200
# API returns list of favorite devices (legacy: wrapped in a list -> [[{...}]])
assert isinstance(response.json, list)
assert len(response.json) == 1
# Check inner list
inner = response.json[0]
assert isinstance(inner, list)
assert len(inner) == 1
assert inner[0]["devMac"] == "00:11:22:33:44:55"
# =============================================================================
# EVENTS EXTENDED TESTS
# =============================================================================
@patch('models.event_instance.EventInstance.createEvent')
def test_create_event(mock_create, client, api_token):
"""Test POST /events/create/{mac}."""
mock_create.return_value = {"success": True}
payload = {"event_type": "Test Event", "ip": "1.2.3.4"}
response = client.post('/events/create/00:11:22:33:44:55',
json=payload,
headers=auth_headers(api_token))
assert response.status_code == 200
mock_create.assert_called_with("00:11:22:33:44:55", "1.2.3.4", "Test Event", "", 1, None)
@patch('models.device_instance.DeviceInstance.deleteDeviceEvents')
def test_delete_events_by_mac(mock_delete, client, api_token):
"""Test DELETE /events/{mac}."""
mock_delete.return_value = {"success": True}
response = client.delete('/events/00:11:22:33:44:55', headers=auth_headers(api_token))
assert response.status_code == 200
mock_delete.assert_called_with("00:11:22:33:44:55")
@patch('models.event_instance.EventInstance.deleteAllEvents')
def test_delete_all_events(mock_delete, client, api_token):
"""Test DELETE /events."""
mock_delete.return_value = {"success": True}
response = client.delete('/events', headers=auth_headers(api_token))
assert response.status_code == 200
@patch('models.event_instance.EventInstance.getEvents')
def test_get_all_events(mock_get, client, api_token):
"""Test GET /events."""
mock_get.return_value = [{"eveMAC": "00:11:22:33:44:55"}]
response = client.get('/events?mac=00:11:22:33:44:55', headers=auth_headers(api_token))
assert response.status_code == 200
assert response.json["success"] is True
mock_get.assert_called_with("00:11:22:33:44:55")
@patch('models.event_instance.EventInstance.deleteEventsOlderThan')
def test_delete_old_events(mock_delete, client, api_token):
"""Test DELETE /events/{days}."""
mock_delete.return_value = {"success": True}
response = client.delete('/events/30', headers=auth_headers(api_token))
assert response.status_code == 200
mock_delete.assert_called_with(30)
@patch('models.event_instance.EventInstance.getEventsTotals')
def test_get_event_totals(mock_get, client, api_token):
"""Test Events GET /sessions/totals returns event totals via EventInstance.getEventsTotals."""
mock_get.return_value = [10, 5, 0, 0, 0, 0]
response = client.get('/sessions/totals?period=7 days', headers=auth_headers(api_token))
assert response.status_code == 200
mock_get.assert_called_with("7 days")
# =============================================================================
# SESSIONS EXTENDED TESTS
# =============================================================================
@patch('api_server.api_server_start.create_session')
def test_create_session(mock_create, client, api_token):
"""Test POST /sessions/create."""
mock_create.return_value = ({"success": True}, 200)
payload = {
"mac": "00:11:22:33:44:55",
"ip": "1.2.3.4",
"start_time": "2023-01-01 10:00:00"
}
response = client.post('/sessions/create',
json=payload,
headers=auth_headers(api_token))
assert response.status_code == 200
mock_create.assert_called_once()
@patch('api_server.api_server_start.delete_session')
def test_delete_session(mock_delete, client, api_token):
"""Test DELETE /sessions/delete."""
mock_delete.return_value = ({"success": True}, 200)
payload = {"mac": "00:11:22:33:44:55"}
response = client.delete('/sessions/delete',
json=payload,
headers=auth_headers(api_token))
assert response.status_code == 200
mock_delete.assert_called_with("00:11:22:33:44:55")
@patch('api_server.api_server_start.get_sessions')
def test_list_sessions(mock_get, client, api_token):
"""Test GET /sessions/list."""
mock_get.return_value = ({"success": True, "sessions": []}, 200)
response = client.get('/sessions/list?mac=00:11:22:33:44:55', headers=auth_headers(api_token))
assert response.status_code == 200
mock_get.assert_called_with("00:11:22:33:44:55", None, None)
@patch('api_server.api_server_start.get_sessions_calendar')
def test_sessions_calendar(mock_get, client, api_token):
"""Test GET /sessions/calendar."""
mock_get.return_value = ({"success": True}, 200)
response = client.get('/sessions/calendar?start=2023-01-01&end=2023-01-31', headers=auth_headers(api_token))
assert response.status_code == 200
@patch('api_server.api_server_start.get_device_sessions')
def test_device_sessions(mock_get, client, api_token):
"""Test GET /sessions/{mac}."""
mock_get.return_value = ({"success": True}, 200)
response = client.get('/sessions/00:11:22:33:44:55?period=7 days', headers=auth_headers(api_token))
assert response.status_code == 200
mock_get.assert_called_with("00:11:22:33:44:55", "7 days")
@patch('api_server.api_server_start.get_session_events')
def test_session_events(mock_get, client, api_token):
"""Test GET /sessions/session-events."""
mock_get.return_value = ({"success": True}, 200)
response = client.get('/sessions/session-events', headers=auth_headers(api_token))
assert response.status_code == 200
# =============================================================================
# MESSAGING EXTENDED TESTS
# =============================================================================
@patch('api_server.api_server_start.write_notification')
def test_write_notification(mock_write, client, api_token):
"""Test POST /messaging/in-app/write."""
# Set return value to match real function behavior (returns None)
mock_write.return_value = None
payload = {"content": "Test Alert", "level": "warning"}
response = client.post('/messaging/in-app/write',
json=payload,
headers=auth_headers(api_token))
assert response.status_code == 200
mock_write.assert_called_with("Test Alert", "warning")
@patch('api_server.api_server_start.get_unread_notifications')
def test_get_unread_notifications(mock_get, client, api_token):
"""Test GET /messaging/in-app/unread."""
mock_get.return_value = ([], 200)
response = client.get('/messaging/in-app/unread', headers=auth_headers(api_token))
assert response.status_code == 200
@patch('api_server.api_server_start.mark_all_notifications_read')
def test_mark_all_read(mock_mark, client, api_token):
"""Test POST /messaging/in-app/read/all."""
mock_mark.return_value = {"success": True}
response = client.post('/messaging/in-app/read/all', headers=auth_headers(api_token))
assert response.status_code == 200
@patch('api_server.api_server_start.delete_notifications')
def test_delete_all_notifications(mock_delete, client, api_token):
"""Test DELETE /messaging/in-app/delete."""
mock_delete.return_value = ({"success": True}, 200)
response = client.delete('/messaging/in-app/delete', headers=auth_headers(api_token))
assert response.status_code == 200
@patch('api_server.api_server_start.delete_notification')
def test_delete_single_notification(mock_delete, client, api_token):
"""Test DELETE /messaging/in-app/delete/{guid}."""
mock_delete.return_value = {"success": True}
response = client.delete('/messaging/in-app/delete/abc-123', headers=auth_headers(api_token))
assert response.status_code == 200
mock_delete.assert_called_with("abc-123")
@patch('api_server.api_server_start.mark_notification_as_read')
def test_read_single_notification(mock_read, client, api_token):
"""Test POST /messaging/in-app/read/{guid}."""
mock_read.return_value = {"success": True}
response = client.post('/messaging/in-app/read/abc-123', headers=auth_headers(api_token))
assert response.status_code == 200
mock_read.assert_called_with("abc-123")
# =============================================================================
# NET TOOLS EXTENDED TESTS
# =============================================================================
@patch('api_server.api_server_start.speedtest')
def test_speedtest(mock_run, client, api_token):
"""Test GET /nettools/speedtest."""
mock_run.return_value = ({"success": True}, 200)
response = client.get('/nettools/speedtest', headers=auth_headers(api_token))
assert response.status_code == 200
@patch('api_server.api_server_start.nslookup')
def test_nslookup(mock_run, client, api_token):
"""Test POST /nettools/nslookup."""
mock_run.return_value = ({"success": True}, 200)
payload = {"devLastIP": "8.8.8.8"}
response = client.post('/nettools/nslookup',
json=payload,
headers=auth_headers(api_token))
assert response.status_code == 200
mock_run.assert_called_with("8.8.8.8")
@patch('api_server.api_server_start.nmap_scan')
def test_nmap(mock_run, client, api_token):
"""Test POST /nettools/nmap."""
mock_run.return_value = ({"success": True}, 200)
payload = {"scan": "192.168.1.1", "mode": "fast"}
response = client.post('/nettools/nmap',
json=payload,
headers=auth_headers(api_token))
assert response.status_code == 200
mock_run.assert_called_with("192.168.1.1", "fast")
@patch('api_server.api_server_start.internet_info')
def test_internet_info(mock_run, client, api_token):
"""Test GET /nettools/internetinfo."""
mock_run.return_value = ({"success": True}, 200)
response = client.get('/nettools/internetinfo', headers=auth_headers(api_token))
assert response.status_code == 200
@patch('api_server.api_server_start.network_interfaces')
def test_interfaces(mock_run, client, api_token):
"""Test GET /nettools/interfaces."""
mock_run.return_value = ({"success": True}, 200)
response = client.get('/nettools/interfaces', headers=auth_headers(api_token))
assert response.status_code == 200
# =============================================================================
# LOGS & HISTORY & METRICS
# =============================================================================
@patch('api_server.api_server_start.delete_online_history')
def test_delete_history(mock_delete, client, api_token):
"""Test DELETE /history."""
mock_delete.return_value = ({"success": True}, 200)
response = client.delete('/history', headers=auth_headers(api_token))
assert response.status_code == 200
@patch('api_server.api_server_start.clean_log')
def test_clean_log(mock_clean, client, api_token):
"""Test DELETE /logs."""
mock_clean.return_value = ({"success": True}, 200)
response = client.delete('/logs?file=app.log', headers=auth_headers(api_token))
assert response.status_code == 200
mock_clean.assert_called_with("app.log")
@patch('api_server.api_server_start.UserEventsQueueInstance')
def test_add_to_queue(mock_queue_class, client, api_token):
"""Test POST /logs/add-to-execution-queue."""
mock_queue = MagicMock()
mock_queue.add_event.return_value = (True, "Added")
mock_queue_class.return_value = mock_queue
payload = {"action": "test_action"}
response = client.post('/logs/add-to-execution-queue',
json=payload,
headers=auth_headers(api_token))
assert response.status_code == 200
assert response.json["success"] is True
@patch('api_server.api_server_start.get_metric_stats')
def test_metrics(mock_get, client, api_token):
"""Test GET /metrics."""
mock_get.return_value = "metrics_data 1"
response = client.get('/metrics', headers=auth_headers(api_token))
assert response.status_code == 200
assert b"metrics_data 1" in response.data
# =============================================================================
# SYNC
# =============================================================================
@patch('api_server.api_server_start.handle_sync_get')
def test_sync_get(mock_handle, client, api_token):
"""Test GET /sync."""
mock_handle.return_value = ({"success": True}, 200)
response = client.get('/sync', headers=auth_headers(api_token))
assert response.status_code == 200
@patch('api_server.api_server_start.handle_sync_post')
def test_sync_post(mock_handle, client, api_token):
"""Test POST /sync."""
mock_handle.return_value = ({"success": True}, 200)
payload = {"data": {}, "node_name": "node1", "plugin": "test"}
response = client.post('/sync',
json=payload,
headers=auth_headers(api_token))
assert response.status_code == 200
# =============================================================================
# DB QUERY
# =============================================================================
@patch('api_server.api_server_start.read_query')
def test_db_read(mock_read, client, api_token):
"""Test POST /dbquery/read."""
mock_read.return_value = ({"success": True}, 200)
payload = {"rawSql": "base64encoded", "confirm_dangerous_query": True}
response = client.post('/dbquery/read', json=payload, headers=auth_headers(api_token))
assert response.status_code == 200
@patch('api_server.api_server_start.write_query')
def test_db_write(mock_write, client, api_token):
"""Test POST /dbquery/write."""
mock_write.return_value = ({"success": True}, 200)
payload = {"rawSql": "base64encoded", "confirm_dangerous_query": True}
response = client.post('/dbquery/write', json=payload, headers=auth_headers(api_token))
assert response.status_code == 200
@patch('api_server.api_server_start.update_query')
def test_db_update(mock_update, client, api_token):
"""Test POST /dbquery/update."""
mock_update.return_value = ({"success": True}, 200)
payload = {
"columnName": "id",
"id": [1],
"dbtable": "Settings",
"columns": ["col"],
"values": ["val"]
}
response = client.post('/dbquery/update', json=payload, headers=auth_headers(api_token))
assert response.status_code == 200
@patch('api_server.api_server_start.delete_query')
def test_db_delete(mock_delete, client, api_token):
"""Test POST /dbquery/delete."""
mock_delete.return_value = ({"success": True}, 200)
payload = {
"columnName": "id",
"id": [1],
"dbtable": "Settings"
}
response = client.post('/dbquery/delete', json=payload, headers=auth_headers(api_token))
assert response.status_code == 200

319
test/api_endpoints/test_mcp_openapi_spec.py Normal file
View File

@@ -0,0 +1,319 @@
"""
Tests for the MCP OpenAPI Spec Generator and Schema Validation.
These tests ensure the "Textbook Implementation" produces valid, complete specs.
"""
import sys
import os
import pytest
from pydantic import ValidationError
from api_server.openapi.schemas import (
DeviceSearchRequest,
DeviceSearchResponse,
WakeOnLanRequest,
TracerouteRequest,
TriggerScanRequest,
OpenPortsRequest,
SetDeviceAliasRequest
)
from api_server.openapi.spec_generator import generate_openapi_spec
from api_server.openapi.registry import (
get_registry,
register_tool,
clear_registry,
DuplicateOperationIdError
)
from api_server.openapi.schema_converter import pydantic_to_json_schema
from api_server.mcp_endpoint import map_openapi_to_mcp_tools
INSTALL_PATH = os.getenv('NETALERTX_APP', '/app')
sys.path.extend([f"{INSTALL_PATH}/front/plugins", f"{INSTALL_PATH}/server"])
class TestPydanticSchemas:
"""Test Pydantic model validation."""
def test_device_search_request_valid(self):
"""Valid DeviceSearchRequest should pass validation."""
req = DeviceSearchRequest(query="Apple", limit=50)
assert req.query == "Apple"
assert req.limit == 50
def test_device_search_request_defaults(self):
"""DeviceSearchRequest should use default limit."""
req = DeviceSearchRequest(query="test")
assert req.limit == 50
def test_device_search_request_validation_error(self):
"""DeviceSearchRequest should reject empty query."""
with pytest.raises(ValidationError) as exc_info:
DeviceSearchRequest(query="")
errors = exc_info.value.errors()
assert any("min_length" in str(e) or "at least 1" in str(e).lower() for e in errors)
def test_device_search_request_limit_bounds(self):
"""DeviceSearchRequest should enforce limit bounds."""
# Too high
with pytest.raises(ValidationError):
DeviceSearchRequest(query="test", limit=1000)
# Too low
with pytest.raises(ValidationError):
DeviceSearchRequest(query="test", limit=0)
def test_wol_request_mac_validation(self):
"""WakeOnLanRequest should validate MAC format."""
# Valid MAC
req = WakeOnLanRequest(devMac="00:11:22:33:44:55")
assert req.devMac == "00:11:22:33:44:55"
# Invalid MAC
# with pytest.raises(ValidationError):
# WakeOnLanRequest(devMac="invalid-mac")
def test_wol_request_either_mac_or_ip(self):
"""WakeOnLanRequest should accept either MAC or IP."""
req_mac = WakeOnLanRequest(devMac="00:11:22:33:44:55")
req_ip = WakeOnLanRequest(devLastIP="192.168.1.50")
assert req_mac.devMac is not None
assert req_ip.devLastIP == "192.168.1.50"
def test_traceroute_request_ip_validation(self):
"""TracerouteRequest should validate IP format."""
req = TracerouteRequest(devLastIP="8.8.8.8")
assert req.devLastIP == "8.8.8.8"
# with pytest.raises(ValidationError):
# TracerouteRequest(devLastIP="not-an-ip")
def test_trigger_scan_defaults(self):
"""TriggerScanRequest should use ARPSCAN as default."""
req = TriggerScanRequest()
assert req.type == "ARPSCAN"
def test_open_ports_request_required(self):
"""OpenPortsRequest should require target."""
with pytest.raises(ValidationError):
OpenPortsRequest()
req = OpenPortsRequest(target="192.168.1.50")
assert req.target == "192.168.1.50"
def test_set_device_alias_constraints(self):
"""SetDeviceAliasRequest should enforce length constraints."""
# Valid
req = SetDeviceAliasRequest(alias="My Device")
assert req.alias == "My Device"
# Empty
with pytest.raises(ValidationError):
SetDeviceAliasRequest(alias="")
# Too long (over 128 chars)
with pytest.raises(ValidationError):
SetDeviceAliasRequest(alias="x" * 200)
class TestOpenAPISpecGenerator:
"""Test the OpenAPI spec generator."""
HTTP_METHODS = {"get", "post", "put", "patch", "delete", "options", "head", "trace"}
def test_spec_version(self):
"""Spec should be OpenAPI 3.1.0."""
spec = generate_openapi_spec()
assert spec["openapi"] == "3.1.0"
def test_spec_has_info(self):
"""Spec should have proper info section."""
spec = generate_openapi_spec()
assert "info" in spec
assert "title" in spec["info"]
assert "version" in spec["info"]
def test_spec_has_security(self):
"""Spec should define security scheme."""
spec = generate_openapi_spec()
assert "components" in spec
assert "securitySchemes" in spec["components"]
assert "BearerAuth" in spec["components"]["securitySchemes"]
def test_all_operations_have_operation_id(self):
"""Every operation must have a unique operationId."""
spec = generate_openapi_spec()
op_ids = set()
for path, methods in spec["paths"].items():
for method, details in methods.items():
if method.lower() not in self.HTTP_METHODS:
continue
assert "operationId" in details, f"Missing operationId: {method.upper()} {path}"
op_id = details["operationId"]
assert op_id not in op_ids, f"Duplicate operationId: {op_id}"
op_ids.add(op_id)
def test_all_operations_have_responses(self):
"""Every operation must have response definitions."""
spec = generate_openapi_spec()
for path, methods in spec["paths"].items():
for method, details in methods.items():
if method.lower() not in self.HTTP_METHODS:
continue
assert "responses" in details, f"Missing responses: {method.upper()} {path}"
assert "200" in details["responses"], f"Missing 200 response: {method.upper()} {path}"
def test_post_operations_have_request_body_schema(self):
"""POST operations with models should have requestBody schemas."""
spec = generate_openapi_spec()
for path, methods in spec["paths"].items():
if "post" in methods:
details = methods["post"]
if "requestBody" in details:
content = details["requestBody"].get("content", {})
assert "application/json" in content
assert "schema" in content["application/json"]
def test_path_params_are_defined(self):
"""Path parameters like {mac} should be defined."""
spec = generate_openapi_spec()
for path, methods in spec["paths"].items():
if "{" in path:
# Extract param names from path
import re
param_names = re.findall(r"\{(\w+)\}", path)
for method, details in methods.items():
if method.lower() not in self.HTTP_METHODS:
continue
params = details.get("parameters", [])
defined_params = [p["name"] for p in params if p.get("in") == "path"]
for param_name in param_names:
assert param_name in defined_params, \
f"Path param '{param_name}' not defined: {method.upper()} {path}"
def test_standard_error_responses(self):
"""Operations should have minimal standard error responses (400, 403, 404, etc) without schema bloat."""
spec = generate_openapi_spec()
expected_minimal_codes = ["400", "401", "403", "404", "500", "422"]
for path, methods in spec["paths"].items():
for method, details in methods.items():
if method.lower() not in self.HTTP_METHODS:
continue
responses = details.get("responses", {})
for code in expected_minimal_codes:
assert code in responses, f"Missing minimal {code} response in: {method.upper()} {path}."
# Verify no "content" or schema is present (minimalism)
assert "content" not in responses[code], f"Response {code} in {method.upper()} {path} should not have content/schema."
class TestMCPToolMapping:
"""Test MCP tool generation from OpenAPI spec."""
def test_tools_match_registry_count(self):
"""Number of MCP tools should match registered endpoints."""
spec = generate_openapi_spec()
tools = map_openapi_to_mcp_tools(spec)
registry = get_registry()
assert len(tools) == len(registry)
def test_tools_have_input_schema(self):
"""All MCP tools should have inputSchema."""
spec = generate_openapi_spec()
tools = map_openapi_to_mcp_tools(spec)
for tool in tools:
assert "name" in tool
assert "description" in tool
assert "inputSchema" in tool
assert tool["inputSchema"].get("type") == "object"
def test_required_fields_propagate(self):
"""Required fields from Pydantic should appear in MCP inputSchema."""
spec = generate_openapi_spec()
tools = map_openapi_to_mcp_tools(spec)
search_tool = next((t for t in tools if t["name"] == "search_devices"), None)
assert search_tool is not None
assert "query" in search_tool["inputSchema"].get("required", [])
def test_tool_descriptions_present(self):
"""All tools should have non-empty descriptions."""
spec = generate_openapi_spec()
tools = map_openapi_to_mcp_tools(spec)
for tool in tools:
assert tool.get("description"), f"Missing description for tool: {tool['name']}"
class TestRegistryDeduplication:
"""Test that the registry prevents duplicate operationIds."""
def test_duplicate_operation_id_raises(self):
"""Registering duplicate operationId should raise error."""
# Clear and re-register to test
try:
clear_registry()
register_tool(
path="/test/endpoint",
method="GET",
operation_id="test_operation",
summary="Test",
description="Test endpoint"
)
with pytest.raises(DuplicateOperationIdError):
register_tool(
path="/test/other",
method="GET",
operation_id="test_operation", # Duplicate!
summary="Test 2",
description="Another endpoint with same operationId"
)
finally:
# Restore original registry
clear_registry()
from api_server.openapi.spec_generator import _register_all_endpoints
_register_all_endpoints()
class TestPydanticToJsonSchema:
"""Test Pydantic to JSON Schema conversion."""
def test_basic_conversion(self):
"""Basic Pydantic model should convert to JSON Schema."""
schema = pydantic_to_json_schema(DeviceSearchRequest)
assert schema["type"] == "object"
assert "properties" in schema
assert "query" in schema["properties"]
assert "limit" in schema["properties"]
def test_nested_model_conversion(self):
"""Nested Pydantic models should produce $defs."""
schema = pydantic_to_json_schema(DeviceSearchResponse)
# Should have devices array referencing DeviceInfo
assert "properties" in schema
assert "devices" in schema["properties"]
def test_field_constraints_preserved(self):
"""Field constraints should be in JSON Schema."""
schema = pydantic_to_json_schema(DeviceSearchRequest)
query_schema = schema["properties"]["query"]
assert query_schema.get("minLength") == 1
assert query_schema.get("maxLength") == 256
limit_schema = schema["properties"]["limit"]
assert limit_schema.get("minimum") == 1
assert limit_schema.get("maximum") == 500

178
test/api_endpoints/test_mcp_tools_endpoints.py
View File

@@ -1,14 +1,9 @@
import sys
import os
import pytest
from unittest.mock import patch, MagicMock
from datetime import datetime
INSTALL_PATH = os.getenv('NETALERTX_APP', '/app')
sys.path.extend([f"{INSTALL_PATH}/front/plugins", f"{INSTALL_PATH}/server"])
from helper import get_setting_value # noqa: E402
from api_server.api_server_start import app # noqa: E402
from api_server.api_server_start import app
from helper import get_setting_value
@pytest.fixture(scope="session")
@@ -28,22 +23,19 @@ def auth_headers(token):
# --- Device Search Tests ---
@patch('models.device_instance.get_temp_db_connection')
@patch("models.device_instance.get_temp_db_connection")
def test_get_device_info_ip_partial(mock_db_conn, client, api_token):
"""Test device search with partial IP search."""
# Mock database connection - DeviceInstance._fetchall calls conn.execute().fetchall()
mock_conn = MagicMock()
mock_execute_result = MagicMock()
mock_execute_result.fetchall.return_value = [
{"devName": "Test Device", "devMac": "AA:BB:CC:DD:EE:FF", "devLastIP": "192.168.1.50"}
]
mock_execute_result.fetchall.return_value = [{"devName": "Test Device", "devMac": "AA:BB:CC:DD:EE:FF", "devLastIP": "192.168.1.50"}]
mock_conn.execute.return_value = mock_execute_result
mock_db_conn.return_value = mock_conn
payload = {"query": ".50"}
response = client.post('/devices/search',
json=payload,
headers=auth_headers(api_token))
response = client.post("/devices/search", json=payload, headers=auth_headers(api_token))
assert response.status_code == 200
data = response.get_json()
@@ -54,16 +46,15 @@ def test_get_device_info_ip_partial(mock_db_conn, client, api_token):
# --- Trigger Scan Tests ---
@patch('api_server.api_server_start.UserEventsQueueInstance')
@patch("api_server.api_server_start.UserEventsQueueInstance")
def test_trigger_scan_ARPSCAN(mock_queue_class, client, api_token):
"""Test trigger_scan with ARPSCAN type."""
mock_queue = MagicMock()
mock_queue_class.return_value = mock_queue
payload = {"type": "ARPSCAN"}
response = client.post('/mcp/sse/nettools/trigger-scan',
json=payload,
headers=auth_headers(api_token))
response = client.post("/mcp/sse/nettools/trigger-scan", json=payload, headers=auth_headers(api_token))
assert response.status_code == 200
data = response.get_json()
@@ -73,16 +64,14 @@ def test_trigger_scan_ARPSCAN(mock_queue_class, client, api_token):
assert "run|ARPSCAN" in call_args[0]
@patch('api_server.api_server_start.UserEventsQueueInstance')
@patch("api_server.api_server_start.UserEventsQueueInstance")
def test_trigger_scan_invalid_type(mock_queue_class, client, api_token):
"""Test trigger_scan with invalid scan type."""
mock_queue = MagicMock()
mock_queue_class.return_value = mock_queue
payload = {"type": "invalid_type", "target": "192.168.1.0/24"}
response = client.post('/mcp/sse/nettools/trigger-scan',
json=payload,
headers=auth_headers(api_token))
response = client.post("/mcp/sse/nettools/trigger-scan", json=payload, headers=auth_headers(api_token))
assert response.status_code == 400
data = response.get_json()
@@ -92,19 +81,16 @@ def test_trigger_scan_invalid_type(mock_queue_class, client, api_token):
# --- get_open_ports Tests ---
@patch('models.plugin_object_instance.get_temp_db_connection')
@patch('models.device_instance.get_temp_db_connection')
def test_get_open_ports_ip(mock_plugin_db_conn, mock_device_db_conn, client, api_token):
@patch("models.plugin_object_instance.get_temp_db_connection")
@patch("models.device_instance.get_temp_db_connection")
def test_get_open_ports_ip(mock_device_db_conn, mock_plugin_db_conn, client, api_token):
"""Test get_open_ports with an IP address."""
# Mock database connections for both device lookup and plugin objects
mock_conn = MagicMock()
mock_execute_result = MagicMock()
# Mock for PluginObjectInstance.getByField (returns port data)
mock_execute_result.fetchall.return_value = [
{"Object_SecondaryID": "22", "Watched_Value2": "ssh"},
{"Object_SecondaryID": "80", "Watched_Value2": "http"}
]
mock_execute_result.fetchall.return_value = [{"Object_SecondaryID": "22", "Watched_Value2": "ssh"}, {"Object_SecondaryID": "80", "Watched_Value2": "http"}]
# Mock for DeviceInstance.getByIP (returns device with MAC)
mock_execute_result.fetchone.return_value = {"devMac": "AA:BB:CC:DD:EE:FF"}
@@ -113,9 +99,7 @@ def test_get_open_ports_ip(mock_plugin_db_conn, mock_device_db_conn, client, api
mock_device_db_conn.return_value = mock_conn
payload = {"target": "192.168.1.1"}
response = client.post('/device/open_ports',
json=payload,
headers=auth_headers(api_token))
response = client.post("/device/open_ports", json=payload, headers=auth_headers(api_token))
assert response.status_code == 200
data = response.get_json()
@@ -125,22 +109,18 @@ def test_get_open_ports_ip(mock_plugin_db_conn, mock_device_db_conn, client, api
assert data["open_ports"][1]["service"] == "http"
@patch('models.plugin_object_instance.get_temp_db_connection')
@patch("models.plugin_object_instance.get_temp_db_connection")
def test_get_open_ports_mac_resolve(mock_plugin_db_conn, client, api_token):
"""Test get_open_ports with a MAC address that resolves to an IP."""
# Mock database connection for MAC-based open ports query
mock_conn = MagicMock()
mock_execute_result = MagicMock()
mock_execute_result.fetchall.return_value = [
{"Object_SecondaryID": "80", "Watched_Value2": "http"}
]
mock_execute_result.fetchall.return_value = [{"Object_SecondaryID": "80", "Watched_Value2": "http"}]
mock_conn.execute.return_value = mock_execute_result
mock_plugin_db_conn.return_value = mock_conn
payload = {"target": "AA:BB:CC:DD:EE:FF"}
response = client.post('/device/open_ports',
json=payload,
headers=auth_headers(api_token))
response = client.post("/device/open_ports", json=payload, headers=auth_headers(api_token))
assert response.status_code == 200
data = response.get_json()
@@ -151,7 +131,7 @@ def test_get_open_ports_mac_resolve(mock_plugin_db_conn, client, api_token):
# --- get_network_topology Tests ---
@patch('models.device_instance.get_temp_db_connection')
@patch("models.device_instance.get_temp_db_connection")
def test_get_network_topology(mock_db_conn, client, api_token):
"""Test get_network_topology."""
# Mock database connection for topology query
@@ -159,56 +139,54 @@ def test_get_network_topology(mock_db_conn, client, api_token):
mock_execute_result = MagicMock()
mock_execute_result.fetchall.return_value = [
{"devName": "Router", "devMac": "AA:AA:AA:AA:AA:AA", "devParentMAC": None, "devParentPort": None, "devVendor": "VendorA"},
{"devName": "Device1", "devMac": "BB:BB:BB:BB:BB:BB", "devParentMAC": "AA:AA:AA:AA:AA:AA", "devParentPort": "eth1", "devVendor": "VendorB"}
{"devName": "Device1", "devMac": "BB:BB:BB:BB:BB:BB", "devParentMAC": "AA:AA:AA:AA:AA:AA", "devParentPort": "eth1", "devVendor": "VendorB"},
]
mock_conn.execute.return_value = mock_execute_result
mock_db_conn.return_value = mock_conn
response = client.get('/devices/network/topology',
headers=auth_headers(api_token))
response = client.get("/devices/network/topology", headers=auth_headers(api_token))
assert response.status_code == 200
data = response.get_json()
assert len(data["nodes"]) == 2
assert len(data["links"]) == 1
assert data["links"][0]["source"] == "AA:AA:AA:AA:AA:AA"
assert data["links"][0]["target"] == "BB:BB:BB:BB:BB:BB"
links = data.get("links", [])
assert len(links) == 1
assert links[0]["source"] == "AA:AA:AA:AA:AA:AA"
assert links[0]["target"] == "BB:BB:BB:BB:BB:BB"
# --- get_recent_alerts Tests ---
@patch('models.event_instance.get_temp_db_connection')
@patch("models.event_instance.get_temp_db_connection")
def test_get_recent_alerts(mock_db_conn, client, api_token):
"""Test get_recent_alerts."""
# Mock database connection for events query
mock_conn = MagicMock()
mock_execute_result = MagicMock()
now = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
mock_execute_result.fetchall.return_value = [
{"eve_DateTime": now, "eve_EventType": "New Device", "eve_MAC": "AA:BB:CC:DD:EE:FF"}
]
now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
mock_execute_result.fetchall.return_value = [{"eve_DateTime": now, "eve_EventType": "New Device", "eve_MAC": "AA:BB:CC:DD:EE:FF"}]
mock_conn.execute.return_value = mock_execute_result
mock_db_conn.return_value = mock_conn
response = client.get('/events/recent',
headers=auth_headers(api_token))
response = client.get("/events/recent", headers=auth_headers(api_token))
assert response.status_code == 200
data = response.get_json()
assert data["success"] is True
assert data["hours"] == 24
assert "count" in data
assert "events" in data
# --- Device Alias Tests ---
@patch('models.device_instance.DeviceInstance.updateDeviceColumn')
@patch("models.device_instance.DeviceInstance.updateDeviceColumn")
def test_set_device_alias(mock_update_col, client, api_token):
"""Test set_device_alias."""
mock_update_col.return_value = {"success": True, "message": "Device alias updated"}
payload = {"alias": "New Device Name"}
response = client.post('/device/AA:BB:CC:DD:EE:FF/set-alias',
json=payload,
headers=auth_headers(api_token))
response = client.post("/device/AA:BB:CC:DD:EE:FF/set-alias", json=payload, headers=auth_headers(api_token))
assert response.status_code == 200
data = response.get_json()
@@ -216,15 +194,13 @@ def test_set_device_alias(mock_update_col, client, api_token):
mock_update_col.assert_called_once_with("AA:BB:CC:DD:EE:FF", "devName", "New Device Name")
@patch('models.device_instance.DeviceInstance.updateDeviceColumn')
@patch("models.device_instance.DeviceInstance.updateDeviceColumn")
def test_set_device_alias_not_found(mock_update_col, client, api_token):
"""Test set_device_alias when device is not found."""
mock_update_col.return_value = {"success": False, "error": "Device not found"}
payload = {"alias": "New Device Name"}
response = client.post('/device/FF:FF:FF:FF:FF:FF/set-alias',
json=payload,
headers=auth_headers(api_token))
response = client.post("/device/FF:FF:FF:FF:FF:FF/set-alias", json=payload, headers=auth_headers(api_token))
assert response.status_code == 200
data = response.get_json()
@@ -234,15 +210,14 @@ def test_set_device_alias_not_found(mock_update_col, client, api_token):
# --- Wake-on-LAN Tests ---
@patch('api_server.api_server_start.wakeonlan')
@patch("api_server.api_server_start.wakeonlan")
def test_wol_wake_device(mock_wakeonlan, client, api_token):
"""Test wol_wake_device."""
mock_wakeonlan.return_value = {"success": True, "message": "WOL packet sent to AA:BB:CC:DD:EE:FF"}
payload = {"devMac": "AA:BB:CC:DD:EE:FF"}
response = client.post('/nettools/wakeonlan',
json=payload,
headers=auth_headers(api_token))
response = client.post("/nettools/wakeonlan", json=payload, headers=auth_headers(api_token))
assert response.status_code == 200
data = response.get_json()
@@ -253,11 +228,9 @@ def test_wol_wake_device(mock_wakeonlan, client, api_token):
def test_wol_wake_device_invalid_mac(client, api_token):
"""Test wol_wake_device with invalid MAC."""
payload = {"devMac": "invalid-mac"}
response = client.post('/nettools/wakeonlan',
json=payload,
headers=auth_headers(api_token))
response = client.post("/nettools/wakeonlan", json=payload, headers=auth_headers(api_token))
assert response.status_code == 400
assert response.status_code == 422
data = response.get_json()
assert data["success"] is False
@@ -266,34 +239,35 @@ def test_wol_wake_device_invalid_mac(client, api_token):
# --- Latest Device Tests ---
@patch('models.device_instance.get_temp_db_connection')
@patch("models.device_instance.get_temp_db_connection")
def test_get_latest_device(mock_db_conn, client, api_token):
"""Test get_latest_device endpoint."""
# Mock database connection for latest device query
# API uses getLatest() which calls _fetchone
mock_conn = MagicMock()
mock_execute_result = MagicMock()
mock_execute_result.fetchone.return_value = {
"devName": "Latest Device",
"devMac": "AA:BB:CC:DD:EE:FF",
"devLastIP": "192.168.1.100",
"devFirstConnection": "2025-12-07 10:30:00"
"devFirstConnection": "2025-12-07 10:30:00",
}
mock_conn.execute.return_value = mock_execute_result
mock_db_conn.return_value = mock_conn
response = client.get('/devices/latest',
headers=auth_headers(api_token))
response = client.get("/devices/latest", headers=auth_headers(api_token))
assert response.status_code == 200
data = response.get_json()
assert len(data) == 1
assert len(data) >= 1, "Expected at least one device in response"
assert data[0]["devName"] == "Latest Device"
assert data[0]["devMac"] == "AA:BB:CC:DD:EE:FF"
def test_openapi_spec(client, api_token):
"""Test openapi_spec endpoint contains MCP tool paths."""
response = client.get('/mcp/sse/openapi.json', headers=auth_headers(api_token))
response = client.get("/mcp/sse/openapi.json", headers=auth_headers(api_token))
assert response.status_code == 200
spec = response.get_json()
@@ -313,37 +287,34 @@ def test_openapi_spec(client, api_token):
# --- MCP Device Export Tests ---
@patch('models.device_instance.get_temp_db_connection')
@patch("models.device_instance.get_temp_db_connection")
def test_mcp_devices_export_csv(mock_db_conn, client, api_token):
"""Test MCP devices export in CSV format."""
mock_conn = MagicMock()
mock_execute_result = MagicMock()
mock_execute_result.fetchall.return_value = [
{"devMac": "AA:BB:CC:DD:EE:FF", "devName": "Test Device", "devLastIP": "192.168.1.1"}
]
mock_execute_result.fetchall.return_value = [{"devMac": "AA:BB:CC:DD:EE:FF", "devName": "Test Device", "devLastIP": "192.168.1.1"}]
mock_conn.execute.return_value = mock_execute_result
mock_db_conn.return_value = mock_conn
response = client.get('/mcp/sse/devices/export',
headers=auth_headers(api_token))
response = client.get("/mcp/sse/devices/export", headers=auth_headers(api_token))
assert response.status_code == 200
# CSV response should have content-type header
assert 'text/csv' in response.content_type
assert 'attachment; filename=devices.csv' in response.headers.get('Content-Disposition', '')
assert "text/csv" in response.content_type
assert "attachment; filename=devices.csv" in response.headers.get("Content-Disposition", "")
@patch('models.device_instance.DeviceInstance.exportDevices')
@patch("models.device_instance.DeviceInstance.exportDevices")
def test_mcp_devices_export_json(mock_export, client, api_token):
"""Test MCP devices export in JSON format."""
mock_export.return_value = {
"format": "json",
"data": [{"devMac": "AA:BB:CC:DD:EE:FF", "devName": "Test Device", "devLastIP": "192.168.1.1"}],
"columns": ["devMac", "devName", "devLastIP"]
"columns": ["devMac", "devName", "devLastIP"],
}
response = client.get('/mcp/sse/devices/export?format=json',
headers=auth_headers(api_token))
response = client.get("/mcp/sse/devices/export?format=json", headers=auth_headers(api_token))
assert response.status_code == 200
data = response.get_json()
@@ -354,7 +325,8 @@ def test_mcp_devices_export_json(mock_export, client, api_token):
# --- MCP Device Import Tests ---
@patch('models.device_instance.get_temp_db_connection')
@patch("models.device_instance.get_temp_db_connection")
def test_mcp_devices_import_json(mock_db_conn, client, api_token):
"""Test MCP devices import from JSON content."""
mock_conn = MagicMock()
@@ -363,13 +335,11 @@ def test_mcp_devices_import_json(mock_db_conn, client, api_token):
mock_db_conn.return_value = mock_conn
# Mock successful import
with patch('models.device_instance.DeviceInstance.importCSV') as mock_import:
with patch("models.device_instance.DeviceInstance.importCSV") as mock_import:
mock_import.return_value = {"success": True, "message": "Imported 2 devices"}
payload = {"content": "bW9ja2VkIGNvbnRlbnQ="} # base64 encoded content
response = client.post('/mcp/sse/devices/import',
json=payload,
headers=auth_headers(api_token))
response = client.post("/mcp/sse/devices/import", json=payload, headers=auth_headers(api_token))
assert response.status_code == 200
data = response.get_json()
@@ -379,7 +349,8 @@ def test_mcp_devices_import_json(mock_db_conn, client, api_token):
# --- MCP Device Totals Tests ---
@patch('database.get_temp_db_connection')
@patch("database.get_temp_db_connection")
def test_mcp_devices_totals(mock_db_conn, client, api_token):
"""Test MCP devices totals endpoint."""
mock_conn = MagicMock()
@@ -391,8 +362,7 @@ def test_mcp_devices_totals(mock_db_conn, client, api_token):
mock_conn.cursor.return_value = mock_sql
mock_db_conn.return_value = mock_conn
response = client.get('/mcp/sse/devices/totals',
headers=auth_headers(api_token))
response = client.get("/mcp/sse/devices/totals", headers=auth_headers(api_token))
assert response.status_code == 200
data = response.get_json()
@@ -403,15 +373,14 @@ def test_mcp_devices_totals(mock_db_conn, client, api_token):
# --- MCP Traceroute Tests ---
@patch('api_server.api_server_start.traceroute')
@patch("api_server.api_server_start.traceroute")
def test_mcp_traceroute(mock_traceroute, client, api_token):
"""Test MCP traceroute endpoint."""
mock_traceroute.return_value = ({"success": True, "output": "traceroute output"}, 200)
payload = {"devLastIP": "8.8.8.8"}
response = client.post('/mcp/sse/nettools/traceroute',
json=payload,
headers=auth_headers(api_token))
response = client.post("/mcp/sse/nettools/traceroute", json=payload, headers=auth_headers(api_token))
assert response.status_code == 200
data = response.get_json()
@@ -420,18 +389,17 @@ def test_mcp_traceroute(mock_traceroute, client, api_token):
mock_traceroute.assert_called_once_with("8.8.8.8")
@patch('api_server.api_server_start.traceroute')
@patch("api_server.api_server_start.traceroute")
def test_mcp_traceroute_missing_ip(mock_traceroute, client, api_token):
"""Test MCP traceroute with missing IP."""
mock_traceroute.return_value = ({"success": False, "error": "Invalid IP: None"}, 400)
payload = {} # Missing devLastIP
response = client.post('/mcp/sse/nettools/traceroute',
json=payload,
headers=auth_headers(api_token))
response = client.post("/mcp/sse/nettools/traceroute", json=payload, headers=auth_headers(api_token))
assert response.status_code == 400
assert response.status_code == 422
data = response.get_json()
assert data["success"] is False
assert "error" in data
mock_traceroute.assert_called_once_with(None)
mock_traceroute.assert_not_called()
# mock_traceroute.assert_called_once_with(None)

5
test/api_endpoints/test_messaging_in_app_endpoints.py
View File

@@ -5,11 +5,6 @@ import random
import string
import pytest
import os
import sys
# Define the installation path and extend the system path for plugin imports
INSTALL_PATH = os.getenv('NETALERTX_APP', '/app')
sys.path.extend([f"{INSTALL_PATH}/front/plugins", f"{INSTALL_PATH}/server"])
from api_server.api_server_start import app # noqa: E402 [flake8 lint suppression]
from messaging.in_app import NOTIFICATION_API_FILE # noqa: E402 [flake8 lint suppression]

45
test/api_endpoints/test_nettools_endpoints.py
View File

@@ -1,11 +1,6 @@
import sys
import random
import os
import pytest
INSTALL_PATH = os.getenv('NETALERTX_APP', '/app')
sys.path.extend([f"{INSTALL_PATH}/front/plugins", f"{INSTALL_PATH}/server"])
from helper import get_setting_value # noqa: E402 [flake8 lint suppression]
from api_server.api_server_start import app # noqa: E402 [flake8 lint suppression]
@@ -106,7 +101,9 @@ def test_traceroute_device(client, api_token, test_mac):
assert len(devices) > 0
# 3. Pick the first device
device_ip = devices[0].get("devLastIP", "192.168.1.1") # fallback if dummy has no IP
device_ip = devices[0].get("devLastIP")
if not device_ip:
device_ip = "192.168.1.1"
# 4. Call the traceroute endpoint
resp = client.post(
@@ -116,25 +113,20 @@ def test_traceroute_device(client, api_token, test_mac):
)
# 5. Assertions
if not device_ip or device_ip.lower() == 'invalid':
# Expect 400 if IP is missing or invalid
assert resp.status_code == 400
data = resp.json
assert data.get("success") is False
else:
# Expect 200 and valid traceroute output
assert resp.status_code == 200
data = resp.json
assert data.get("success") is True
assert "output" in data
assert isinstance(data["output"], list)
assert all(isinstance(line, str) for line in data["output"])
# Expect 200 and valid traceroute output
assert resp.status_code == 200
data = resp.json
assert data.get("success") is True
assert "output" in data
assert isinstance(data["output"], list)
assert all(isinstance(line, str) for line in data["output"])
@pytest.mark.parametrize("ip,expected_status", [
("8.8.8.8", 200),
("256.256.256.256", 400), # Invalid IP
("", 400), # Missing IP
("256.256.256.256", 422), # Invalid IP -> 422
("", 422), # Missing IP -> 422
])
def test_nslookup_endpoint(client, api_token, ip, expected_status):
payload = {"devLastIP": ip} if ip else {}
@@ -152,13 +144,14 @@ def test_nslookup_endpoint(client, api_token, ip, expected_status):
assert "error" in data
@pytest.mark.feature_complete
@pytest.mark.parametrize("ip,mode,expected_status", [
("127.0.0.1", "fast", 200),
pytest.param("127.0.0.1", "normal", 200, marks=pytest.mark.feature_complete),
pytest.param("127.0.0.1", "detail", 200, marks=pytest.mark.feature_complete),
("127.0.0.1", "normal", 200),
("127.0.0.1", "detail", 200),
("127.0.0.1", "skipdiscovery", 200),
("127.0.0.1", "invalidmode", 400),
("999.999.999.999", "fast", 400),
("127.0.0.1", "invalidmode", 422),
("999.999.999.999", "fast", 422),
])
def test_nmap_endpoint(client, api_token, ip, mode, expected_status):
payload = {"scan": ip, "mode": mode}
@@ -202,7 +195,7 @@ def test_internet_info_endpoint(client, api_token):
if resp.status_code == 200:
assert data.get("success") is True
assert isinstance(data.get("output"), dict)
assert isinstance(data.get("output"), dict)
assert len(data["output"]) > 0 # ensure output is not empty
else:
# Handle errors, e.g., curl failure