More updates to v1.0.0

Author: pdudotdev

testing/automated/test_input_models.py

@@ -2,7 +2,7 @@
 import pytest
 from pydantic import ValidationError
 
-from input_models.models import InterfacesQuery, IntentQuery, KBQuery, OspfQuery, RoutingQuery
+from input_models.models import InterfacesQuery, IntentQuery, KBQuery, OspfQuery, RoutingQuery, TracerouteInput
 
 
 class TestOspfQuery:
@@ -72,6 +72,33 @@ def test_valid(self):
         assert q.device == "R1"
 
 
+class TestTracerouteInput:
+    def test_valid_minimal(self):
+        q = TracerouteInput(device="R1", destination="10.0.0.1")
+        assert q.device == "R1"
+        assert q.destination == "10.0.0.1"
+        assert q.source is None
+        assert q.vrf is None
+
+    def test_valid_full(self):
+        q = TracerouteInput(device="R1", destination="10.0.0.1", source="192.168.1.1", vrf="VRF1")
+        assert q.source == "192.168.1.1"
+        assert q.vrf == "VRF1"
+
+    def test_vrf_injection_blocked(self):
+        with pytest.raises(ValidationError):
+            TracerouteInput(device="R1", destination="10.0.0.1", vrf="VRF1; reboot")
+
+    def test_json_string_parsing(self):
+        q = TracerouteInput.model_validate('{"device": "R1", "destination": "10.0.0.1"}')
+        assert q.device == "R1"
+        assert q.destination == "10.0.0.1"
+
+    def test_missing_destination_rejected(self):
+        with pytest.raises(ValidationError):
+            TracerouteInput(device="R1")
+
+
 class TestKBQuery:
     def test_valid_full(self):
         q = KBQuery(query="OSPF timers", vendor="cisco_ios", topic="rfc", top_k=3)

testing/automated/test_intent_tool.py

@@ -1,8 +1,6 @@
 """UT-014: query_intent tool."""
 from unittest.mock import patch
 
-import pytest
-
 from input_models.models import IntentQuery
 from tools.intent import query_intent
 

testing/automated/test_list_devices.py

@@ -1,6 +1,4 @@
 """UT-011: list_devices tool."""
-import pytest
-
 from input_models.models import DeviceListQuery
 from tools.inventory_tool import list_devices
 

testing/automated/test_mcp_server.py

@@ -2,14 +2,14 @@
 
 
 class TestMcpToolRegistration:
-    async def test_seven_tools_registered(self):
-        """The MCP server must expose exactly 7 tools."""
+    async def test_eight_tools_registered(self):
+        """The MCP server must expose exactly 8 tools."""
         from server.MCPServer import mcp
         tools = await mcp.list_tools()
-        assert len(tools) == 7
+        assert len(tools) == 8
 
     async def test_tool_names(self):
-        """All seven expected tool names are registered."""
+        """All eight expected tool names are registered."""
         from server.MCPServer import mcp
         tools = await mcp.list_tools()
         names = {t.name for t in tools}
@@ -21,4 +21,5 @@ async def test_tool_names(self):
             "query_intent",
             "get_status",
             "list_devices",
+            "traceroute",
         }

testing/automated/test_platform_map.py

@@ -26,6 +26,10 @@ def test_routing_queries_complete(self, cli_style):
     def test_interfaces_present(self, cli_style):
         assert "interface_status" in PLATFORM_MAP[cli_style]["interfaces"]
 
+    @pytest.mark.parametrize("cli_style", EXPECTED_CLI_STYLES)
+    def test_traceroute_present(self, cli_style):
+        assert "traceroute" in PLATFORM_MAP[cli_style]["tools"], f"{cli_style} missing traceroute"
+
 
 class TestApplyVrf:
     def test_dict_with_vrf(self):
@@ -78,6 +82,39 @@ def test_routeros_ip_route(self):
         assert result == "/ip route print without-paging"
 
 
