mirror of
https://github.com/jokob-sk/NetAlertX.git
synced 2025-12-07 09:36:05 -08:00
Jokob @NetAlertX
GitHub
@@ -2,6 +2,7 @@ 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"])
|
||||
@@ -25,82 +26,94 @@ def auth_headers(token):
|
||||
return {"Authorization": f"Bearer {token}"}
|
||||
|
||||
|
||||
# --- get_device_info Tests ---
|
||||
@patch('api_server.tools_routes.get_temp_db_connection')
|
||||
# --- Device Search Tests ---
|
||||
|
||||
@patch('models.device_instance.get_temp_db_connection')
|
||||
def test_get_device_info_ip_partial(mock_db_conn, client, api_token):
|
||||
"""Test get_device_info with partial IP search."""
|
||||
mock_cursor = MagicMock()
|
||||
# Mock return of a device with IP ending in .50
|
||||
mock_cursor.fetchall.return_value = [
|
||||
"""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_db_conn.return_value.cursor.return_value = mock_cursor
|
||||
mock_conn.execute.return_value = mock_execute_result
|
||||
mock_db_conn.return_value = mock_conn
|
||||
|
||||
payload = {"query": ".50"}
|
||||
response = client.post('/api/tools/get_device_info',
|
||||
json=payload,
|
||||
headers=auth_headers(api_token))
|
||||
|
||||
assert response.status_code == 200
|
||||
devices = response.get_json()
|
||||
assert len(devices) == 1
|
||||
assert devices[0]["devLastIP"] == "192.168.1.50"
|
||||
|
||||
# Verify SQL query included 3 params (MAC, Name, IP)
|
||||
args, _ = mock_cursor.execute.call_args
|
||||
assert args[0].count("?") == 3
|
||||
assert len(args[1]) == 3
|
||||
|
||||
|
||||
# --- trigger_scan Tests ---
|
||||
@patch('subprocess.run')
|
||||
def test_trigger_scan_nmap_fast(mock_run, client, api_token):
|
||||
"""Test trigger_scan with nmap_fast."""
|
||||
mock_run.return_value = MagicMock(stdout="Scan completed", returncode=0)
|
||||
|
||||
payload = {"scan_type": "nmap_fast", "target": "192.168.1.1"}
|
||||
response = client.post('/api/tools/trigger_scan',
|
||||
response = client.post('/devices/search',
|
||||
json=payload,
|
||||
headers=auth_headers(api_token))
|
||||
|
||||
assert response.status_code == 200
|
||||
data = response.get_json()
|
||||
assert data["success"] is True
|
||||
assert "nmap -F 192.168.1.1" in data["command"]
|
||||
mock_run.assert_called_once()
|
||||
assert len(data["devices"]) == 1
|
||||
assert data["devices"][0]["devLastIP"] == "192.168.1.50"
|
||||
|
||||
|
||||
@patch('subprocess.run')
|
||||
def test_trigger_scan_invalid_type(mock_run, client, api_token):
|
||||
"""Test trigger_scan with invalid scan_type."""
|
||||
payload = {"scan_type": "invalid_type", "target": "192.168.1.1"}
|
||||
response = client.post('/api/tools/trigger_scan',
|
||||
# --- Trigger Scan Tests ---
|
||||
|
||||
@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))
|
||||
|
||||
assert response.status_code == 200
|
||||
data = response.get_json()
|
||||
assert data["success"] is True
|
||||
mock_queue.add_event.assert_called_once()
|
||||
call_args = mock_queue.add_event.call_args[0]
|
||||
assert "run|ARPSCAN" in call_args[0]
|
||||
|
||||
|
||||
@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))
|
||||
|
||||
assert response.status_code == 400
|
||||
mock_run.assert_not_called()
|
||||
data = response.get_json()
|
||||
assert data["success"] is False
|
||||
|
||||
|
||||
# --- get_open_ports Tests ---
|
||||
|
||||
|
||||
@patch('subprocess.run')
|
||||
def test_get_open_ports_ip(mock_run, 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_plugin_db_conn, mock_device_db_conn, client, api_token):
|
||||
"""Test get_open_ports with an IP address."""
|
||||
mock_output = """
|
||||
Starting Nmap 7.80 ( https://nmap.org ) at 2023-10-27 10:00 UTC
|
||||
Nmap scan report for 192.168.1.1
|
||||
Host is up (0.0010s latency).
|
||||
Not shown: 98 closed ports
|
||||
PORT STATE SERVICE
|
||||
22/tcp open ssh
|
||||
80/tcp open http
|
||||
Nmap done: 1 IP address (1 host up) scanned in 0.10 seconds
|
||||
"""
|
||||
mock_run.return_value = MagicMock(stdout=mock_output, returncode=0)
|
||||
# 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 for DeviceInstance.getByIP (returns device with MAC)
|
||||
mock_execute_result.fetchone.return_value = {"devMac": "AA:BB:CC:DD:EE:FF"}
|
||||
|
||||
mock_conn.execute.return_value = mock_execute_result
|
||||
mock_plugin_db_conn.return_value = mock_conn
|
||||
mock_device_db_conn.return_value = mock_conn
|
||||
|
||||
payload = {"target": "192.168.1.1"}
|
||||
response = client.post('/api/tools/get_open_ports',
|
||||
response = client.post('/device/open_ports',
|
||||
json=payload,
|
||||
headers=auth_headers(api_token))
|
||||
|
||||
@@ -112,43 +125,46 @@ Nmap done: 1 IP address (1 host up) scanned in 0.10 seconds
|
||||
assert data["open_ports"][1]["service"] == "http"
|
||||
|
||||
|
||||
@patch('api_server.tools_routes.get_temp_db_connection')
|
||||
@patch('subprocess.run')
|
||||
def test_get_open_ports_mac_resolve(mock_run, mock_db_conn, client, api_token):
|
||||
@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 DB to resolve MAC to IP
|
||||
mock_cursor = MagicMock()
|
||||
mock_cursor.fetchone.return_value = {"devLastIP": "192.168.1.50"}
|
||||
mock_db_conn.return_value.cursor.return_value = mock_cursor
|
||||
|
||||
# Mock Nmap output
|
||||
mock_run.return_value = MagicMock(stdout="80/tcp open http", returncode=0)
|
||||
# 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_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('/api/tools/get_open_ports',
|
||||
response = client.post('/device/open_ports',
|
||||
json=payload,
|
||||
headers=auth_headers(api_token))
|
||||
|
||||
assert response.status_code == 200
|
||||
data = response.get_json()
|
||||
assert data["target"] == "192.168.1.50" # Should be resolved IP
|
||||
mock_run.assert_called_once()
|
||||
args, _ = mock_run.call_args
|
||||
assert "192.168.1.50" in args[0]
|
||||
assert data["success"] is True
|
||||
assert "target" in data
|
||||
assert len(data["open_ports"]) == 1
|
||||
assert data["open_ports"][0]["port"] == 80
|
||||
|
||||
|
||||
# --- get_network_topology Tests ---
|
||||
@patch('api_server.tools_routes.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_cursor = MagicMock()
|
||||
mock_cursor.fetchall.return_value = [
|
||||
# Mock database connection for topology query
|
||||
mock_conn = MagicMock()
|
||||
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"}
|
||||
]
|
||||
mock_db_conn.return_value.cursor.return_value = mock_cursor
|
||||
mock_conn.execute.return_value = mock_execute_result
|
||||
mock_db_conn.return_value = mock_conn
|
||||
|
||||
response = client.get('/api/tools/get_network_topology',
|
||||
response = client.get('/devices/network/topology',
|
||||
headers=auth_headers(api_token))
|
||||
|
||||
assert response.status_code == 200
|
||||
@@ -160,92 +176,71 @@ def test_get_network_topology(mock_db_conn, client, api_token):
|
||||
|
||||
|
||||
# --- get_recent_alerts Tests ---
|
||||
@patch('api_server.tools_routes.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_cursor = MagicMock()
|
||||
mock_cursor.fetchall.return_value = [
|
||||
{"eve_DateTime": "2023-10-27 10:00:00", "eve_EventType": "New Device", "eve_MAC": "CC:CC:CC:CC:CC:CC", "eve_IP": "192.168.1.100", "devName": "Unknown"}
|
||||
# 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"}
|
||||
]
|
||||
mock_db_conn.return_value.cursor.return_value = mock_cursor
|
||||
mock_conn.execute.return_value = mock_execute_result
|
||||
mock_db_conn.return_value = mock_conn
|
||||
|
||||
payload = {"hours": 24}
|
||||
response = client.post('/api/tools/get_recent_alerts',
|
||||
json=payload,
|
||||
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 len(data) == 1
|
||||
assert data[0]["eve_EventType"] == "New Device"
|
||||
assert data["success"] is True
|
||||
assert data["hours"] == 24
|
||||
|
||||
|
||||
# --- set_device_alias Tests ---
|
||||
@patch('api_server.tools_routes.get_temp_db_connection')
|
||||
def test_set_device_alias(mock_db_conn, client, api_token):
|
||||
# --- Device Alias Tests ---
|
||||
|
||||
@patch('api_server.api_server_start.update_device_column')
|
||||
def test_set_device_alias(mock_update_col, client, api_token):
|
||||
"""Test set_device_alias."""
|
||||
mock_cursor = MagicMock()
|
||||
mock_cursor.rowcount = 1 # Simulate successful update
|
||||
mock_db_conn.return_value.cursor.return_value = mock_cursor
|
||||
mock_update_col.return_value = {"success": True, "message": "Device alias updated"}
|
||||
|
||||
payload = {"mac": "AA:BB:CC:DD:EE:FF", "alias": "New Name"}
|
||||
response = client.post('/api/tools/set_device_alias',
|
||||
payload = {"alias": "New Device Name"}
|
||||
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()
|
||||
assert data["success"] is True
|
||||
mock_update_col.assert_called_once_with("AA:BB:CC:DD:EE:FF", "devName", "New Device Name")
|
||||
|
||||
|
||||
@patch('api_server.tools_routes.get_temp_db_connection')
|
||||
def test_set_device_alias_not_found(mock_db_conn, client, api_token):
|
||||
@patch('api_server.api_server_start.update_device_column')
|
||||
def test_set_device_alias_not_found(mock_update_col, client, api_token):
|
||||
"""Test set_device_alias when device is not found."""
|
||||
mock_cursor = MagicMock()
|
||||
mock_cursor.rowcount = 0 # Simulate no rows updated
|
||||
mock_db_conn.return_value.cursor.return_value = mock_cursor
|
||||
mock_update_col.return_value = {"success": False, "error": "Device not found"}
|
||||
|
||||
payload = {"mac": "AA:BB:CC:DD:EE:FF", "alias": "New Name"}
|
||||
response = client.post('/api/tools/set_device_alias',
|
||||
json=payload,
|
||||
headers=auth_headers(api_token))
|
||||
|
||||
assert response.status_code == 404
|
||||
|
||||
|
||||
# --- wol_wake_device Tests ---
|
||||
@patch('subprocess.run')
|
||||
def test_wol_wake_device(mock_subprocess, client, api_token):
|
||||
"""Test wol_wake_device."""
|
||||
mock_subprocess.return_value.stdout = "Sending magic packet to 255.255.255.255:9 with AA:BB:CC:DD:EE:FF"
|
||||
mock_subprocess.return_value.returncode = 0
|
||||
|
||||
payload = {"mac": "AA:BB:CC:DD:EE:FF"}
|
||||
response = client.post('/api/tools/wol_wake_device',
|
||||
payload = {"alias": "New Device Name"}
|
||||
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()
|
||||
assert data["success"] is True
|
||||
mock_subprocess.assert_called_with(["wakeonlan", "AA:BB:CC:DD:EE:FF"], capture_output=True, text=True, check=True)
|
||||
assert data["success"] is False
|
||||
assert "Device not found" in data["error"]
|
||||
|
||||
|
||||
@patch('api_server.tools_routes.get_temp_db_connection')
|
||||
@patch('subprocess.run')
|
||||
def test_wol_wake_device_by_ip(mock_subprocess, mock_db_conn, client, api_token):
|
||||
"""Test wol_wake_device with IP address."""
|
||||
# Mock DB for IP resolution
|
||||
mock_cursor = MagicMock()
|
||||
mock_cursor.fetchone.return_value = {"devMac": "AA:BB:CC:DD:EE:FF"}
|
||||
mock_db_conn.return_value.cursor.return_value = mock_cursor
|
||||
# --- Wake-on-LAN Tests ---
|
||||
|
||||
# Mock subprocess
|
||||
mock_subprocess.return_value.stdout = "Sending magic packet to 255.255.255.255:9 with AA:BB:CC:DD:EE:FF"
|
||||
mock_subprocess.return_value.returncode = 0
|
||||
@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 = {"ip": "192.168.1.50"}
|
||||
response = client.post('/api/tools/wol_wake_device',
|
||||
payload = {"devMac": "AA:BB:CC:DD:EE:FF"}
|
||||
response = client.post('/nettools/wakeonlan',
|
||||
json=payload,
|
||||
headers=auth_headers(api_token))
|
||||
|
||||
@@ -254,34 +249,58 @@ def test_wol_wake_device_by_ip(mock_subprocess, mock_db_conn, client, api_token)
|
||||
assert data["success"] is True
|
||||
assert "AA:BB:CC:DD:EE:FF" in data["message"]
|
||||
|
||||
# Verify DB lookup
|
||||
mock_cursor.execute.assert_called_with("SELECT devMac FROM Devices WHERE devLastIP = ?", ("192.168.1.50",))
|
||||
|
||||
# Verify subprocess call
|
||||
mock_subprocess.assert_called_with(["wakeonlan", "AA:BB:CC:DD:EE:FF"], capture_output=True, text=True, check=True)
|
||||
|
||||
|
||||
def test_wol_wake_device_invalid_mac(client, api_token):
|
||||
"""Test wol_wake_device with invalid MAC."""
|
||||
payload = {"mac": "invalid-mac"}
|
||||
response = client.post('/api/tools/wol_wake_device',
|
||||
payload = {"devMac": "invalid-mac"}
|
||||
response = client.post('/nettools/wakeonlan',
|
||||
json=payload,
|
||||
headers=auth_headers(api_token))
|
||||
|
||||
assert response.status_code == 400
|
||||
data = response.get_json()
|
||||
assert data["success"] is False
|
||||
|
||||
|
||||
# --- openapi_spec Tests ---
|
||||
def test_openapi_spec(client):
|
||||
"""Test openapi_spec endpoint contains new paths."""
|
||||
response = client.get('/api/tools/openapi.json')
|
||||
# --- OpenAPI Spec Tests ---
|
||||
|
||||
# --- Latest Device Tests ---
|
||||
|
||||
@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
|
||||
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"
|
||||
}
|
||||
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))
|
||||
|
||||
assert response.status_code == 200
|
||||
data = response.get_json()
|
||||
assert len(data) == 1
|
||||
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))
|
||||
assert response.status_code == 200
|
||||
spec = response.get_json()
|
||||
|
||||
# Check for new endpoints
|
||||
assert "/trigger_scan" in spec["paths"]
|
||||
assert "/get_open_ports" in spec["paths"]
|
||||
assert "/get_network_topology" in spec["paths"]
|
||||
assert "/get_recent_alerts" in spec["paths"]
|
||||
assert "/set_device_alias" in spec["paths"]
|
||||
assert "/wol_wake_device" in spec["paths"]
|
||||
# Check for MCP tool endpoints in the spec with correct paths
|
||||
assert "/nettools/trigger-scan" in spec["paths"]
|
||||
assert "/device/open_ports" in spec["paths"]
|
||||
assert "/devices/network/topology" in spec["paths"]
|
||||
assert "/events/recent" in spec["paths"]
|
||||
assert "/device/{mac}/set-alias" in spec["paths"]
|
||||
assert "/nettools/wakeonlan" in spec["paths"]
|
||||
@@ -1,79 +0,0 @@
|
||||
import sys
|
||||
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]
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def api_token():
|
||||
return get_setting_value("API_TOKEN")
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def client():
|
||||
with app.test_client() as client:
|
||||
yield client
|
||||
|
||||
|
||||
def auth_headers(token):
|
||||
return {"Authorization": f"Bearer {token}"}
|
||||
|
||||
|
||||
def test_openapi_spec(client):
|
||||
"""Test OpenAPI spec endpoint."""
|
||||
response = client.get('/api/tools/openapi.json')
|
||||
assert response.status_code == 200
|
||||
spec = response.get_json()
|
||||
assert "openapi" in spec
|
||||
assert "info" in spec
|
||||
assert "paths" in spec
|
||||
assert "/list_devices" in spec["paths"]
|
||||
assert "/get_device_info" in spec["paths"]
|
||||
|
||||
|
||||
def test_list_devices(client, api_token):
|
||||
"""Test list_devices endpoint."""
|
||||
response = client.post('/api/tools/list_devices', headers=auth_headers(api_token))
|
||||
assert response.status_code == 200
|
||||
devices = response.get_json()
|
||||
assert isinstance(devices, list)
|
||||
# If there are devices, check structure
|
||||
if devices:
|
||||
device = devices[0]
|
||||
assert "devName" in device
|
||||
assert "devMac" in device
|
||||
|
||||
|
||||
def test_get_device_info(client, api_token):
|
||||
"""Test get_device_info endpoint."""
|
||||
# Test with a query that might not exist
|
||||
payload = {"query": "nonexistent_device"}
|
||||
response = client.post('/api/tools/get_device_info',
|
||||
json=payload,
|
||||
headers=auth_headers(api_token))
|
||||
# Should return 404 if no match, or 200 with results
|
||||
assert response.status_code in [200, 404]
|
||||
if response.status_code == 200:
|
||||
devices = response.get_json()
|
||||
assert isinstance(devices, list)
|
||||
elif response.status_code == 404:
|
||||
# Expected for no matches
|
||||
pass
|
||||
|
||||
|
||||
def test_list_devices_unauthorized(client):
|
||||
"""Test list_devices without authorization."""
|
||||
response = client.post('/api/tools/list_devices')
|
||||
assert response.status_code == 401
|
||||
|
||||
|
||||
def test_get_device_info_unauthorized(client):
|
||||
"""Test get_device_info without authorization."""
|
||||
payload = {"query": "test"}
|
||||
response = client.post('/api/tools/get_device_info', json=payload)
|
||||
assert response.status_code == 401
|
||||
Reference in New Issue
Block a user