diff --git a/README.md b/README.md index ce8683d..43a7809 100644 --- a/README.md +++ b/README.md @@ -87,9 +87,35 @@ arguments. | `get_bgp_neighbors` | ✅ ² | `GET /api/quagga/bgp/get` + `GET /api/quagga/diagnostics/bgpneighbors` | | `get_vlans` | ✅ | `GET /api/interfaces/vlan_settings/search_item` | | `get_mac_address_table` | ❌ | Not applicable (firewall, no L2 switching) | +| `ping` | ✅ | `POST /api/diagnostics/ping/set` + `start` + `search_jobs` + `stop`/`remove` | +| `ping_sweep` | ✅ ³ | same endpoints, one batch of parallel jobs at a time | > ¹ Requires the `os-lldpd` plugin. Returns empty dict if the plugin is not installed. > ² Requires the `os-frr` (FRR/Quagga) plugin. Returns empty dict if the plugin is not installed or FRR is not running. +> ³ Overrides the generic per-host loop from `napalm-device-types`. + +### Ping + +OPNsense has no synchronous ping endpoint: `/api/diagnostics/ping` is a *job* +API — create, start, read statistics, stop, remove. A single ping therefore +costs five requests and about a second of waiting, which makes the generic +sequential sweep from `napalm-device-types` unusable for a whole subnet. + +`ping_sweep` exploits what the job API does offer instead: jobs are +independent and run on the firewall in parallel, and `search_jobs` reports all +of them in one response. It creates and starts a batch +(`PING_SWEEP_BATCH_SIZE`, default 32), waits once, reads every result with a +single request, then cleans the batch up — waiting time per batch is constant +rather than linear in hosts. `PING_SWEEP_MAX_TARGETS` (default 512) bounds the +sweep as a whole; both are deliberately conservative, since this runs on +production firewalls. + +Results are polled rather than read once: `search_jobs` signals the running +ping with `SIGINFO` and parses whatever it has written so far, so the +statistics line lands in the log slightly after the request that triggered it. + +`ttl` and `vrf` are accepted for NAPALM compatibility and ignored — the API +has no equivalent. ## Config management diff --git a/napalm_opnsense/opnsense.py b/napalm_opnsense/opnsense.py index 6350f1a..d24077f 100644 --- a/napalm_opnsense/opnsense.py +++ b/napalm_opnsense/opnsense.py @@ -50,8 +50,10 @@ from requests.exceptions import RequestException from napalm_device_types import FingerprintRule, FirewallDriver from napalm.base.exceptions import ConnectionException, ConnectionClosedException, MergeConfigException +from napalm_opnsense.ping_mixin import OPNsensePingMixin -class OPNsenseDriver(FirewallDriver): + +class OPNsenseDriver(OPNsensePingMixin, FirewallDriver): """NAPALM driver for OPNsense (read-only, REST API).""" VENDOR = "OPNsense" diff --git a/napalm_opnsense/ping_mixin.py b/napalm_opnsense/ping_mixin.py new file mode 100644 index 0000000..1c411fd --- /dev/null +++ b/napalm_opnsense/ping_mixin.py @@ -0,0 +1,356 @@ +# -*- coding: utf-8 -*- +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Ping support for OPNsense — NAPALM ``ping()`` plus a batched ``ping_sweep()``. + +OPNsense has no synchronous "ping once and tell me the result" endpoint. +``/api/diagnostics/ping`` is a *job* API: a job is created (``set``), started +(``start``), then runs in the background writing statistics that are read back +via ``search_jobs``, and finally has to be stopped and removed again. A single +ping therefore costs five API calls and roughly one second of waiting. + +That makes the generic sequential sweep in ``napalm-device-types`` far too slow +here, but it also hands us something better: jobs are independent and run on +the firewall in parallel, and ``search_jobs`` reports *all* of them in one +response. :meth:`OPNsensePingMixin.ping_sweep` therefore works in batches — +create and start a whole batch, wait once, read every result with a single +request, then clean the batch up. Cost per batch is constant in waiting time +instead of linear in hosts. + +``PING_SWEEP_BATCH_SIZE`` bounds how many ping processes the firewall is asked +to run at once; ``PING_SWEEP_MAX_TARGETS`` bounds the sweep as a whole. Both +are deliberately conservative — this runs on production firewalls. + +Why the results are *polled* rather than read once: ``search_jobs`` signals the +running ping with ``SIGINFO`` and then parses whatever the process has written +to its log so far. The statistics line therefore lands in the log slightly +after the request that triggered it, so the first read of a healthy host can +still show zero probes. Every job is asked repeatedly until it reports the +requested probe count or the time budget runs out. + +Not supported by the API and therefore ignored: ``ttl``, ``vrf``. +""" + +from __future__ import annotations + +import logging +import time +from ipaddress import ip_address +from typing import Any, Callable, Iterable, Optional + +logger = logging.getLogger(__name__) + +_PING_API = "/api/diagnostics/ping" + +#: Seconds between two ``search_jobs`` polls while waiting for probe results. +_POLL_INTERVAL = 0.5 + +#: The API's ``interval`` field is in whole seconds and has a minimum of 1, +#: so one probe takes at least this long. +_PROBE_INTERVAL_SECONDS = 1 + +#: Extra grace on top of the expected probe duration before giving up on a job. +_POLL_GRACE_SECONDS = 1.0 + + +class OPNsensePingMixin: + """NAPALM ``ping()`` / ``ping_sweep()`` on the OPNsense diagnostics ping API.""" + + #: Ping jobs started on the firewall at the same time. + PING_SWEEP_BATCH_SIZE: int = 32 + + #: Hosts probed in a single sweep. A /24 fits; a /16 must be split by + #: the caller — 65k background jobs is not something to point at a firewall. + PING_SWEEP_MAX_TARGETS: int = 512 + + # Provided by the driver this mixin is mixed into. + _ping_model_node_cache: Optional[str] = None + + # ── NAPALM ─────────────────────────────────────────────────────────────── + + def ping( + self, + destination: str, + source: str = "", + ttl: int = 255, + timeout: int = 2, + size: int = 100, + count: int = 5, + vrf: str = "", + ) -> dict[str, Any]: + """Ping *destination* from the firewall. + + Runs the full job lifecycle and returns NAPALM's standard ping shape: + ``{"success": {...}}`` with probe counts and RTTs, or ``{"error": ...}`` + if the job could not be created or its statistics could not be read. + """ + try: + job_id = self._ping_job_create(destination, source=source, size=size) + except Exception as exc: # noqa: BLE001 - reported to the caller as-is + return {"error": str(exc)} + + try: + self._post(f"{_PING_API}/start/{job_id}", {}) + rows = self._await_ping_rows([job_id], count=count, timeout=timeout) + except Exception as exc: # noqa: BLE001 + return {"error": str(exc)} + finally: + self._ping_job_cleanup([job_id]) + + return self._row_to_napalm_reply(destination, rows.get(job_id), count=count) + + def ping_sweep( + self, + destinations: Iterable[str], + *, + count: int = 1, + timeout: int = 1, + max_targets: Optional[int] = None, + on_progress: Optional[Callable[[int, int], None]] = None, + should_stop: Optional[Callable[[], bool]] = None, + ) -> dict[str, Any]: + """Ping many destinations, one batch of parallel firewall jobs at a time. + + Same contract as + :meth:`napalm_device_types.ping_sweep.PingSweepMixin.ping_sweep`; + ``should_stop`` is polled between batches rather than between hosts, + because a batch is the smallest unit of work here. + """ + limit = self.PING_SWEEP_MAX_TARGETS if max_targets is None else max_targets + targets = list(destinations) + truncated = len(targets) > limit + if truncated: + targets = targets[:limit] + + total = len(targets) + entries: list[dict[str, Any]] = [] + for start in range(0, total, self.PING_SWEEP_BATCH_SIZE): + if should_stop is not None and should_stop(): + break + batch = targets[start : start + self.PING_SWEEP_BATCH_SIZE] + entries.extend(self._sweep_batch(batch, count=count, timeout=timeout)) + if on_progress is not None: + on_progress(len(entries), total) + + return { + "entries": entries, + "scanned": len(entries), + "alive_count": sum(1 for entry in entries if entry["alive"]), + "truncated": truncated, + } + + # ── Batch execution ────────────────────────────────────────────────────── + + def _sweep_batch( + self, destinations: list[str], *, count: int, timeout: int + ) -> list[dict[str, Any]]: + """Create, start, harvest and remove one batch of ping jobs.""" + job_ids: dict[str, str] = {} # destination -> job id + errors: dict[str, str] = {} # destination -> creation error + + for destination in destinations: + try: + job_ids[destination] = self._ping_job_create(destination) + except Exception as exc: # noqa: BLE001 - one bad host must not kill the batch + errors[destination] = str(exc) + + rows: dict[str, dict[str, Any]] = {} + batch_error: str | None = None + try: + for job_id in job_ids.values(): + self._post(f"{_PING_API}/start/{job_id}", {}) + rows = self._await_ping_rows(list(job_ids.values()), count=count, timeout=timeout) + except Exception as exc: # noqa: BLE001 - the whole batch loses its results + batch_error = str(exc) + finally: + self._ping_job_cleanup(list(job_ids.values())) + + entries: list[dict[str, Any]] = [] + for destination in destinations: + if destination in errors: + entries.append(self._parse_ping_reply(destination, {"error": errors[destination]})) + elif batch_error is not None: + entries.append(self._parse_ping_reply(destination, {"error": batch_error})) + else: + reply = self._row_to_napalm_reply( + destination, rows.get(job_ids[destination]), count=count + ) + entries.append(self._parse_ping_reply(destination, reply)) + return entries + + # ── Job lifecycle ──────────────────────────────────────────────────────── + + def _ping_job_create(self, destination: str, *, source: str = "", size: int = 100) -> str: + """Create a ping job for *destination* and return its job id.""" + settings = { + "hostname": destination, + "fam": self._address_family(destination), + "packetsize": str(size), + "interval": str(_PROBE_INTERVAL_SECONDS), + "description": "netork ping sweep", + } + if source: + settings["source_address"] = source + + response = self._post(f"{_PING_API}/set", {self._ping_model_node(): settings}) + job_id = response.get("uuid") or response.get("id") + if not job_id: + raise RuntimeError(f"ping job creation failed for {destination}: {response}") + return str(job_id) + + def _ping_job_cleanup(self, job_ids: list[str]) -> None: + """Stop and remove *job_ids*, never raising — cleanup is best effort. + + Every job left behind keeps a ping process and a log file in ``/tmp/ping`` + on the firewall, so this runs even when the sweep itself failed. + """ + for job_id in job_ids: + for action in ("stop", "remove"): + try: + self._post(f"{_PING_API}/{action}/{job_id}", {}) + except Exception as exc: # noqa: BLE001 + logger.debug("ping job %s: %s failed: %s", job_id, action, exc) + + def _await_ping_rows( + self, job_ids: list[str], *, count: int, timeout: int + ) -> dict[str, dict[str, Any]]: + """Poll ``search_jobs`` until every job sent *count* probes, or time is up. + + Returns the last rows seen, keyed by job id — a job that never reported + is simply absent, which the caller reads as "no answer". + """ + if not job_ids: + return {} + + wanted = set(job_ids) + budget = max(timeout, 1) * max(count, 1) + _POLL_GRACE_SECONDS + attempts = max(1, int(budget / _POLL_INTERVAL)) + rows: dict[str, dict[str, Any]] = {} + + for attempt in range(attempts): + if attempt: + time.sleep(_POLL_INTERVAL) + rows = { + job_id: row + for job_id, row in self._read_ping_jobs().items() + if job_id in wanted + } + if wanted and all( + _as_int(rows.get(job_id, {}).get("send")) >= count for job_id in wanted + ): + break + return rows + + def _read_ping_jobs(self) -> dict[str, dict[str, Any]]: + """All ping jobs currently known to the firewall, keyed by job id.""" + response = self._get(f"{_PING_API}/search_jobs") + rows = response.get("rows") or [] + result: dict[str, dict[str, Any]] = {} + for row in rows: + job_id = row.get("uuid") or row.get("id") + if job_id: + result[str(job_id)] = row + return result + + def _ping_model_node(self) -> str: + """Name of the model's root node, i.e. the key ``set`` expects. + + Read once from ``GET /api/diagnostics/ping/get`` instead of hardcoded, + so a model rename in a future OPNsense release does not silently break + job creation. Falls back to ``"settings"``. + """ + if self._ping_model_node_cache is not None: + return self._ping_model_node_cache + + node = "settings" + try: + response = self._get(f"{_PING_API}/get") + candidates = [key for key, value in response.items() if isinstance(value, dict)] + if len(candidates) == 1: + node = candidates[0] + except Exception as exc: # noqa: BLE001 - the default is a safe guess + logger.debug("Could not read ping model node, assuming %r: %s", node, exc) + + self._ping_model_node_cache = node + return node + + # ── Parsing ────────────────────────────────────────────────────────────── + + @staticmethod + def _address_family(destination: str) -> str: + """``"ip"`` for IPv4 / hostnames, ``"ip6"`` for IPv6 literals.""" + try: + return "ip6" if ip_address(destination).version == 6 else "ip" + except ValueError: + return "ip" + + @staticmethod + def _row_to_napalm_reply( + destination: str, row: Optional[dict[str, Any]], *, count: int + ) -> dict[str, Any]: + """Turn one ``search_jobs`` row into NAPALM's ping reply shape. + + A missing row, or one that never sent a probe, is reported as "all + probes lost" rather than as an error: from the caller's point of view + an unreachable host and a host the firewall never got around to + pinging look the same, and the sweep only asks "did it answer?". + """ + if row is None: + return _all_lost(count) + + sent = _as_int(row.get("send")) + received = _as_int(row.get("received")) + if sent == 0: + return _all_lost(count) + + avg = _as_float(row.get("avg")) + return { + "success": { + "probes_sent": sent, + "packet_loss": max(sent - received, 0), + "rtt_min": _as_float(row.get("min")), + "rtt_avg": avg, + "rtt_max": _as_float(row.get("max")), + "rtt_stddev": _as_float(row.get("std-dev")), + "results": [{"ip_address": destination, "rtt": avg}] if received else [], + } + } + + +def _all_lost(count: int) -> dict[str, Any]: + probes = max(count, 1) + 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": [], + } + } + + +def _as_int(value: Any, default: int = 0) -> int: + try: + return int(value) + except (TypeError, ValueError): + return default + + +def _as_float(value: Any, default: float = 0.0) -> float: + try: + return float(value) + except (TypeError, ValueError): + return default diff --git a/tests/unit/test_ping.py b/tests/unit/test_ping.py new file mode 100644 index 0000000..5e6e6e1 --- /dev/null +++ b/tests/unit/test_ping.py @@ -0,0 +1,325 @@ +"""Unit tests for OPNsense ping / ping_sweep — no real device required. + +The OPNsense ping API is job-based: a job is created (``set``), started +(``start``), produces statistics that are read back via ``search_jobs``, and +has to be stopped and removed again. These tests drive that whole lifecycle +against a fake ``_get`` / ``_post``. +""" + +import pytest +from unittest.mock import MagicMock, patch + +from napalm_opnsense.opnsense import OPNsenseDriver + + +# --------------------------------------------------------------------------- +# Fake API +# --------------------------------------------------------------------------- + + +class FakePingAPI: + """Minimal stand-in for the OPNsense diagnostics ping controller.""" + + def __init__(self, stats=None, model_node="settings"): + #: hostname -> row returned by search_jobs + self.stats = stats or {} + self.model_node = model_node + self.created = [] # payloads passed to set + self.started = [] # job ids passed to start + self.stopped = [] # job ids passed to stop + self.removed = [] # job ids passed to remove + self.jobs = {} # job id -> hostname + self.search_calls = 0 + self._next_id = 0 + + def get(self, path): + if path == "/api/diagnostics/ping/get": + return {self.model_node: {"hostname": "", "fam": "ip"}} + if path == "/api/diagnostics/ping/search_jobs": + self.search_calls += 1 + return {"rows": [self._row(jid, host) for jid, host in self.jobs.items()]} + raise AssertionError(f"unexpected GET {path}") + + def post(self, path, data=None): + if path == "/api/diagnostics/ping/set": + self.created.append(data) + self._next_id += 1 + job_id = f"job-{self._next_id}" + settings = (data or {}).get(self.model_node, {}) + self.jobs[job_id] = settings.get("hostname", "") + return {"result": "saved", "uuid": job_id} + for action, sink in (("start", self.started), ("stop", self.stopped)): + if path.startswith(f"/api/diagnostics/ping/{action}/"): + sink.append(path.rsplit("/", 1)[1]) + return {"status": "ok"} + if path.startswith("/api/diagnostics/ping/remove/"): + job_id = path.rsplit("/", 1)[1] + self.removed.append(job_id) + self.jobs.pop(job_id, None) + return {"status": "ok"} + raise AssertionError(f"unexpected POST {path}") + + def _row(self, job_id, hostname): + row = {"id": job_id, "hostname": hostname, "status": "running"} + row.update(self.stats.get(hostname, {"send": 0, "received": 0})) + return row + + +def alive_row(rtt=1.5, send=1, received=1): + return { + "send": send, + "received": received, + "loss": "0.0 %", + "min": rtt, + "avg": rtt, + "max": rtt, + "std-dev": 0.0, + "last_error": None, + } + + +def dead_row(send=1): + return { + "send": send, + "received": 0, + "loss": "100.0 %", + "min": 0.0, + "avg": 0.0, + "max": 0.0, + "std-dev": 0.0, + "last_error": None, + } + + +@pytest.fixture +def api(): + return FakePingAPI() + + +@pytest.fixture +def driver(api): + """Driver whose HTTP layer is replaced by the fake ping API.""" + with patch("napalm_opnsense.opnsense.requests.Session"): + drv = OPNsenseDriver( + hostname="fw.example.com", + username="api_key", + password="api_secret", + optional_args={"verify": False}, + ) + drv.session = MagicMock() + drv._get = api.get + drv._post = api.post + with patch("napalm_opnsense.ping_mixin.time.sleep"): + yield drv + + +# --------------------------------------------------------------------------- +# ping() +# --------------------------------------------------------------------------- + + +def test_ping_reachable_host_returns_napalm_success(driver, api): + api.stats["10.0.0.1"] = alive_row(rtt=2.5, send=2, received=2) + + result = driver.ping("10.0.0.1", count=2) + + assert result["success"]["probes_sent"] == 2 + assert result["success"]["packet_loss"] == 0 + assert result["success"]["rtt_avg"] == 2.5 + assert result["success"]["results"] == [{"ip_address": "10.0.0.1", "rtt": 2.5}] + + +def test_ping_unreachable_host_reports_full_packet_loss(driver, api): + api.stats["10.0.0.9"] = dead_row(send=2) + + result = driver.ping("10.0.0.9", count=2) + + assert result["success"]["probes_sent"] == 2 + assert result["success"]["packet_loss"] == 2 + assert result["success"]["results"] == [] + + +def test_ping_creates_job_with_destination_and_packetsize(driver, api): + api.stats["10.0.0.1"] = alive_row() + + driver.ping("10.0.0.1", size=64, source="10.0.0.254") + + assert api.created == [ + { + "settings": { + "hostname": "10.0.0.1", + "fam": "ip", + "packetsize": "64", + "interval": "1", + "source_address": "10.0.0.254", + "description": "netork ping sweep", + } + } + ] + + +def test_ping_uses_ipv6_family_for_v6_destination(driver, api): + api.stats["2001:db8::1"] = alive_row() + + driver.ping("2001:db8::1") + + assert api.created[0]["settings"]["fam"] == "ip6" + + +def test_ping_starts_stops_and_removes_the_job(driver, api): + api.stats["10.0.0.1"] = alive_row() + + driver.ping("10.0.0.1") + + assert api.started == ["job-1"] + assert api.stopped == ["job-1"] + assert api.removed == ["job-1"] + + +def test_ping_removes_job_even_when_reading_stats_fails(driver, api): + def _boom(path): + if path == "/api/diagnostics/ping/search_jobs": + raise RuntimeError("API down") + return api.get(path) + + driver._get = _boom + + result = driver.ping("10.0.0.1") + + assert "error" in result + assert api.removed == ["job-1"] + + +def test_ping_returns_error_when_job_creation_fails(driver, api): + api.post = lambda path, data=None: {"result": "failed", "validations": {"hostname": "bad"}} + driver._post = api.post + + result = driver.ping("nope") + + assert "error" in result + assert "hostname" in result["error"] + + +def test_ping_honours_the_model_root_node_reported_by_the_api(): + fake = FakePingAPI(model_node="ping") + fake.stats["10.0.0.1"] = alive_row() + with patch("napalm_opnsense.opnsense.requests.Session"): + drv = OPNsenseDriver(hostname="fw", username="k", password="s") + drv.session = MagicMock() + drv._get = fake.get + drv._post = fake.post + + with patch("napalm_opnsense.ping_mixin.time.sleep"): + drv.ping("10.0.0.1") + + assert "ping" in fake.created[0] + + +# --------------------------------------------------------------------------- +# ping_sweep() +# --------------------------------------------------------------------------- + + +def test_ping_sweep_probes_all_destinations_in_parallel_batches(driver, api): + api.stats["10.0.0.1"] = alive_row(rtt=1.0) + api.stats["10.0.0.3"] = alive_row(rtt=3.0) + + result = driver.ping_sweep(["10.0.0.1", "10.0.0.2", "10.0.0.3"]) + + assert result["scanned"] == 3 + assert result["alive_count"] == 2 + assert result["entries"] == [ + {"ip": "10.0.0.1", "alive": True, "rtt_ms": 1.0}, + {"ip": "10.0.0.2", "alive": False, "rtt_ms": None}, + {"ip": "10.0.0.3", "alive": True, "rtt_ms": 3.0}, + ] + + +def test_ping_sweep_starts_one_job_per_destination_and_cleans_up(driver, api): + result = driver.ping_sweep(["10.0.0.1", "10.0.0.2"]) + + assert len(api.created) == 2 + assert sorted(api.started) == ["job-1", "job-2"] + assert sorted(api.removed) == ["job-1", "job-2"] + assert result["scanned"] == 2 + + +def test_ping_sweep_reads_stats_once_per_batch_not_once_per_host(driver, api): + driver.PING_SWEEP_BATCH_SIZE = 8 + api.stats.update({f"10.0.0.{i}": alive_row() for i in range(1, 5)}) + + driver.ping_sweep([f"10.0.0.{i}" for i in range(1, 5)]) + + assert api.search_calls == 1 + + +def test_ping_sweep_splits_into_batches(driver, api): + driver.PING_SWEEP_BATCH_SIZE = 2 + api.stats.update({f"10.0.0.{i}": alive_row() for i in range(1, 6)}) + + result = driver.ping_sweep([f"10.0.0.{i}" for i in range(1, 6)]) + + assert api.search_calls == 3 # 2 + 2 + 1 + assert result["scanned"] == 5 + + +def test_ping_sweep_truncates_at_max_targets(driver, api): + result = driver.ping_sweep([f"10.0.0.{i}" for i in range(1, 11)], max_targets=3) + + assert result["scanned"] == 3 + assert result["truncated"] is True + assert len(api.created) == 3 + + +def test_ping_sweep_stops_between_batches_when_asked(driver, api): + driver.PING_SWEEP_BATCH_SIZE = 2 + calls = {"n": 0} + + def _should_stop(): + calls["n"] += 1 + return calls["n"] > 1 + + result = driver.ping_sweep( + [f"10.0.0.{i}" for i in range(1, 6)], + should_stop=_should_stop, + ) + + assert result["scanned"] == 2 + + +def test_ping_sweep_reports_progress_per_batch(driver, api): + driver.PING_SWEEP_BATCH_SIZE = 2 + seen = [] + + driver.ping_sweep( + [f"10.0.0.{i}" for i in range(1, 6)], + on_progress=lambda done, total: seen.append((done, total)), + ) + + assert seen == [(2, 5), (4, 5), (5, 5)] + + +def test_ping_sweep_marks_hosts_as_errored_when_a_batch_fails(driver, api): + def _boom(path): + if path == "/api/diagnostics/ping/search_jobs": + raise RuntimeError("API down") + return api.get(path) + + driver._get = _boom + + result = driver.ping_sweep(["10.0.0.1", "10.0.0.2"]) + + assert result["alive_count"] == 0 + assert all(entry["error"] == "API down" for entry in result["entries"]) + assert sorted(api.removed) == ["job-1", "job-2"] + + +def test_ping_sweep_on_empty_list_touches_no_api(driver, api): + result = driver.ping_sweep([]) + + assert result == {"entries": [], "scanned": 0, "alive_count": 0, "truncated": False} + assert api.created == [] + + +def test_opnsense_driver_reports_ping_support(): + assert OPNsenseDriver.supports_ping() is True