Update docker compose and unit tests

docker-compose.yml

@@ -19,6 +19,9 @@ services:
       - CHOWN                                       # Required for root-entrypoint to chown /data + /tmp before dropping privileges
       - SETUID                                      # Required for root-entrypoint to switch to non-root user
       - SETGID                                      # Required for root-entrypoint to switch to non-root group                  
+    sysctls:                                        # ARP flux mitigation for host networking accuracy
+      net.ipv4.conf.all.arp_ignore: 1
+      net.ipv4.conf.all.arp_announce: 2
     volumes:
 
       - type: volume                                # Persistent Docker-managed Named Volume for storage

install/docker/docker-compose.dev.yml

@@ -13,6 +13,9 @@ services:
       - CHOWN
       - SETUID
       - SETGID
+    sysctls:
+      net.ipv4.conf.all.arp_ignore: 1
+      net.ipv4.conf.all.arp_announce: 2
     volumes:
       - type: volume
         source: netalertx_data

install/docker/docker-compose.yml

@@ -13,6 +13,9 @@ services:
       - CHOWN
       - SETUID
       - SETGID
+    sysctls:
+      net.ipv4.conf.all.arp_ignore: 1
+      net.ipv4.conf.all.arp_announce: 2
     volumes:
       - type: volume
         source: netalertx_data

test/api_endpoints/test_devices_endpoints.py

@@ -43,6 +43,10 @@ def create_dummy(client, api_token, test_mac):
     client.post(f"/device/{test_mac}", json=payload, headers=auth_headers(api_token))
 
 
+def delete_dummy(client, api_token, test_mac):
+    client.delete("/devices", json={"macs": [test_mac]}, headers=auth_headers(api_token))
+
+
 def test_get_all_devices(client, api_token, test_mac):
     # Ensure there is at least one device
     create_dummy(client, api_token, test_mac)
@@ -149,53 +153,55 @@ def test_export_import_cycle_base64(client, api_token, test_mac):
 
 
 def test_devices_totals(client, api_token, test_mac):
-    # 1. Create a dummy device
     create_dummy(client, api_token, test_mac)
+    try:
+        # 1. Call the totals endpoint
+        resp = client.get("/devices/totals", headers=auth_headers(api_token))
+        assert resp.status_code == 200
 
-    # 2. Call the totals endpoint
-    resp = client.get("/devices/totals", headers=auth_headers(api_token))
-    assert resp.status_code == 200
+        # 2. Ensure the response is a JSON list
+        data = resp.json
+        assert isinstance(data, list)
 
-    # 3. Ensure the response is a JSON list
-    data = resp.json
-    assert isinstance(data, list)
+        # 3. Dynamically get expected length
+        conditions = get_device_conditions()
+        expected_length = len(conditions)
+        assert len(data) == expected_length
 
-    # 4. Dynamically get expected length
-    conditions = get_device_conditions()
-    expected_length = len(conditions)
-    assert len(data) == expected_length
-
-    # 5. Check that at least 1 device exists
-    assert data[0] >= 1  # 'devices' count includes the dummy device
+        # 4. Check that at least 1 device exists
+        assert data[0] >= 1  # 'devices' count includes the dummy device
+    finally:
+        delete_dummy(client, api_token, test_mac)
 
 
 def test_devices_by_status(client, api_token, test_mac):
-    # 1. Create a dummy device
     create_dummy(client, api_token, test_mac)
+    try:
+        # 1. Request devices by a valid status
+        resp = client.get("/devices/by-status?status=my", headers=auth_headers(api_token))
+        assert resp.status_code == 200
+        data = resp.json
+        assert isinstance(data, list)
+        assert any(d["id"] == test_mac for d in data)
 
