diff --git a/backend/app/services/fping_runner.py b/backend/app/services/fping_runner.py new file mode 100644 index 0000000..19d73c7 --- /dev/null +++ b/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, + ) diff --git a/backend/app/services/pinger.py b/backend/app/services/pinger.py index 4f8d2f0..6c5a80d 100644 --- a/backend/app/services/pinger.py +++ b/backend/app/services/pinger.py @@ -1,242 +1,265 @@ -""" -异步 Ping 引擎 +"""Persist batched probe summaries and apply device health transitions.""" -核心逻辑： -1. 每轮从数据库加载所有启用设备，批量 fping -2. 记录每台设备本轮 ping 结果（存活/响应时间） -3. 状态机管理设备状态，判断是否从 online→offline 或 offline→online -4. 结果通过回调或队列通知 alerter -""" - -import asyncio -import subprocess -import time import logging -from datetime import datetime -from typing import Optional, Callable, Awaitable +import time 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.models.device import Device, DeviceTypeEnum +from app.config import Settings, settings +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: - """单台设备一轮 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 - - +@dataclass(frozen=True) class DeviceStateChange: - """设备状态变化事件""" - def __init__(self, device: Device, old_status: str, new_status: str, consecutive_failures: int): - self.device = device - self.old_status = old_status - self.new_status = new_status - self.consecutive_failures = consecutive_failures + """A persisted device health transition produced by one probe round.""" + + device: Device + old_status: str + new_status: str + consecutive_failures: int + event_type: str | None = None + reason: str = "" class Pinger: - """ - Ping 引擎，使用 fping 批量并发检测。 - 对所有设备进 ping，返回存活状态和响应时间。 - """ + """Run multi-packet fping probes and update monitored device health.""" - 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]: - """ - 执行一轮 ping 检测： - 1. 加载所有启用设备 - 2. 批量 fping - 3. 记录结果 - 4. 更新设备状态 - """ + async def run_one_round( + self, + db: AsyncSession, + ) -> list[DeviceStateChange]: + """Persist one summary per enabled device and commit state changes once.""" self._round_num += 1 round_num = self._round_num - - # 1. 加载启用设备 - result = await db.execute( - select(Device).where(Device.is_enabled == True) + query_result = await db.execute( + select(Device).where(Device.is_enabled.is_(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)} 台设备") - - # 2. 批量 ping - start_time = time.time() - ip_to_device = {d.ip: d for d in devices} - 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, + started_at = time.monotonic() + policies = { + device.id: DeviceMonitoringPolicy.from_device( + device, + self._settings, ) - for r in results - ] - db.add_all(records) + for device in devices + } + 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. 更新设备状态（状态机） - device_map = {d.id: d for d in devices} - for r in results: - dev = device_map.get(r.device_id) - if not dev: + changes: list[DeviceStateChange] = [] + for device in devices: + record = records_by_device[device.id] + device.last_ping_time = observed_at + if not record.is_valid: continue - old_status = dev.current_status - if r.is_alive: - dev.consecutive_failures = 0 - dev.last_ping_time = now - dev.last_online_time = now - dev.current_status = "online" - else: - dev.consecutive_failures = (dev.consecutive_failures or 0) + 1 - dev.last_ping_time = now - if dev.consecutive_failures >= dev.alert_threshold: - if dev.current_status != "offline": - dev.current_status = "offline" - dev.last_offline_time = now - else: - if dev.current_status == "online": - dev.current_status = "checking" - - # 状态变化回调 - if old_status != dev.current_status and self._on_state_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) + policy = policies[device.id] + recent = await self._load_recent_valid( + db, + device.id, + policy, + ) + decision = evaluate_health( + device.current_status or "unknown", + recent, + policy, + ) + self._update_observation_fields(device, recent, observed_at) + change = self._apply_decision( + device, + decision, + observed_at, + ) + if change is not None: + changes.append(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)} 在线, " - f"耗时 {elapsed:.2f}s" + "[Round %s] persisted %s device summaries with %s transitions " + "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]]]: - """ - 使用 fping 批量 ping - 返回: { ip: (is_alive, response_time_ms) } - """ - if not ip_list: - return {} + @staticmethod + def _to_record( + device: Device, + probe: ProbeResult, + round_num: int, + observed_at: datetime, + ) -> 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: - # fping 一次性 ping 多个 IP - # -c 1: 每个 IP 发 1 个包 - # -t: 超时毫秒 - timeout_ms = int(settings.PING_TIMEOUT_SECONDS * 1000) - cmd = [ - settings.FPING_PATH, - "-c", "1", - "-t", str(timeout_ms), - "-e", # 显示响应时间 - ] + ip_list + @staticmethod + async def _load_recent_valid( + db: AsyncSession, + device_id: int, + policy: DeviceMonitoringPolicy, + ) -> list[PingRecord]: + """Load enough valid summaries for every configured transition rule.""" + history_size = max( + policy.offline_consecutive_rounds, + policy.degraded_window_rounds, + 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( - *cmd, - stdout=subprocess.PIPE, - stderr=subprocess.PIPE, - ) - stdout, stderr = await proc.communicate() + @staticmethod + def _update_observation_fields( + device: Device, + recent: list[PingRecord], + observed_at: datetime, + ) -> 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" - # 或 "IP : xmt/rcv/%loss = 1/0/100%" - for line in stdout.decode("utf-8", errors="replace").splitlines(): - line = line.strip() - if ":" not in line: - continue - ip = line.split(":")[0].strip() - # 解析响应时间 - if "rtt" in line: - try: - # 提取 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) + @staticmethod + def _apply_decision( + device: Device, + decision: StateDecision, + observed_at: datetime, + ) -> DeviceStateChange | None: + """Apply one pure decision and materialize its transition value.""" + old_status = device.current_status or "unknown" + if decision.next_status == old_status: + return 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: - logger.warning("fping 未找到，回退到系统 ping (串行)") - return await self._fallback_ping(ip_list) - except Exception as e: - logger.error(f"fping 异常: {e}") - return await self._fallback_ping(ip_list) - - 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} + return DeviceStateChange( + device=device, + old_status=old_status, + new_status=decision.next_status, + consecutive_failures=device.consecutive_failures or 0, + event_type=decision.event_type, + reason=decision.reason, + ) diff --git a/backend/app/services/state_machine.py b/backend/app/services/state_machine.py new file mode 100644 index 0000000..cc3bb0e --- /dev/null +++ b/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") diff --git a/backend/tests/test_fping_runner.py b/backend/tests/test_fping_runner.py new file mode 100644 index 0000000..4f26f94 --- /dev/null +++ b/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 diff --git a/backend/tests/test_state_machine.py b/backend/tests/test_state_machine.py new file mode 100644 index 0000000..3273c89 --- /dev/null +++ b/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