+class TestTracerouteVrf:
+    def test_ios_traceroute_with_vrf(self):
+        device = {"cli_style": "ios"}
+        result = get_action(device, "tools", "traceroute", vrf="VRF1")
+        assert result == "traceroute vrf VRF1"
+
+    def test_ios_traceroute_no_vrf(self):
+        device = {"cli_style": "ios"}
+        result = get_action(device, "tools", "traceroute")
+        assert result == "traceroute"
+
+    def test_eos_traceroute_with_vrf(self):
+        device = {"cli_style": "eos"}
+        result = get_action(device, "tools", "traceroute", vrf="VRF1")
+        assert result == "traceroute vrf VRF1"
+
+    def test_junos_traceroute_with_vrf(self):
+        device = {"cli_style": "junos"}
+        result = get_action(device, "tools", "traceroute", vrf="VRF1")
+        assert result == "traceroute routing-instance VRF1"
+
+    def test_routeros_traceroute_no_vrf_variant(self):
+        device = {"cli_style": "routeros"}
+        result = get_action(device, "tools", "traceroute")
+        assert result == "/tool/traceroute"
+
+    def test_routeros_traceroute_ignores_vrf(self):
+        """RouterOS has no VRF variant — VRF is silently ignored."""
+        device = {"cli_style": "routeros"}
+        result = get_action(device, "tools", "traceroute", vrf="VRF1")
+        assert result == "/tool/traceroute"
+
+
 class TestGetAction:
     def test_resolves_ios_ospf_neighbors(self):
         device = {"cli_style": "ios"}

testing/automated/test_security.py

@@ -7,7 +7,7 @@
 import pytest
 from pydantic import ValidationError
 
-from input_models.models import OspfQuery, RoutingQuery
+from input_models.models import OspfQuery, RoutingQuery, TracerouteInput
 
 
 # ── VRF injection: each pattern MUST be rejected by the Pydantic validator ────
@@ -48,6 +48,13 @@ def test_vrf_injection_blocked_routing(bad_vrf):
         RoutingQuery(device="R1", query="ip_route", vrf=bad_vrf)
 
 
