test(plugins): add parser unit tests for scapy_recon plugin (#1431)

testing/backend/test_scapy_recon_parser.py

@@ -0,0 +1,350 @@
+import json
+import sys
+from pathlib import Path
+
+import pytest
+
+REPO_ROOT = Path(__file__).resolve().parents[2]
+sys.path.insert(0, str(REPO_ROOT))
+
+from plugins.scapy_recon.parser import parse
+
+PLUGIN_DIR = REPO_ROOT / "plugins" / "scapy_recon"
+
+
+# ---------------------------------------------------------------------------
+# Metadata contract tests
+# ---------------------------------------------------------------------------
+
+
+def test_scapy_recon_metadata_file_exists():
+    """metadata.json must exist at the expected plugin path."""
+    assert (PLUGIN_DIR / "metadata.json").exists(), (
+        "metadata.json not found in plugins/scapy_recon/"
+    )
+
+
+def test_scapy_recon_parser_file_exists():
+    """parser.py must exist alongside metadata.json."""
+    assert (PLUGIN_DIR / "parser.py").exists(), (
+        "parser.py not found in plugins/scapy_recon/"
+    )
+
+
+def test_scapy_recon_metadata_is_valid_json():
+    """metadata.json must be valid, parseable JSON."""
+    raw = (PLUGIN_DIR / "metadata.json").read_text(encoding="utf-8")
+    data = json.loads(raw)
+    assert isinstance(data, dict)
+
+
+def test_scapy_recon_metadata_id_matches_directory():
+    """Plugin id in metadata.json must match the directory name."""
+    data = json.loads((PLUGIN_DIR / "metadata.json").read_text(encoding="utf-8"))
+    assert data["id"] == "scapy_recon"
+
+
+def test_scapy_recon_metadata_has_required_top_level_fields():
+    """metadata.json must contain all required top-level keys."""
+    data = json.loads((PLUGIN_DIR / "metadata.json").read_text(encoding="utf-8"))
+    for key in ("id", "name", "version", "description", "engine", "fields", "output"):
+        assert key in data, f"metadata.json missing required key: {key}"
+
+
+def test_scapy_recon_metadata_output_parser_is_custom():
+    """Output parser type must be 'custom', backed by parser.py."""
+    data = json.loads((PLUGIN_DIR / "metadata.json").read_text(encoding="utf-8"))
+    assert data["output"]["parser"] == "custom"
+
+
+def test_scapy_recon_metadata_engine_binary_is_python3():
+    """Engine binary must be 'python3' (Scapy runs via Python)."""
+    data = json.loads((PLUGIN_DIR / "metadata.json").read_text(encoding="utf-8"))
+    assert data["engine"]["type"] == "cli"
+    assert data["engine"]["binary"] == "python3"
+
+
+def test_scapy_recon_metadata_has_required_target_field():
+    """Plugin must declare a required 'target' field."""
+    data = json.loads((PLUGIN_DIR / "metadata.json").read_text(encoding="utf-8"))
+    fields = {f["id"]: f for f in data["fields"]}
+    assert "target" in fields, "Missing required field: target"
+    assert fields["target"]["required"] is True
+
+
+def test_scapy_recon_metadata_has_type_field_with_defaults():
+    """Plugin must declare a 'type' field with valid probe-type options."""
+    data = json.loads((PLUGIN_DIR / "metadata.json").read_text(encoding="utf-8"))
+    fields = {f["id"]: f for f in data["fields"]}
+    assert "type" in fields, "Missing 'type' selector field"
+    type_field = fields["type"]
+    option_values = [opt["value"] for opt in type_field.get("options", [])]
+    assert "arp_ping" in option_values, "Expected 'arp_ping' probe type option"
+    assert "icmp_ping" in option_values, "Expected 'icmp_ping' probe type option"
+
+
+# ---------------------------------------------------------------------------
+# Parser tests -- normal / representative sample output
+# ---------------------------------------------------------------------------
+
+_ARP_OUTPUT = (
+    "UP: 192.168.1.1 - aa:bb:cc:dd:ee:ff\n"
+    "UP: 192.168.1.10 - 11:22:33:44:55:66\n"
+    "UP: 10.0.0.5 - de:ad:be:ef:00:01\n"
+)
+
+_ICMP_OUTPUT = (
+    "UP: 192.168.1.1\n"
+    "UP: 192.168.1.20\n"
+)
+
+
+def test_scapy_recon_parser_returns_dict():
+    """parse() must return a dict."""
+    result = parse(_ARP_OUTPUT)
+    assert isinstance(result, dict)
+
+
+def test_scapy_recon_parser_returns_required_keys():
+    """parse() must return a dict with 'findings', 'count', and 'hosts' keys."""
+    result = parse(_ARP_OUTPUT)
+    assert "findings" in result, "Missing 'findings' key"
+    assert "count" in result, "Missing 'count' key"
+    assert "hosts" in result, "Missing 'hosts' key"
+
+
+def test_scapy_recon_parser_count_matches_hosts():
+    """'count' must equal the number of hosts discovered."""
+    result = parse(_ARP_OUTPUT)
+    assert result["count"] == len(result["hosts"])
+
+
+def test_scapy_recon_parser_arp_host_count():
+    """Parser must detect all three hosts from the ARP sample output."""
+    result = parse(_ARP_OUTPUT)
+    assert result["count"] == 3
+    assert len(result["findings"]) == 3
+
+
+def test_scapy_recon_parser_arp_host_ip_and_mac():
+    """Parser must correctly extract IP and MAC from an ARP-style line."""
+    result = parse(_ARP_OUTPUT)
+    hosts = {h["ip"]: h["mac"] for h in result["hosts"]}
+
+    assert "192.168.1.1" in hosts
+    assert hosts["192.168.1.1"] == "aa:bb:cc:dd:ee:ff"
+
+    assert "192.168.1.10" in hosts
+    assert hosts["192.168.1.10"] == "11:22:33:44:55:66"
+
+    assert "10.0.0.5" in hosts
+    assert hosts["10.0.0.5"] == "de:ad:be:ef:00:01"
+
+
+def test_scapy_recon_parser_finding_has_required_keys():
+    """Each finding must contain title, category, severity, description, remediation, metadata."""
+    result = parse(_ARP_OUTPUT)
+    assert result["findings"], "Expected at least one finding"
+    for finding in result["findings"]:
+        for key in ("title", "category", "severity", "description", "remediation", "metadata"):
+            assert key in finding, f"Finding missing required key: {key}"
+
+
+def test_scapy_recon_parser_finding_title_contains_ip():
+    """Each finding title must reference the discovered host IP."""
+    result = parse(_ARP_OUTPUT)
+    host_ips = {h["ip"] for h in result["hosts"]}
+    for finding in result["findings"]:
+        assert any(ip in finding["title"] for ip in host_ips), (
+            f"Finding title '{finding['title']}' does not reference any known host IP"
+        )
+
+
+def test_scapy_recon_parser_finding_category_is_network_discovery():
+    """All findings must use 'Network Discovery' as their category."""
+    result = parse(_ARP_OUTPUT)
+    for finding in result["findings"]:
+        assert finding["category"] == "Network Discovery", (
+            f"Unexpected category: {finding['category']}"
+        )
+
+
+def test_scapy_recon_parser_finding_severity_is_info():
+    """All findings must use 'info' severity (host discovery is informational)."""
+    result = parse(_ARP_OUTPUT)
+    for finding in result["findings"]:
+        assert finding["severity"] == "info", (
+            f"Unexpected severity: {finding['severity']}"
+        )
+
+
+def test_scapy_recon_parser_finding_metadata_contains_ip_and_mac():
+    """Each finding's metadata must carry both 'ip' and 'mac' keys."""
+    result = parse(_ARP_OUTPUT)
+    for finding in result["findings"]:
+        assert "ip" in finding["metadata"], "Finding metadata missing 'ip'"
+        assert "mac" in finding["metadata"], "Finding metadata missing 'mac'"
+
+
+def test_scapy_recon_parser_finding_metadata_ip_matches_host():
+    """The 'ip' stored in each finding's metadata must match a host entry."""
+    result = parse(_ARP_OUTPUT)
+    known_ips = {h["ip"] for h in result["hosts"]}
+    for finding in result["findings"]:
+        assert finding["metadata"]["ip"] in known_ips, (
+            f"finding metadata ip '{finding['metadata']['ip']}' not in host list"
+        )
+
+
+def test_scapy_recon_parser_finding_has_remediation():
+    """Each finding must include a non-empty remediation string."""
+    result = parse(_ARP_OUTPUT)
+    for finding in result["findings"]:
+        assert isinstance(finding["remediation"], str)
+        assert len(finding["remediation"].strip()) > 0, "Remediation must not be blank"
+
+
+def test_scapy_recon_parser_finding_description_contains_ip_and_mac():
+    """Description for ARP hosts must mention both IP and MAC."""
+    result = parse(_ARP_OUTPUT)
+    for finding in result["findings"]:
+        ip = finding["metadata"]["ip"]
+        mac = finding["metadata"]["mac"]
+        desc = finding["description"]
+        assert ip in desc, f"Description missing IP '{ip}': {desc}"
+        assert mac in desc, f"Description missing MAC '{mac}': {desc}"
+
+
+# ---------------------------------------------------------------------------
+# Parser tests -- ICMP output (host line with no MAC address)
+# ---------------------------------------------------------------------------
+
+
+def test_scapy_recon_parser_icmp_output_host_count():
+    """Parser must detect both hosts from the ICMP sample output."""
+    result = parse(_ICMP_OUTPUT)
+    assert result["count"] == 2
+    assert len(result["findings"]) == 2
+
+
+def test_scapy_recon_parser_icmp_missing_mac_defaults_to_unknown():
+    """When a line contains no MAC address, mac must default to 'Unknown'."""
+    result = parse(_ICMP_OUTPUT)
+    for host in result["hosts"]:
+        assert host["mac"] == "Unknown", (
+            f"Expected 'Unknown' mac for ICMP-style line, got '{host['mac']}'"
+        )
+
+
+def test_scapy_recon_parser_icmp_ips_are_extracted():
+    """Parser must correctly extract IPs from ICMP-style lines (no MAC)."""
+    result = parse(_ICMP_OUTPUT)
+    ips = {h["ip"] for h in result["hosts"]}
+    assert "192.168.1.1" in ips
+    assert "192.168.1.20" in ips
+
+
+# ---------------------------------------------------------------------------
+# Parser tests -- single host output
+# ---------------------------------------------------------------------------
+
+
+def test_scapy_recon_parser_single_host_with_mac():
+    """Parser handles a single ARP-style line correctly."""
+    output = "UP: 172.16.0.1 - 00:11:22:33:44:55\n"
+    result = parse(output)
+    assert result["count"] == 1
+    assert len(result["findings"]) == 1
+    assert result["hosts"][0]["ip"] == "172.16.0.1"
+    assert result["hosts"][0]["mac"] == "00:11:22:33:44:55"
+
+
+def test_scapy_recon_parser_single_host_description_contains_ip_and_mac():
+    """Description for a single host must mention both IP and MAC."""
+    output = "UP: 10.10.10.1 - ff:ee:dd:cc:bb:aa\n"
+    result = parse(output)
+    finding = result["findings"][0]
+    assert "10.10.10.1" in finding["description"]
+    assert "ff:ee:dd:cc:bb:aa" in finding["description"]
+
+
+# ---------------------------------------------------------------------------
+# Parser tests -- empty / malformed output (must not crash)
+# ---------------------------------------------------------------------------
+
+
+def test_scapy_recon_parser_empty_string():
+    """Parser must not crash on empty input and must return empty results."""
+    result = parse("")
+    assert isinstance(result, dict)
+    assert result["findings"] == []
+    assert result["count"] == 0
+    assert result["hosts"] == []
+
+
+def test_scapy_recon_parser_whitespace_only():
+    """Parser must handle whitespace-only input gracefully."""
+    result = parse("   \n\n\t  \n")
+    assert isinstance(result, dict)
+    assert result["findings"] == []
+    assert result["count"] == 0
+    assert result["hosts"] == []
+
+
+def test_scapy_recon_parser_no_up_lines():
+    """Parser must return empty results when no 'UP:' lines are present."""
+    output = (
+        "Starting Scapy scan...\n"
+        "Sending ARP requests to 192.168.1.0/24\n"
+        "Scan complete. No live hosts found.\n"
+    )
+    result = parse(output)
+    assert result["findings"] == []
+    assert result["count"] == 0
+    assert result["hosts"] == []
+
+
+def test_scapy_recon_parser_mixed_valid_and_noise_lines():
+    """Parser must extract only 'UP:' lines and ignore all other content."""
+    output = (
+        "Starting Scapy scan...\n"
+        "UP: 192.168.1.5 - 00:aa:bb:cc:dd:ee\n"
+        "Some random debug line\n"
+        "UP: 192.168.1.9 - ff:00:11:22:33:44\n"
+        "Scan finished.\n"
+    )
+    result = parse(output)
+    assert result["count"] == 2
+    ips = {h["ip"] for h in result["hosts"]}
+    assert "192.168.1.5" in ips
+    assert "192.168.1.9" in ips
+
+
+def test_scapy_recon_parser_malformed_up_line_no_crash():
+    """Parser must not crash when a 'UP:' line lacks the expected format."""
+    malformed_outputs = [
+        "UP:\n",
+        "UP:  \n",
+        "UP: \n",
+        "UP: no-dash-separator\n",
+        "UP: 999.999.999.999 - INVALID:MAC:HERE\n",
+    ]
+    for bad_output in malformed_outputs:
+        try:
+            result = parse(bad_output)
+            assert isinstance(result, dict), (
+                f"parse() returned non-dict for input: {repr(bad_output)}"
+            )
+        except Exception as exc:
+            pytest.fail(
+                f"parse() raised {type(exc).__name__} for input {repr(bad_output)}: {exc}"
+            )
+
+
+def test_scapy_recon_parser_line_missing_mac_does_not_crash():
+    """A 'UP:' line without a ' - <mac>' separator must be handled gracefully."""
+    output = "UP: 192.168.1.1\n"
+    result = parse(output)
+    assert isinstance(result, dict)
+    assert result["count"] == 1
+    assert result["hosts"][0]["mac"] == "Unknown"