-    # 2. Request devices by a valid status
-    resp = client.get("/devices/by-status?status=my", headers=auth_headers(api_token))
-    assert resp.status_code == 200
-    data = resp.json
-    assert isinstance(data, list)
-    assert any(d["id"] == test_mac for d in data)
+        # 2. Request devices with an invalid/unknown status
+        resp_invalid = client.get("/devices/by-status?status=invalid_status", headers=auth_headers(api_token))
+        # Strict validation now returns 422 for invalid status enum values
+        assert resp_invalid.status_code == 422
 
-    # 3. Request devices with an invalid/unknown status
-    resp_invalid = client.get("/devices/by-status?status=invalid_status", headers=auth_headers(api_token))
-    # 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
-    client.post(
-        f"/device/{test_mac}",
-        json={"devFavorite": 1},
-        headers=auth_headers(api_token)
-    )
-    resp_fav = client.get("/devices/by-status?status=my", headers=auth_headers(api_token))
-    fav_data = next((d for d in resp_fav.json if d["id"] == test_mac), None)
-    assert fav_data is not None
-    assert "&#9733" in fav_data["title"]
+        # 3. Check favorite formatting if devFavorite = 1
+        # Update dummy device to favorite
+        client.post(
+            f"/device/{test_mac}",
+            json={"devFavorite": 1},
+            headers=auth_headers(api_token)
+        )
+        resp_fav = client.get("/devices/by-status?status=my", headers=auth_headers(api_token))
+        fav_data = next((d for d in resp_fav.json if d["id"] == test_mac), None)
+        assert fav_data is not None
+        assert "&#9733" in fav_data["title"]
+    finally:
+        delete_dummy(client, api_token, test_mac)
 
 
 def test_delete_test_devices(client, api_token):

test/api_endpoints/test_mcp_tools_endpoints.py

@@ -1,6 +1,7 @@
 import pytest
 from unittest.mock import patch, MagicMock
 from datetime import datetime
+import random
 
 from api_server.api_server_start import app
 from helper import get_setting_value
@@ -21,6 +22,21 @@ def auth_headers(token):
     return {"Authorization": f"Bearer {token}"}
 
 
+def create_dummy(client, api_token, test_mac):
+    payload = {
+        "createNew": True,
+        "devName": "Test Device MCP",
+        "devOwner": "Unit Test",
+        "devType": "Router",
+        "devVendor": "TestVendor",
+    }
+    client.post(f"/device/{test_mac}", json=payload, headers=auth_headers(api_token))
+
+
+def delete_dummy(client, api_token, test_mac):
+    client.delete("/devices", json={"macs": [test_mac]}, headers=auth_headers(api_token))
+
+
 # --- Device Search Tests ---
 
 
@@ -350,25 +366,22 @@ def test_mcp_devices_import_json(mock_db_conn, client, api_token):
 # --- MCP Device Totals Tests ---
 
 
-@patch("database.get_temp_db_connection")
-def test_mcp_devices_totals(mock_db_conn, client, api_token):
+def test_mcp_devices_totals(client, api_token):
     """Test MCP devices totals endpoint."""
-    mock_conn = MagicMock()
-    mock_sql = MagicMock()
-    mock_execute_result = MagicMock()
-    # Mock the getTotals method to return sample data
-    mock_execute_result.fetchone.return_value = [10, 8, 2, 0, 1, 3]  # devices, connected, favorites, new, down, archived
-    mock_sql.execute.return_value = mock_execute_result
-    mock_conn.cursor.return_value = mock_sql
-    mock_db_conn.return_value = mock_conn
+    test_mac = "aa:bb:cc:" + ":".join(f"{random.randint(0, 255):02X}" for _ in range(3)).lower()
+    create_dummy(client, api_token, test_mac)
 
-    response = client.get("/devices/totals", headers=auth_headers(api_token))
+    try:
+        response = client.get("/devices/totals", headers=auth_headers(api_token))
 
