feat(ping): add a generic ping sweep every driver inherits

Sweeping a range is orchestration, not device mechanics: the only vendor-specific part is executing a single ping, and NAPALM already standardises that. PingSweepMixin therefore owns the loop, the reply parsing, the target cap and the progress reporting, and is mixed into DeviceTypeDriver so any driver implementing ping() becomes a usable sweep source without writing sweep code of its own. driver_supports_ping() answers "can this driver ping?" by introspection instead of a hand-maintained list, with SUPPORTS_PING = False as the opt-out for a driver that inherits a ping it cannot actually use. The generic implementation is deliberately sequential — a NAPALM connection is a single session and not safe to drive from several threads at once. A driver whose device offers something faster overrides ping_sweep and keeps the return shape; see napalm-opnsense's batched job API version.
2026-08-13 16:50:47 +07:00
parent 90b8e08789
commit a211629875
6 changed files with 445 additions and 3 deletions
@@ -0,0 +1,221 @@
+"""Tests for the generic ping sweep (PingSweepMixin + driver_supports_ping)."""
+
+import pytest
+from napalm.base import NetworkDriver
+
+from napalm_device_types import DeviceTypeDriver, PingSweepMixin, driver_supports_ping
+
+
+# ── Fakes ─────────────────────────────────────────────────────────────────────
+
+
+def _napalm_ok(rtt: float, probes: int = 1) -> dict:
+    """A NAPALM-format ping() reply for a reachable destination."""
+    return {
+        "success": {
+            "probes_sent": probes,
+            "packet_loss": 0,
+            "rtt_min": rtt,
+            "rtt_avg": rtt,
+            "rtt_max": rtt,
+            "rtt_stddev": 0.0,
+            "results": [{"ip_address": "10.0.0.1", "rtt": rtt}],
+        }
+    }
+
+
+def _napalm_lost(probes: int = 1) -> dict:
+    """A NAPALM-format ping() reply where every probe was lost."""
+    return {
+        "success": {
+            "probes_sent": probes,
+            "packet_loss": probes,
+            "rtt_min": 0.0,
+            "rtt_avg": 0.0,
+            "rtt_max": 0.0,
+            "rtt_stddev": 0.0,
+            "results": [],
+        }
+    }
+
+
+class FakePingDriver(PingSweepMixin):
+    """Minimal driver exposing ping() — stands in for a real vendor driver."""
+
+    def __init__(self, replies=None, raises=None):
+        self.replies = replies or {}
+        self.raises = raises or {}
+        self.calls = []
+
+    def ping(self, destination, source="", ttl=255, timeout=2, size=100, count=5, vrf=""):
+        self.calls.append({"destination": destination, "timeout": timeout, "count": count})
+        if destination in self.raises:
+            raise self.raises[destination]
+        return self.replies.get(destination, _napalm_lost())
+
+
+class NoPingDriver(PingSweepMixin):
+    """Driver without its own ping() — inherits NAPALM's NotImplementedError stub."""
+
+    ping = NetworkDriver.ping
+
+
+class OptedOutDriver(FakePingDriver):
+    """Driver that implements ping() but declares it unusable for sweeps."""
+
+    SUPPORTS_PING = False
+
+
+# ── driver_supports_ping ──────────────────────────────────────────────────────
+
+
+def test_driver_supports_ping_false_for_unoverridden_ping():
+    assert driver_supports_ping(NoPingDriver) is False
+
+
+def test_driver_supports_ping_false_for_base_network_driver():
+    assert driver_supports_ping(NetworkDriver) is False
+
+
+def test_driver_supports_ping_true_when_overridden():
+    assert driver_supports_ping(FakePingDriver) is True
+
+
+def test_driver_supports_ping_honours_explicit_opt_out():
+    assert driver_supports_ping(OptedOutDriver) is False
+
+
+def test_driver_supports_ping_false_for_class_without_ping():
+    class Bare:
+        pass
+
+    assert driver_supports_ping(Bare) is False
+
+
+def test_device_type_driver_exposes_supports_ping_classmethod():
+    assert DeviceTypeDriver.supports_ping() is False
+    assert FakePingDriver.supports_ping() is True
+
+
+# ── ping_sweep ────────────────────────────────────────────────────────────────
+
+
+def test_ping_sweep_marks_reachable_and_unreachable_hosts():
+    driver = FakePingDriver(replies={"10.0.0.1": _napalm_ok(1.5)})
+
+    result = driver.ping_sweep(["10.0.0.1", "10.0.0.2"])
+
+    assert result["scanned"] == 2
+    assert result["alive_count"] == 1
+    assert result["truncated"] is False
+    assert result["entries"] == [
+        {"ip": "10.0.0.1", "alive": True, "rtt_ms": 1.5},
+        {"ip": "10.0.0.2", "alive": False, "rtt_ms": None},
+    ]
+
+
+def test_ping_sweep_uses_single_fast_probe_by_default():
+    driver = FakePingDriver()
+
+    driver.ping_sweep(["10.0.0.1"])
+
+    assert driver.calls == [{"destination": "10.0.0.1", "timeout": 1, "count": 1}]
+
+
+def test_ping_sweep_forwards_count_and_timeout():
+    driver = FakePingDriver()
+
+    driver.ping_sweep(["10.0.0.1"], count=3, timeout=5)
+
+    assert driver.calls == [{"destination": "10.0.0.1", "timeout": 5, "count": 3}]
+
+
+def test_ping_sweep_treats_error_reply_as_unreachable():
+    driver = FakePingDriver(replies={"10.0.0.9": {"error": "unknown host"}})
+
+    result = driver.ping_sweep(["10.0.0.9"])
+
+    assert result["entries"][0]["alive"] is False
+    assert result["entries"][0]["error"] == "unknown host"
+    assert result["alive_count"] == 0
+
+
+def test_ping_sweep_records_exception_and_continues():
+    driver = FakePingDriver(
+        replies={"10.0.0.2": _napalm_ok(2.0)},
+        raises={"10.0.0.1": RuntimeError("session closed")},
+    )
+
+    result = driver.ping_sweep(["10.0.0.1", "10.0.0.2"])
+
+    assert result["entries"][0] == {
+        "ip": "10.0.0.1",
+        "alive": False,
+        "rtt_ms": None,
+        "error": "session closed",
+    }
+    assert result["entries"][1]["alive"] is True
+    assert result["scanned"] == 2
+
+
+def test_ping_sweep_truncates_at_max_targets():
+    driver = FakePingDriver()
+
+    result = driver.ping_sweep([f"10.0.0.{i}" for i in range(1, 11)], max_targets=4)
+
+    assert result["scanned"] == 4
+    assert result["truncated"] is True
+    assert len(driver.calls) == 4
+
+
+def test_ping_sweep_respects_class_level_max_targets():
+    class SmallSweepDriver(FakePingDriver):
+        PING_SWEEP_MAX_TARGETS = 2
+
+    driver = SmallSweepDriver()
+
+    result = driver.ping_sweep(["10.0.0.1", "10.0.0.2", "10.0.0.3"])
+
+    assert result["scanned"] == 2
+    assert result["truncated"] is True
+
+
+def test_ping_sweep_reports_progress_per_destination():
+    driver = FakePingDriver(replies={"10.0.0.1": _napalm_ok(1.0)})
+    seen = []
+
+    driver.ping_sweep(
+        ["10.0.0.1", "10.0.0.2"],
+        on_progress=lambda done, total: seen.append((done, total)),
+    )
+
+    assert seen == [(1, 2), (2, 2)]
+
+
+def test_ping_sweep_on_empty_destination_list():
+    driver = FakePingDriver()
+
+    result = driver.ping_sweep([])
+
+    assert result == {"entries": [], "scanned": 0, "alive_count": 0, "truncated": False}
+
+
+def test_ping_sweep_raises_when_driver_cannot_ping():
+    driver = NoPingDriver()
+
+    with pytest.raises(NotImplementedError):
+        driver.ping_sweep(["10.0.0.1"])
+
+
+def test_ping_sweep_stops_when_stop_requested():
+    driver = FakePingDriver()
+    calls = {"n": 0}
+
+    def _should_stop():
+        calls["n"] += 1
+        return calls["n"] > 1
+
+    result = driver.ping_sweep(["10.0.0.1", "10.0.0.2", "10.0.0.3"], should_stop=_should_stop)
+
+    assert result["scanned"] == 1
+    assert len(driver.calls) == 1