+@pytest.mark.parametrize("bad_vrf", _INJECTION_PATTERNS)
+def test_vrf_injection_blocked_traceroute(bad_vrf):
+    """Every injection payload must raise ValidationError in TracerouteInput."""
+    with pytest.raises(ValidationError):
+        TracerouteInput(device="R1", destination="10.0.0.1", vrf=bad_vrf)
+
+
 # ── VRF positive cases: valid names MUST be accepted ─────────────────────────
 
 _VALID_VRFS = [
@@ -70,3 +77,9 @@ def test_vrf_valid_names_accepted_ospf(good_vrf):
 def test_vrf_valid_names_accepted_routing(good_vrf):
     q = RoutingQuery(device="R1", query="ip_route", vrf=good_vrf)
     assert q.vrf == good_vrf
+
+
+@pytest.mark.parametrize("good_vrf", _VALID_VRFS)
+def test_vrf_valid_names_accepted_traceroute(good_vrf):
+    q = TracerouteInput(device="R1", destination="10.0.0.1", vrf=good_vrf)
+    assert q.vrf == good_vrf

testing/automated/test_tools.py

@@ -1,11 +1,11 @@
-"""UT-003: Tool layer — get_ospf, get_interfaces, _error_response, and VRF integration."""
+"""UT-003: Tool layer — get_ospf, get_interfaces, traceroute, _error_response, and VRF integration."""
 from unittest.mock import AsyncMock, patch
 
 import pytest
 
-from input_models.models import InterfacesQuery, OspfQuery
+from input_models.models import InterfacesQuery, OspfQuery, TracerouteInput
 from tools import _error_response
-from tools.operational import get_interfaces
+from tools.operational import get_interfaces, traceroute
 from tools.ospf import get_ospf
 
 
@@ -72,6 +72,48 @@ async def test_valid_device_junos(self):
         assert result["cli_style"] == "junos"  # R3 is junos in MOCK_DEVICES
 
 
+class TestTraceroute:
+    async def test_unknown_device(self):
+        result = await traceroute(TracerouteInput(device="NONEXISTENT", destination="10.0.0.1"))
+        assert "error" in result
+        assert "Unknown device" in result["error"]
+
+    async def test_valid_device_ios_basic(self):
+        with patch("transport.execute_ssh", new_callable=AsyncMock) as mock_ssh:
+            mock_ssh.return_value = "1  10.0.0.254  1 ms"
+            result = await traceroute(TracerouteInput(device="R1", destination="10.0.0.1"))
+        assert result["device"] == "R1"
+        assert result["cli_style"] == "ios"
+        assert "traceroute" in result["_command"]
+        assert "10.0.0.1" in result["_command"]
+
+    async def test_ios_source_appended(self):
+        with patch("transport.execute_ssh", new_callable=AsyncMock) as mock_ssh:
+            mock_ssh.return_value = "1  10.0.0.254  1 ms"
+            result = await traceroute(TracerouteInput(device="R1", destination="10.0.0.1", source="192.168.1.1"))
+        assert "source 192.168.1.1" in result["_command"]
+
+    async def test_eos_vrf_in_command(self):
+        with patch("transport.execute_ssh", new_callable=AsyncMock) as mock_ssh:
+            mock_ssh.return_value = "mock output"
+            result = await traceroute(TracerouteInput(device="R2", destination="10.0.0.1", vrf="VRF1"))
+        assert "vrf VRF1" in result["_command"]
+        assert "10.0.0.1" in result["_command"]
+
+    async def test_routeros_address_syntax(self):
+        with patch("transport.execute_ssh", new_callable=AsyncMock) as mock_ssh:
+            mock_ssh.return_value = "mock output"
+            result = await traceroute(TracerouteInput(device="R5", destination="10.0.0.1"))
+        assert "address=10.0.0.1" in result["_command"]
+
+    async def test_routeros_source_syntax(self):
+        with patch("transport.execute_ssh", new_callable=AsyncMock) as mock_ssh:
+            mock_ssh.return_value = "mock output"
+            result = await traceroute(TracerouteInput(device="R5", destination="10.0.0.1", source="192.168.1.1"))
+        assert "address=10.0.0.1" in result["_command"]
+        assert "src-address=192.168.1.1" in result["_command"]
+
+
 class TestVrfEndToEnd:
     async def test_explicit_vrf_in_final_command(self):
         """VRF flows from validated input → get_action substitution → final CLI command."""

testing/live/test_platform_coverage.py

@@ -1,4 +1,4 @@
-"""LT-001: Live platform coverage — all vendors × all OSPF queries + interfaces.
+"""LT-001: Live platform coverage — all vendors × all OSPF, routing_table, tools + interfaces.
 
 Generates platform_coverage_results.md with detailed output per test.
 """
@@ -7,9 +7,10 @@
 import pytest
 
 from core.inventory import devices
-from input_models.models import InterfacesQuery, OspfQuery
-from tools.operational import get_interfaces
+from input_models.models import InterfacesQuery, OspfQuery, RoutingQuery, TracerouteInput
+from tools.operational import get_interfaces, traceroute
 from tools.ospf import get_ospf
+from tools.routing import get_routing
 
 # ── Test devices: one per vendor ─────────────────────────────────────────────
 TEST_DEVICES = {
@@ -21,6 +22,9 @@
 }
 
 OSPF_QUERIES = ["neighbors", "database", "borders", "config", "interfaces", "details"]
+ROUTING_QUERIES = ["ip_route", "route_maps", "prefix_lists", "policy_based_routing", "access_lists"]
+TRACEROUTE_DEST = "172.20.20.207"  # C1J management IP — core device reachable from all vendors
+
 RESULTS = []  # collected during test run
 
 
@@ -124,10 +128,29 @@ async def test_ospf_query(device, query):
     assert status != "FAIL", f"{device} ospf/{query}: {result.get('error', result.get('raw', '')[:200])}"
 
 
+@pytest.mark.parametrize("device", TEST_DEVICES.keys())
+@pytest.mark.parametrize("query", ROUTING_QUERIES)
+async def test_routing_query(device, query):
+    """Test routing_table query against a live device."""
+    result = await get_routing(RoutingQuery(device=device, query=query))
+    status = classify(result)
+    record(device, "routing_table", query, result, status)
+    assert status != "FAIL", f"{device} routing_table/{query}: {result.get('error', result.get('raw', '')[:200])}"
+
+
 @pytest.mark.parametrize("device", TEST_DEVICES.keys())
 async def test_interfaces(device):
     """Test interface status query against a live device."""
     result = await get_interfaces(InterfacesQuery(device=device))
     status = classify(result)
     record(device, "interfaces", "interface_status", result, status)
     assert status != "FAIL", f"{device} interfaces: {result.get('error', result.get('raw', '')[:200])}"
+
+
+@pytest.mark.parametrize("device", TEST_DEVICES.keys())
+async def test_traceroute(device):
+    """Test traceroute against a live device, tracing to C1J management IP."""
+    result = await traceroute(TracerouteInput(device=device, destination=TRACEROUTE_DEST))
+    status = classify(result)
+    record(device, "tools", "traceroute", result, status)
+    assert status != "FAIL", f"{device} traceroute→{TRACEROUTE_DEST}: {result.get('error', result.get('raw', '')[:200])}"

testing/run_tests.sh

@@ -67,7 +67,7 @@ echo ""
 printf "%sLive Tests%s\n" "$BOLD" "$NC"
 echo "-----------------------------------------"
 if [[ "${1:-}" == "--live" ]]; then
-    run_suite "LT-001" "Platform Coverage (5 vendors)" "testing/live/test_platform_coverage.py" "--live"
+    run_suite "LT-001" "Platform Coverage (5 vendors × OSPF+routing+traceroute+interfaces)" "testing/live/test_platform_coverage.py" "--live"
     echo ""
     echo "Report: testing/live/platform_coverage_results.md"
 else