feat: classify offline and degraded connectivity

backend/app/services/fping_runner.py

@@ -0,0 +1,178 @@
+"""Execute and parse multi-packet fping probes without shell interpolation."""
+
+import asyncio
+from dataclasses import dataclass
+import re
+from typing import Iterable
+
+
+FPING_RESULT_PATTERN = re.compile(r"^\s*(\S+)\s+:\s*(.*?)\s*$")
+
+
+@dataclass(frozen=True)
+class ProbeResult:
+    """Aggregate result for one IP in one monitoring round."""
+
+    ip: str
+    sent_count: int
+    received_count: int
+    average_rtt_ms: float | None
+    is_valid: bool
+    failure_reason: str | None = None
+
+    @property
+    def packet_loss_percent(self) -> float:
+        """Return packet loss as a percentage of the attempted probes."""
+        if self.sent_count <= 0:
+            return 0.0
+        return (
+            (self.sent_count - self.received_count)
+            / self.sent_count
+            * 100.0
+        )
+
+
+def _invalid_result(
+    ip: str,
+    packets_per_round: int,
+    reason: str,
+) -> ProbeResult:
+    """Build an invalid aggregate that cannot resemble measured packet loss."""
+    return ProbeResult(
+        ip=ip,
+        sent_count=packets_per_round,
+        received_count=0,
+        average_rtt_ms=None,
+        is_valid=False,
+        failure_reason=reason,
+    )
+
+
+def parse_fping_count_output(
+    output: str,
+    expected_ips: set[str],
+    packets_per_round: int,
+) -> dict[str, ProbeResult]:
+    """Parse ``fping -C`` reply tokens for every expected IP.
+
+    A numeric token is a measured RTT and ``-`` is packet loss. Missing,
+    duplicated, or malformed lines are represented as invalid aggregates.
+    """
+    if packets_per_round <= 0:
+        raise ValueError("packets_per_round must be positive")
+
+    parsed_lines: dict[str, list[str]] = {}
+    duplicate_ips: set[str] = set()
+    for line in output.splitlines():
+        match = FPING_RESULT_PATTERN.match(line)
+        if match is None:
+            continue
+        ip, replies = match.groups()
+        if ip not in expected_ips:
+            continue
+        if ip in parsed_lines:
+            duplicate_ips.add(ip)
+            continue
+        parsed_lines[ip] = replies.split()
+
+    results: dict[str, ProbeResult] = {}
+    for ip in expected_ips:
+        tokens = parsed_lines.get(ip)
+        if tokens is None:
+            results[ip] = _invalid_result(
+                ip,
+                packets_per_round,
+                "missing fping output",
+            )
+            continue
+        if ip in duplicate_ips or len(tokens) != packets_per_round:
+            results[ip] = _invalid_result(
+                ip,
+                packets_per_round,
+                "malformed fping replies",
+            )
+            continue
+
+        reply_times: list[float] = []
+        malformed = False
+        for token in tokens:
+            if token == "-":
+                continue
+            try:
+                reply_time = float(token)
+            except ValueError:
+                malformed = True
+                break
+            if reply_time < 0:
+                malformed = True
+                break
+            reply_times.append(reply_time)
+
+        if malformed:
+            results[ip] = _invalid_result(
+                ip,
+                packets_per_round,
+                "malformed fping replies",
+            )
+            continue
+
+        average_rtt_ms = (
+            sum(reply_times) / len(reply_times)
+            if reply_times
+            else None
+        )
+        results[ip] = ProbeResult(
+            ip=ip,
+            sent_count=packets_per_round,
+            received_count=len(reply_times),
+            average_rtt_ms=average_rtt_ms,
+            is_valid=True,
+        )
+
+    return results
+
+
+async def run_fping_count(
+    ips: Iterable[str],
+    packets_per_round: int,
+    timeout_ms: int,
+    executable: str,
+) -> dict[str, ProbeResult]:
+    """Run one fping count probe and return an aggregate for every IP."""
+    ip_list = list(ips)
+    if not ip_list:
+        return {}
+
+    command = [
+        executable,
+        "-C",
+        str(packets_per_round),
+        "-q",
+        "-t",
+        str(timeout_ms),
+        *ip_list,
+    ]
+    try:
+        process = await asyncio.create_subprocess_exec(
+            *command,
+            stdout=asyncio.subprocess.PIPE,
+            stderr=asyncio.subprocess.PIPE,
+        )
+        stdout, stderr = await process.communicate()
+    except Exception as error:
+        reason = f"fping execution failed: {type(error).__name__}"
+        return {
+            ip: _invalid_result(ip, packets_per_round, reason)
+            for ip in ip_list
+        }
+
+    output = "\n".join(
+        stream.decode("utf-8", errors="replace")
+        for stream in (stdout, stderr)
+        if stream
+    )
+    return parse_fping_count_output(
+        output,
+        set(ip_list),
+        packets_per_round,
+    )