-    assert response.status_code == 200
-    data = response.get_json()
-    # Should return device counts as array
-    assert isinstance(data, list)
-    assert len(data) >= 4  # At least online, offline, etc.
+        assert response.status_code == 200
+        data = response.get_json()
+        # Should return device counts as array
+        assert isinstance(data, list)
+        assert len(data) >= 4  # At least online, offline, etc.
+        assert data[0] >= 1
+    finally:
+        delete_dummy(client, api_token, test_mac)
 
 
 # --- MCP Traceroute Tests ---

test/docker_tests/test_docker_compose_scenarios.py

@@ -344,6 +344,7 @@ def _write_normal_startup_compose(
 
     service_env = service.setdefault("environment", {})
     service_env.setdefault("NETALERTX_CHECK_ONLY", "1")
+    service_env.setdefault("SKIP_STARTUP_CHECKS", "host optimization")
 
     if env_overrides:
         service_env.update(env_overrides)
@@ -885,9 +886,14 @@ def test_normal_startup_no_warnings_compose(tmp_path: pathlib.Path) -> None:
         f"Unexpected mount row values for /data: {data_parts[2:4]}"
     )
 
+    allowed_warning = "⚠️  WARNING: ARP flux sysctls are not set."
+
     assert "Write permission denied" not in default_output
     assert "CRITICAL" not in default_output
-    assert "⚠️" not in default_output
+    assert all(
+        "⚠️" not in line or allowed_warning in line
+        for line in default_output.splitlines()
+    ), "Unexpected warning found in default output"
 
     custom_http = _select_custom_ports({default_http_port})
     custom_graphql = _select_custom_ports({default_http_port, custom_http})
@@ -922,7 +928,10 @@ def test_normal_startup_no_warnings_compose(tmp_path: pathlib.Path) -> None:
     assert "❌" not in custom_output
     assert "Write permission denied" not in custom_output
     assert "CRITICAL" not in custom_output
-    assert "⚠️" not in custom_output
+    assert all(
+        "⚠️" not in line or allowed_warning in line
+        for line in custom_output.splitlines()
+    ), "Unexpected warning found in custom output"
     lowered_custom = custom_output.lower()
     assert "arning" not in lowered_custom
     assert "rror" not in lowered_custom

test/docker_tests/test_entrypoint.py

@@ -90,3 +90,40 @@ def test_skip_startup_checks_env_var():
     result = _run_entrypoint(env={"SKIP_STARTUP_CHECKS": "mandatory folders"}, check_only=True)
     assert "Creating NetAlertX log directory" not in result.stdout
     assert result.returncode == 0
+
+
+@pytest.mark.docker
+@pytest.mark.feature_complete
+def test_host_optimization_warning_matches_sysctl():
+    """Validate host-optimization warning matches actual host sysctl values."""
+    ignore_proc = subprocess.run(
+        ["sysctl", "-n", "net.ipv4.conf.all.arp_ignore"],
+        capture_output=True,
+        text=True,
+        check=False,
+        timeout=10,
+    )
+    announce_proc = subprocess.run(
+        ["sysctl", "-n", "net.ipv4.conf.all.arp_announce"],
+        capture_output=True,
+        text=True,
+        check=False,
+        timeout=10,
+    )
+
+    if ignore_proc.returncode != 0 or announce_proc.returncode != 0:
+        pytest.skip("sysctl values unavailable on host; skipping host-optimization warning check")
+
+    arp_ignore = ignore_proc.stdout.strip()
+    arp_announce = announce_proc.stdout.strip()
+    expected_warning = not (arp_ignore == "1" and arp_announce == "2")
+
+    result = _run_entrypoint(check_only=True)
+    combined_output = result.stdout + result.stderr
+    warning_present = "WARNING: ARP flux sysctls are not set." in combined_output
+
+    assert warning_present == expected_warning, (
+        "host-optimization warning mismatch: "
+        f"arp_ignore={arp_ignore}, arp_announce={arp_announce}, "
+        f"expected_warning={expected_warning}, warning_present={warning_present}"
+    )