backend/app/services/pinger.py

@@ -1,242 +1,265 @@
-"""
+"""Persist batched probe summaries and apply device health transitions."""
-异步 Ping 引擎
 
-核心逻辑：
-1. 每轮从数据库加载所有启用设备，批量 fping
-2. 记录每台设备本轮 ping 结果（存活/响应时间）
-3. 状态机管理设备状态，判断是否从 online→offline 或 offline→online
-4. 结果通过回调或队列通知 alerter
-"""
-
-import asyncio
-import subprocess
-import time
 import logging
-from datetime import datetime
+import time
-from typing import Optional, Callable, Awaitable
 from collections import defaultdict
+from dataclasses import dataclass
+from datetime import datetime
+from typing import Awaitable, Callable
 
-from sqlalchemy import select, update
+from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
 
-from app.config import settings
+from app.config import Settings, settings
-from app.models.device import Device, DeviceTypeEnum
+from app.models.device import Device, DeviceMonitoringPolicy
 from app.models.ping_record import PingRecord
+from app.services.fping_runner import ProbeResult, run_fping_count
+from app.services.state_machine import StateDecision, evaluate_health
+
 
 logger = logging.getLogger("pingwatch.pinger")
 
 
-class PingResult:
+@dataclass(frozen=True)
-    """单台设备一轮 ping 的结果"""
-    def __init__(self, device_id: int, is_alive: bool, response_time_ms: Optional[float] = None):
-        self.device_id = device_id
-        self.is_alive = is_alive
-        self.response_time_ms = response_time_ms
-
-
 class DeviceStateChange:
-    """设备状态变化事件"""
+    """A persisted device health transition produced by one probe round."""
-    def __init__(self, device: Device, old_status: str, new_status: str, consecutive_failures: int):
+
-        self.device = device
+    device: Device
-        self.old_status = old_status
+    old_status: str
-        self.new_status = new_status
+    new_status: str
-        self.consecutive_failures = consecutive_failures
+    consecutive_failures: int
+    event_type: str | None = None
+    reason: str = ""
 
 
 class Pinger:
-    """
+    """Run multi-packet fping probes and update monitored device health."""
-    Ping 引擎，使用 fping 批量并发检测。
-    对所有设备进 ping，返回存活状态和响应时间。
-    """
 
-    def __init__(self):
+    def __init__(self, runtime_settings: Settings | None = None):
+        self._settings = runtime_settings or settings
         self._round_num = 0
-        self._on_state_change: Optional[Callable[[DeviceStateChange], Awaitable[None]]] = None
+        self._on_state_change: (
+            Callable[[DeviceStateChange], Awaitable[None]] | None
+        ) = None
 
-    def on_state_change(self, callback: Callable[[DeviceStateChange], Awaitable[None]]):
+    def on_state_change(
-        """注册状态变化回调"""
+        self,
+        callback: Callable[[DeviceStateChange], Awaitable[None]],
+    ) -> None:
+        """Register the legacy transition callback during scheduler migration."""
         self._on_state_change = callback
 
-    async def run_one_round(self, db: AsyncSession) -> list[PingResult]:
+    async def run_one_round(
-        """
+        self,
-        执行一轮 ping 检测：
+        db: AsyncSession,
-        1. 加载所有启用设备
+    ) -> list[DeviceStateChange]:
-        2. 批量 fping
+        """Persist one summary per enabled device and commit state changes once."""
-        3. 记录结果
-        4. 更新设备状态
-        """
         self._round_num += 1
         round_num = self._round_num
-
+        query_result = await db.execute(
-        # 1. 加载启用设备
+            select(Device).where(Device.is_enabled.is_(True))
-        result = await db.execute(
-            select(Device).where(Device.is_enabled == True)
         )
-        devices = list(result.scalars().all())
+        devices = list(query_result.scalars().all())
-
         if not devices:
-            logger.info(f"[Round {round_num}] 没有启用的设备")
+            logger.info("[Round %s] no enabled devices", round_num)
             return []
 
-        logger.info(f"[Round {round_num}] 开始检测 {len(devices)} 台设备")
+        started_at = time.monotonic()
-
+        policies = {
-        # 2. 批量 ping
+            device.id: DeviceMonitoringPolicy.from_device(
-        start_time = time.time()
+                device,
-        ip_to_device = {d.ip: d for d in devices}
+                self._settings,
-        ip_list = list(ip_to_device.keys())
-
-        ping_results_map = await self._batch_ping(ip_list)
-
-        # 3. 构造结果
-        results: list[PingResult] = []
-        for ip, dev in ip_to_device.items():
-            is_alive, rtt = ping_results_map.get(ip, (False, None))
-            results.append(PingResult(device_id=dev.id, is_alive=is_alive, response_time_ms=rtt))
-
-        elapsed = time.time() - start_time
-        alive_count = sum(1 for r in results if r.is_alive)
-
-        # 4. 批量写入 ping_records
-        now = datetime.now()
-        records = [
-            PingRecord(
-                device_id=r.device_id,
-                is_alive=r.is_alive,
-                response_time_ms=r.response_time_ms,
-                round_num=round_num,
-                created_at=now,
             )
-            for r in results
+            for device in devices
-        ]
+        }
-        db.add_all(records)
+        probe_results = await self._probe_by_packet_count(devices, policies)
+        observed_at = datetime.now()
+
+        records_by_device: dict[int, PingRecord] = {}
+        for device in devices:
+            policy = policies[device.id]
+            probe = probe_results.get(
+                (device.id, policy.probe_packets_per_round)
+            )
+            if probe is None:
+                probe = ProbeResult(
+                    ip=device.ip,
+                    sent_count=policy.probe_packets_per_round,
+                    received_count=0,
+                    average_rtt_ms=None,
+                    is_valid=False,
+                    failure_reason="missing runner result",
+                )
+            record = self._to_record(
+                device=device,
+                probe=probe,
+                round_num=round_num,
+                observed_at=observed_at,
+            )
+            records_by_device[device.id] = record
+            db.add(record)
+
         await db.flush()
 
-        # 5. 更新设备状态（状态机）
+        changes: list[DeviceStateChange] = []
-        device_map = {d.id: d for d in devices}
+        for device in devices:
-        for r in results:
+            record = records_by_device[device.id]
-            dev = device_map.get(r.device_id)
+            device.last_ping_time = observed_at
-            if not dev:
+            if not record.is_valid:
                 continue
 
-            old_status = dev.current_status
+            policy = policies[device.id]
-            if r.is_alive:
+            recent = await self._load_recent_valid(
-                dev.consecutive_failures = 0
+                db,
-                dev.last_ping_time = now
+                device.id,
-                dev.last_online_time = now
+                policy,
-                dev.current_status = "online"
+            )
-            else:
+            decision = evaluate_health(
-                dev.consecutive_failures = (dev.consecutive_failures or 0) + 1
+                device.current_status or "unknown",
-                dev.last_ping_time = now
+                recent,
-                if dev.consecutive_failures >= dev.alert_threshold:
+                policy,
-                    if dev.current_status != "offline":
+            )
-                        dev.current_status = "offline"
+            self._update_observation_fields(device, recent, observed_at)
-                        dev.last_offline_time = now
+            change = self._apply_decision(
-                else:
+                device,
-                    if dev.current_status == "online":
+                decision,
-                        dev.current_status = "checking"
+                observed_at,
-
+            )
-            # 状态变化回调
+            if change is not None:
-            if old_status != dev.current_status and self._on_state_change:
+                changes.append(change)
-                change = DeviceStateChange(
-                    device=dev,
-                    old_status=old_status,
-                    new_status=dev.current_status,
-                    consecutive_failures=dev.consecutive_failures,
-                )
-                await self._on_state_change(change)
 
         await db.commit()
 
+        if self._on_state_change is not None:
+            for change in changes:
+                await self._on_state_change(change)
+
+        elapsed = time.monotonic() - started_at
         logger.info(
-            f"[Round {round_num}] 完成: {alive_count}/{len(devices)} 在线, "
+            "[Round %s] persisted %s device summaries with %s transitions "
-            f"耗时 {elapsed:.2f}s"
+            "in %.2fs",
+            round_num,
+            len(devices),
+            len(changes),
+            elapsed,
         )
+        return changes
+
+    async def _probe_by_packet_count(
+        self,
+        devices: list[Device],
+        policies: dict[int, DeviceMonitoringPolicy],
+    ) -> dict[tuple[int, int], ProbeResult]:
+        """Run one argv-safe batch for each distinct packet-count policy."""
+        groups: dict[int, list[Device]] = defaultdict(list)
+        for device in devices:
+            groups[policies[device.id].probe_packets_per_round].append(device)
+
+        results: dict[tuple[int, int], ProbeResult] = {}
+        timeout_ms = int(self._settings.PING_TIMEOUT_SECONDS * 1000)
+        for packet_count, grouped_devices in groups.items():
+            unique_ips = list(dict.fromkeys(
+                device.ip for device in grouped_devices
+            ))
+            batch = await run_fping_count(
+                unique_ips,
+                packets_per_round=packet_count,
+                timeout_ms=timeout_ms,
+                executable=self._settings.FPING_PATH,
+            )
+            for device in grouped_devices:
+                probe = batch.get(device.ip)
+                if probe is not None:
+                    results[(device.id, packet_count)] = probe
         return results
 
-    async def _batch_ping(self, ip_list: list[str]) -> dict[str, tuple[bool, Optional[float]]]:
+    @staticmethod
-        """
+    def _to_record(
-        使用 fping 批量 ping
+        device: Device,
-        返回: { ip: (is_alive, response_time_ms) }
+        probe: ProbeResult,
-        """
+        round_num: int,
-        if not ip_list:
+        observed_at: datetime,
-            return {}
+    ) -> PingRecord:
+        """Convert a runner aggregate into the persisted compatibility model."""
+        return PingRecord(
+            device_id=device.id,
+            is_alive=probe.is_valid and probe.received_count > 0,
+            response_time_ms=probe.average_rtt_ms,
+            round_num=round_num,
+            created_at=observed_at,
+            sent_count=probe.sent_count,
+            received_count=probe.received_count,
+            packet_loss_percent=(
+                probe.packet_loss_percent if probe.is_valid else None
+            ),
+            average_rtt_ms=probe.average_rtt_ms,
+            is_valid=probe.is_valid,
+            failure_reason=probe.failure_reason,
+        )
 
-        try:
+    @staticmethod
-            # fping 一次性 ping 多个 IP
+    async def _load_recent_valid(
-            # -c 1: 每个 IP 发 1 个包
+        db: AsyncSession,
-            # -t: 超时毫秒
+        device_id: int,
-            timeout_ms = int(settings.PING_TIMEOUT_SECONDS * 1000)
+        policy: DeviceMonitoringPolicy,
-            cmd = [
+    ) -> list[PingRecord]:
-                settings.FPING_PATH,
+        """Load enough valid summaries for every configured transition rule."""
-                "-c", "1",
+        history_size = max(
-                "-t", str(timeout_ms),
+            policy.offline_consecutive_rounds,
-                "-e",  # 显示响应时间
+            policy.degraded_window_rounds,
-            ] + ip_list
+            policy.recovery_consecutive_clean_rounds,
+        )
+        query_result = await db.execute(
+            select(PingRecord)
+            .where(PingRecord.device_id == device_id)
+            .where(PingRecord.is_valid.is_(True))
+            .order_by(PingRecord.created_at.desc(), PingRecord.id.desc())
+            .limit(history_size)
+        )
+        newest_first = list(query_result.scalars().all())
+        return list(reversed(newest_first))
 
-            proc = await asyncio.create_subprocess_exec(
+    @staticmethod
-                *cmd,
+    def _update_observation_fields(
-                stdout=subprocess.PIPE,
+        device: Device,
-                stderr=subprocess.PIPE,
+        recent: list[PingRecord],
-            )
+        observed_at: datetime,
-            stdout, stderr = await proc.communicate()
+    ) -> None:
+        """Maintain compatibility timestamps and the full-loss streak counter."""
+        current = recent[-1]
+        if current.received_count > 0:
+            device.last_online_time = observed_at
 
-            result_map: dict[str, tuple[bool, Optional[float]]] = {}
+        full_loss_count = 0
+        for record in reversed(recent):
+            if record.sent_count and record.received_count == 0:
+                full_loss_count += 1
+            else:
+                break
+        device.consecutive_failures = full_loss_count
 
-            # fping 标准输出逐行: "IP : xmt/rcv/%loss = 1/1/0%, rtt min/avg/max = 0.12/0.12/0.12"
+    @staticmethod
-            # 或 "IP : xmt/rcv/%loss = 1/0/100%"
+    def _apply_decision(
-            for line in stdout.decode("utf-8", errors="replace").splitlines():
+        device: Device,
-                line = line.strip()
+        decision: StateDecision,
-                if ":" not in line:
+        observed_at: datetime,
-                    continue
+    ) -> DeviceStateChange | None:
-                ip = line.split(":")[0].strip()
+        """Apply one pure decision and materialize its transition value."""
-                # 解析响应时间
+        old_status = device.current_status or "unknown"
-                if "rtt" in line:
+        if decision.next_status == old_status:
-                    try:
+            return None
-                        # 提取 avg rtt
-                        rtt_part = line.split("rtt")[1]
-                        # 格式: min/avg/max = 0.12/0.12/0.12
-                        if "=" in rtt_part:
-                            avg_rtt_str = rtt_part.split("=")[1].strip().split("/")[1]
-                            rtt_ms = float(avg_rtt_str)
-                        else:
-                            rtt_ms = None
-                    except (IndexError, ValueError):
-                        rtt_ms = None
-                    result_map[ip] = (True, rtt_ms)
-                else:
-                    result_map[ip] = (False, None)
 
-            return result_map
+        device.current_status = decision.next_status
+        if decision.next_status == "offline":
+            device.last_offline_time = observed_at
+        if decision.next_status == "online":
+            device.last_online_time = observed_at
 
-        except FileNotFoundError:
+        return DeviceStateChange(
-            logger.warning("fping 未找到，回退到系统 ping (串行)")
+            device=device,
-            return await self._fallback_ping(ip_list)
+            old_status=old_status,
-        except Exception as e:
+            new_status=decision.next_status,
-            logger.error(f"fping 异常: {e}")
+            consecutive_failures=device.consecutive_failures or 0,
-            return await self._fallback_ping(ip_list)
+            event_type=decision.event_type,
-
+            reason=decision.reason,
-    async def _fallback_ping(self, ip_list: list[str]) -> dict[str, tuple[bool, Optional[float]]]:
+        )
-        """回退方案：使用系统 ping，并发执行"""
-        async def ping_one(ip: str) -> tuple[str, bool, Optional[float]]:
-            try:
-                timeout = settings.PING_TIMEOUT_SECONDS
-                cmd = ["ping", "-c", "1", "-W", str(int(timeout)), ip]
-                start = time.time()
-                proc = await asyncio.create_subprocess_exec(
-                    *cmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
-                )
-                await proc.wait()
-                elapsed = (time.time() - start) * 1000
-                return ip, proc.returncode == 0, round(elapsed, 2)
-            except Exception:
-                return ip, False, None
-
-        tasks = [ping_one(ip) for ip in ip_list]
-        sem = asyncio.Semaphore(settings.PING_CONCURRENCY)
-
-        async def bounded_ping(ip: str):
-            async with sem:
-                return await ping_one(ip)
-
-        results = await asyncio.gather(*[bounded_ping(ip) for ip in ip_list])
-        return {ip: (alive, rtt) for ip, alive, rtt in results}

backend/app/services/state_machine.py

@@ -0,0 +1,126 @@
+"""Pure state transitions for device connectivity health."""
+
+from dataclasses import dataclass
+from typing import Protocol, Sequence
+
+from app.models.device import DeviceMonitoringPolicy
+
+
+class ProbeSample(Protocol):
+    """Packet summary attributes required for health evaluation."""
+
+    sent_count: int
+    received_count: int
+    is_valid: bool
+
+
+@dataclass(frozen=True)
+class StateDecision:
+    """Result of evaluating recent probes against a monitoring policy."""
+
+    next_status: str
+    event_type: str | None
+    should_notify: bool
+    reason: str
+
+
+def _retain(previous_status: str, reason: str) -> StateDecision:
+    """Return a non-notifying decision that preserves current health."""
+    return StateDecision(
+        next_status=previous_status,
+        event_type=None,
+        should_notify=False,
+        reason=reason,
+    )
+
+
+def _transition(
+    previous_status: str,
+    next_status: str,
+    event_type: str,
+    reason: str,
+) -> StateDecision:
+    """Return a transition event only when the persisted status changes."""
+    if previous_status == next_status:
+        return _retain(previous_status, reason)
+    return StateDecision(
+        next_status=next_status,
+        event_type=event_type,
+        should_notify=True,
+        reason=reason,
+    )
+
+
+def _is_valid(sample: ProbeSample) -> bool:
+    """Reject impossible summaries before they influence device health."""
+    return (
+        sample.is_valid
+        and sample.sent_count > 0
+        and 0 <= sample.received_count <= sample.sent_count
+    )
+
+
+def evaluate_health(
+    previous_status: str,
+    recent: Sequence[ProbeSample],
+    policy: DeviceMonitoringPolicy,
+) -> StateDecision:
+    """Evaluate ordered oldest-to-newest probe summaries.
+
+    Invalid input retains state. Offline has priority over aggregate degraded
+    loss, and an alerted state recovers only after the configured clean streak.
+    """
+    if not recent:
+        return _retain(previous_status, "no probe history")
+    if any(not _is_valid(sample) for sample in recent):
+        return _retain(previous_status, "invalid probe history")
+
+    offline_rounds = policy.offline_consecutive_rounds
+    if len(recent) >= offline_rounds and all(
+        sample.received_count == 0
+        for sample in recent[-offline_rounds:]
+    ):
+        return _transition(
+            previous_status,
+            "offline",
+            "offline",
+            f"{offline_rounds} consecutive full-loss rounds",
+        )
+
+    recovery_rounds = policy.recovery_consecutive_clean_rounds
+    if (
+        previous_status in {"offline", "degraded"}
+        and len(recent) >= recovery_rounds
+        and all(
+            sample.received_count == sample.sent_count
+            for sample in recent[-recovery_rounds:]
+        )
+    ):
+        return _transition(
+            previous_status,
+            "online",
+            "recovered",
+            f"{recovery_rounds} consecutive clean rounds",
+        )
+
+    degraded_rounds = policy.degraded_window_rounds
+    if len(recent) >= degraded_rounds:
+        window = recent[-degraded_rounds:]
+        sent_count = sum(sample.sent_count for sample in window)
+        received_count = sum(sample.received_count for sample in window)
+        loss_percent = (sent_count - received_count) / sent_count * 100.0
+        if (
+            previous_status != "offline"
+            and loss_percent >= policy.degraded_loss_percent
+        ):
+            return _transition(
+                previous_status,
+                "degraded",
+                "degraded",
+                (
+                    f"window packet loss {loss_percent:.2f}% is at or above "
+                    f"{policy.degraded_loss_percent:.2f}%"
+                ),
+            )
+
+    return _retain(previous_status, "no transition threshold met")

backend/tests/test_fping_runner.py

@@ -0,0 +1,322 @@
+"""Behavior tests for multi-packet fping execution and persistence."""
+
+import asyncio
+from datetime import datetime, timedelta
+
+import pytest
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
+
+from app.config import Settings
+from app.models.device import Base, Device, DeviceTypeEnum
+from app.models.ping_record import PingRecord
+from app.services.fping_runner import (
+    ProbeResult,
+    parse_fping_count_output,
+    run_fping_count,
+)
+from app.services.pinger import Pinger
+
+
+@pytest.fixture
+async def db_session() -> AsyncSession:
+    """Provide an isolated database for pinger transaction tests."""
+    engine = create_async_engine("sqlite+aiosqlite:///:memory:")
+    try:
+        async with engine.begin() as connection:
+            await connection.run_sync(Base.metadata.create_all)
+        async with AsyncSession(engine, expire_on_commit=False) as session:
+            yield session
+    finally:
+        await engine.dispose()
+
+
+def test_parses_loss_and_average_rtt():
+    """Numeric replies count as received packets and dashes count as loss."""
+    output = "10.0.0.8 : 12.4 - 12.6\n10.0.0.9 : - - -"
+
+    result = parse_fping_count_output(
+        output,
+        {"10.0.0.8", "10.0.0.9"},
+        3,
+    )
+
+    assert result["10.0.0.8"].received_count == 2
+    assert result["10.0.0.8"].average_rtt_ms == 12.5
+    assert result["10.0.0.8"].packet_loss_percent == pytest.approx(
+        33.3333333333
+    )
+    assert result["10.0.0.9"].packet_loss_percent == 100.0
+
+
+def test_marks_missing_and_malformed_expected_output_invalid():
+    """Incomplete fping output cannot be interpreted as device packet loss."""
+    output = "10.0.0.8 : 10.2 bad-token -"
+
+    result = parse_fping_count_output(
+        output,
+        {"10.0.0.8", "10.0.0.9"},
+        3,
+    )
+
+    assert result["10.0.0.8"].is_valid is False
+    assert result["10.0.0.8"].failure_reason == "malformed fping replies"
+    assert result["10.0.0.9"].is_valid is False
+    assert result["10.0.0.9"].failure_reason == "missing fping output"
+
+
+def test_parses_ipv6_without_confusing_address_colons_for_delimiter():
+    """An IPv6 address remains the result key when parsing the output delimiter."""
+    result = parse_fping_count_output(
+        "2001:db8::8 : 1.25 - 1.75",
+        {"2001:db8::8"},
+        3,
+    )
+
+    assert result["2001:db8::8"].is_valid is True
+    assert result["2001:db8::8"].received_count == 2
+    assert result["2001:db8::8"].average_rtt_ms == 1.5
+
+
+async def test_runner_uses_argument_vector_and_parses_quiet_stderr(monkeypatch):
+    """The runner passes IPs as argv entries and reads fping -C quiet output."""
+    observed = {}
+
+    class CompletedProcess:
+        returncode = 1
+
+        async def communicate(self):
+            return b"", b"10.0.0.8 : 4.2 - 4.4\n"
+
+    async def fake_create_subprocess_exec(*args, **kwargs):
+        observed["args"] = args
+        observed["kwargs"] = kwargs
+        return CompletedProcess()
+
+    monkeypatch.setattr(
+        asyncio,
+        "create_subprocess_exec",
+        fake_create_subprocess_exec,
+    )
+
+    result = await run_fping_count(
+        ["10.0.0.8"],
+        packets_per_round=3,
+        timeout_ms=800,
+        executable="/test/fping",
+    )
+
+    assert observed["args"] == (
+        "/test/fping",
+        "-C",
+        "3",
+        "-q",
+        "-t",
+        "800",
+        "10.0.0.8",
+    )
+    assert observed["kwargs"] == {
+        "stdout": asyncio.subprocess.PIPE,
+        "stderr": asyncio.subprocess.PIPE,
+    }
+    assert result["10.0.0.8"].received_count == 2
+
+
+async def test_runner_failure_returns_invalid_result_instead_of_packet_loss(
+    monkeypatch,
+):
+    """A process communication failure remains an invalid probe aggregate."""
+
+    class BrokenProcess:
+        async def communicate(self):
+            raise RuntimeError("simulated process pipe failure")
+
+    async def fake_create_subprocess_exec(*args, **kwargs):
+        return BrokenProcess()
+
+    monkeypatch.setattr(
+        asyncio,
+        "create_subprocess_exec",
+        fake_create_subprocess_exec,
+    )
+
+    result = await run_fping_count(
+        ["10.0.0.8"],
+        packets_per_round=3,
+        timeout_ms=800,
+        executable="/test/fping",
+    )
+
+    assert result["10.0.0.8"].is_valid is False
+    assert result["10.0.0.8"].failure_reason == (
+        "fping execution failed: RuntimeError"
+    )
+
+
+async def test_pinger_persists_one_summary_for_each_enabled_device(
+    db_session,
+    monkeypatch,
+):
+    """One monitoring round writes exactly one aggregate per enabled device."""
+    first = Device(
+        name="edge-a",
+        ip="10.0.0.8",
+        device_type=DeviceTypeEnum.switch,
+        current_status="online",
+        is_enabled=True,
+    )
+    second = Device(
+        name="edge-b",
+        ip="10.0.0.9",
+        device_type=DeviceTypeEnum.switch,
+        current_status="online",
+        is_enabled=True,
+    )
+    disabled = Device(
+        name="edge-disabled",
+        ip="10.0.0.10",
+        device_type=DeviceTypeEnum.switch,
+        current_status="online",
+        is_enabled=False,
+    )
+    db_session.add_all([first, second, disabled])
+    await db_session.commit()
+
+    async def fake_run_fping_count(
+        ips,
+        packets_per_round,
+        timeout_ms,
+        executable,
+    ):
+        assert set(ips) == {"10.0.0.8", "10.0.0.9"}
+        return {
+            "10.0.0.8": ProbeResult("10.0.0.8", 3, 2, 8.5, True),
+            "10.0.0.9": ProbeResult("10.0.0.9", 3, 3, 9.5, True),
+        }
+
+    monkeypatch.setattr(
+        "app.services.pinger.run_fping_count",
+        fake_run_fping_count,
+    )
+
+    changes = await Pinger(Settings()).run_one_round(db_session)
+    records = list(
+        (
+            await db_session.execute(
+                select(PingRecord).order_by(PingRecord.device_id)
+            )
+        )
+        .scalars()
+        .all()
+    )
+
+    assert changes == []
+    assert len(records) == 2
+    assert [record.device_id for record in records] == [first.id, second.id]
+    assert records[0].sent_count == 3
+    assert records[0].received_count == 2
+    assert records[0].packet_loss_percent == pytest.approx(33.3333333333)
+    assert records[0].average_rtt_ms == 8.5
+    assert records[0].is_valid is True
+
+
+async def test_pinger_persists_invalid_probe_without_changing_device_state(
+    db_session,
+    monkeypatch,
+):
+    """A runner failure remains observable but cannot mark a device offline."""
+    device = Device(
+        name="edge-a",
+        ip="10.0.0.8",
+        device_type=DeviceTypeEnum.switch,
+        current_status="online",
+        is_enabled=True,
+    )
+    db_session.add(device)
+    await db_session.commit()
+
+    async def fake_run_fping_count(*args, **kwargs):
+        return {
+            device.ip: ProbeResult(
+                device.ip,
+                3,
+                0,
+                None,
+                False,
+                "fping executable not found",
+            )
+        }
+
+    monkeypatch.setattr(
+        "app.services.pinger.run_fping_count",
+        fake_run_fping_count,
+    )
+
+    changes = await Pinger(Settings()).run_one_round(db_session)
+    record = (
+        await db_session.execute(select(PingRecord).where(PingRecord.device_id == device.id))
+    ).scalar_one()
+
+    assert changes == []
+    assert device.current_status == "online"
+    assert record.is_valid is False
+    assert record.packet_loss_percent is None
+    assert record.failure_reason == "fping executable not found"
+
+
+async def test_pinger_evaluates_recent_valid_summaries_and_returns_change(
+    db_session,
+    monkeypatch,
+):
+    """A newly persisted full-loss summary completes the offline streak."""
+    device = Device(
+        name="edge-a",
+        ip="10.0.0.8",
+        device_type=DeviceTypeEnum.switch,
+        current_status="online",
+        is_enabled=True,
+        offline_consecutive_rounds=2,
+    )
+    db_session.add(device)
+    await db_session.flush()
+    db_session.add(
+        PingRecord(
+            device_id=device.id,
+            is_alive=False,
+            response_time_ms=None,
+            round_num=1,
+            sent_count=3,
+            received_count=0,
+            packet_loss_percent=100.0,
+            average_rtt_ms=None,
+            is_valid=True,
+            created_at=datetime.now() - timedelta(seconds=30),
+        )
+    )
+    await db_session.commit()
+
+    async def fake_run_fping_count(*args, **kwargs):
+        return {
+            device.ip: ProbeResult(
+                device.ip,
+                3,
+                0,
+                None,
+                True,
+            )
+        }
+
+    monkeypatch.setattr(
+        "app.services.pinger.run_fping_count",
+        fake_run_fping_count,
+    )
+
+    changes = await Pinger(Settings()).run_one_round(db_session)
+
+    assert len(changes) == 1
+    assert changes[0].device is device
+    assert changes[0].old_status == "online"
+    assert changes[0].new_status == "offline"
+    assert changes[0].event_type == "offline"
+    assert device.current_status == "offline"
+    assert device.last_offline_time is not None

backend/tests/test_state_machine.py

@@ -0,0 +1,139 @@
+"""Unit tests for the connectivity health state machine."""
+
+from dataclasses import dataclass
+
+import pytest
+
+from app.models.device import DeviceMonitoringPolicy
+from app.services.state_machine import evaluate_health
+
+
+@dataclass(frozen=True)
+class RecentProbe:
+    """Minimal real-data-shaped record consumed by the pure state machine."""
+
+    sent_count: int
+    received_count: int
+    is_valid: bool = True
+
+
+def loss(sent_count: int) -> RecentProbe:
+    return RecentProbe(sent_count=sent_count, received_count=0)
+
+
+def partial(sent_count: int, received_count: int) -> RecentProbe:
+    return RecentProbe(
+        sent_count=sent_count,
+        received_count=received_count,
+    )
+
+
+def clean(sent_count: int) -> RecentProbe:
+    return RecentProbe(sent_count=sent_count, received_count=sent_count)
+
+
+@pytest.fixture
+def policy() -> DeviceMonitoringPolicy:
+    return DeviceMonitoringPolicy(
+        probe_packets_per_round=3,
+        offline_consecutive_rounds=2,
+        degraded_window_rounds=5,
+        degraded_loss_percent=20.0,
+        recovery_consecutive_clean_rounds=3,
+    )
+
+
+def test_two_full_loss_rounds_take_device_offline(policy):
+    """Configured consecutive full-loss rounds transition to offline once."""
+    decision = evaluate_health("online", [loss(3), loss(3)], policy)
+
+    assert (decision.next_status, decision.event_type) == (
+        "offline",
+        "offline",
+    )
+    assert decision.should_notify is True
+
+
+def test_offline_priority_wins_over_aggregate_degraded_loss(policy):
+    """Full-loss streaks are classified offline even when the window also degrades."""
+    recent = [partial(3, 2)] * 3 + [loss(3), loss(3)]
+
+    decision = evaluate_health("online", recent, policy)
+
+    assert decision.next_status == "offline"
+    assert decision.event_type == "offline"
+
+
+def test_five_round_window_with_loss_is_degraded(policy):
+    """Aggregate loss at the configured window threshold is degraded."""
+    decision = evaluate_health(
+        "online",
+        [partial(3, 2)] * 5,
+        policy,
+    )
+
+    assert decision.next_status == "degraded"
+    assert decision.event_type == "degraded"
+
+
+def test_degraded_requires_a_complete_window(policy):
+    """A partial history cannot satisfy an aggregate-window decision."""
+    decision = evaluate_health(
+        "online",
+        [partial(3, 2)] * 4,
+        policy,
+    )
+
+    assert decision.next_status == "online"
+    assert decision.event_type is None
+
+
+def test_three_clean_rounds_recovers(policy):
+    """An alerted device recovers only after the configured clean streak."""
+    decision = evaluate_health(
+        "offline",
+        [clean(3)] * 3,
+        policy,
+    )
+
+    assert (decision.next_status, decision.event_type) == (
+        "online",
+        "recovered",
+    )
+    assert decision.should_notify is True
+
+
+def test_clean_online_device_does_not_emit_duplicate_event(policy):
+    """Clean probes retain an already-online state without notifying."""
+    decision = evaluate_health(
+        "online",
+        [clean(3)] * 5,
+        policy,
+    )
+
+    assert decision.next_status == "online"
+    assert decision.event_type is None
+    assert decision.should_notify is False
+
+
+def test_latest_invalid_probe_retains_state(policy):
+    """A malformed current result cannot reuse old loss to trigger a fault."""
+    recent = [loss(3), RecentProbe(3, 0, is_valid=False)]
+
+    decision = evaluate_health("online", recent, policy)
+
+    assert decision.next_status == "online"
+    assert decision.event_type is None
+    assert decision.should_notify is False
+
+
+def test_offline_does_not_recover_before_clean_streak_is_complete(policy):
+    """Recovery debounce retains offline until enough clean rounds exist."""
+    decision = evaluate_health(
+        "offline",
+        [clean(3), clean(3)],
+        policy,
+    )
+
+    assert decision.next_status == "offline"
+    assert decision.event_type is None