From 406fdc61ac357c1a1b6a31ecd9678abc41f28893 Mon Sep 17 00:00:00 2001 From: Jonas Bengtsson Date: Thu, 8 Oct 2026 07:26:04 +0200 Subject: [PATCH] =?UTF-8?q?=F0=9F=9A=A7=20WIP?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- CHANGELOG.md | 20 + README.md | 6 +- TODO.md | 44 +- src/opcda_diagnozer/events.py | 2 + src/opcda_diagnozer/main.py | 64 ++- src/opcda_diagnozer/monitor/detectors.py | 491 ++++++++++++++++++++++- src/opcda_diagnozer/monitor/poller.py | 291 +++++++++++++- src/opcda_diagnozer/monitor/state.py | 100 ++++- src/opcda_diagnozer/opc/fake.py | 45 +++ src/opcda_diagnozer/quality.py | 23 +- tests/test_detectors.py | 400 +++++++++++++++++- tests/test_poller_fake.py | 251 +++++++++++- tests/test_state.py | 102 ++++- 13 files changed, 1750 insertions(+), 89 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 7b4221e..5e4360f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,26 @@ bump per [SemVer](https://semver.org/). Dates are YYYY-MM-DD. ## [Unreleased] +### Added — Phase 1 core engine (the continuous monitoring loop) +- `run` verb: the foreground monitoring loop — coordinator + dedicated COM + worker thread (the client is born on the worker; COM is per-thread), + 1-second default polling of configured tags **and** device indicators, + heartbeat (D15) every `heartbeat_s`, clean Ctrl+C stop. +- Detection engine, all pure and fake-client tested: latency spike / sustained + streak / rolling p95 (D4–D6), per-tag quality transitions, persistence and + flapping (D8–D11), stale timestamp & frozen value (D12–D13), mass-quality + degradation with latch (D14), device-link indicators (D16, edge-level + DOWN/UP against configurable `ok_values`). +- Hang watchdog (spec §10.4): polls exceeding `timeout_ms` raise the D7 + soft-hang event and keep the thread; silence beyond `hard_hang_s` abandons + the stuck COM thread (never joined — dead RPC channel) and rebuilds. +- Connection state machine with exponential backoff (1 s → 60 s, reset on + success) and reconnect accounting (D1/D2 with outage duration, D3 flapping). +- Fake-client scenario engine: scripted polls and whole-call failures power + the entire poller/detector test suite without COM. +- Clear-text status rendering (`status_text`): `BAD / COMM FAILURE (0x18)` + everywhere events are shown. + ### Added — server discovery & browsing - `list-servers` verb: enumerates OPC DA servers locally or on a DCOM host via the Automation Wrapper (`GetOPCServers`), with a no-COM registry-scan fallback diff --git a/README.md b/README.md index b42aa0e..dd668e0 100644 --- a/README.md +++ b/README.md @@ -38,6 +38,10 @@ See the full interpretation matrix in the [specification](docs/SPECIFICATION.md) - Local or remote (DCOM) connections to any OPC DA server - Server browsing: `list-servers` (discovery) and `list-items` (address-space browse: full tree / branches / tags) — live now +- **Continuous monitoring** (`run`): DCOM disconnect/reconnect accounting, + latency spike/streak/trend, hang watchdog (soft flag + thread rebuild), + BAD/UNCERTAIN transitions with persistence & flapping, stale/frozen tags, + mass-degradation alarm, per-device link indicators (D16) — live now - Latency monitoring: spikes, streaks, and p95 trends against configurable thresholds - Hang watchdog: flags RPC calls that never return — the silent killer - Quality monitoring: BAD/UNCERTAIN transitions with decoded subcodes, persistent-bad @@ -76,7 +80,7 @@ opcda-diagnozer check-config # connect once and read every configured tag, print results opcda-diagnozer probe -# foreground run (console) — best for first deployment validation +# foreground run (console) — continuous monitoring; Ctrl+C stops opcda-diagnozer run # install as Windows service (dedicated account required for DCOM auth) diff --git a/TODO.md b/TODO.md index 7764f42..5cf891a 100644 --- a/TODO.md +++ b/TODO.md @@ -15,42 +15,46 @@ Legend: **[done]** complete & verified · **[partial]** some of it landed · - [done] Dependencies chosen for a 32-bit, no-compiler install: ruamel.yaml (PyYAML has no cp314 win32 wheels), pywin32. -## Phase 1 — Core engine +## Phase 1 — Core engine **[done]** - [done] 1. `config.py` — typed YAML config + strict validation; **D16 `devices:` indicators with configurable `ok_values`**; `check-config` prints them. -- [done] 2. `quality.py` — decode tables (spec §7). - - [todo] `status_text(raw) -> "NAME (0xNN)"` clear-text helper (spec §7 - closing rule; used by every reporter). +- [done] 2. `quality.py` — decode tables (spec §7) + + `status_text(raw) -> "NAME / SUBCODE (0xNN)"` clear-text helper. - [done] 3. `opc/base.py` — client/browser protocols, `ReadResult`, `OpcError` with decoded HRESULT. - [done] 4. `opc/automation.py` — connect / browse / read (per-item `Item.Read`; `SyncRead` is unmarshallable from pywin32 — verified), per-tag errors, safe COM teardown. - - [skip] Single persistent group + static items (poller will create its own + - [skip] Single persistent group + static items (poller creates its own per client; the throwaway-group read pattern is verified). -- [todo] 5. `monitor/state.py` — connection state machine, exponential - backoff (1 s → 60 s cap), reconnect accounting (only the enum exists). -- [todo] 6. `monitor/detectors.py` — D1–D15 pure detectors + rolling window - (only the `PollOutcome` dataclass exists). - - [todo] **D16 device-link detector** (`ok_values` semantics, edge-level, - `DEVICE_LINK_DOWN`/`DEVICE_LINK_UP`). -- [todo] 7. `monitor/poller.py` — the COM thread poll loop + hang watchdog - (abandon & rebuild past `hard_hang_s`). - - [todo] Startup tag validation: probe every configured tag (incl. device - indicators) before entering the loop; fail fast on invalid ItemIDs. +- [done] 5. `monitor/state.py` — connection state machine, exponential + backoff (1 s → 60 s cap, reset on success), reconnect accounting, D3 + flapping window (pure, clock-fed). +- [done] 6. `monitor/detectors.py` — D4–D16 pure detectors + rolling window: + latency spike/streak/p95 (D4–D6), quality transitions/persistence/flapping + (D8–D11), stale/frozen (D12–D13), mass-bad latch (D14), **D16 + device-link** (`ok_values` edge-level DOWN/UP). +- [done] 7. `monitor/poller.py` — coordinator + dedicated COM worker thread + (client born on the worker), D7 soft-hang flag, hard-hang watchdog + (abandon + rebuild), backoff between sessions. - [open] Poll source default (CACHE vs DEVICE) — needs a spec decision: Matrikon shows BAD qualities on some tags only via CACHE. -- [todo] 8. `run` verb — foreground monitoring loop, event pipeline to - reporters, heartbeat (D15). -- [done] 9. `opc/fake.py` — scriptable fake client (reads already scriptable - via `read_results`; hang/error sequences land with the poller). +- [done] 8. `run` verb — foreground monitoring loop, console reporter with + clear-text status, heartbeat (D15), Ctrl+C stop (second Ctrl+C aborts). +- [done] 9. `opc/fake.py` — scenario engine: scripted polls (exact results / + `"good"` shorthand) and `PollScriptError` whole-call failures; hang clients + are plain sleeping fakes in tests. + - [open] D12 stale events re-emit every poll while a tag stays stale — the + spec defines the condition per-poll; ops may want debounce/latch (decide + before the 24 h soak; live sim output showed the spam clearly). ## Phase 2 — Reporters - [todo] 1. `report/jsonl.py` — rotating JSONL (size + backups), config-digest - header, atomic-ish append. + header, atomic-ish append. **The console reporter already emits the same + events in `run` (Phase 1); Phase 2 swaps in the fan-out.** - [todo] 2. `report/textlog.py` — **plain-text anomaly log** (spec §9.1): `anomalies-YYYYMMDD.log`, one line per detected anomaly, clear-text status `NAME (0xNN)`, recoveries logged so it reads as an incident narrative. diff --git a/src/opcda_diagnozer/events.py b/src/opcda_diagnozer/events.py index 8abb1ab..84d99a9 100644 --- a/src/opcda_diagnozer/events.py +++ b/src/opcda_diagnozer/events.py @@ -45,6 +45,8 @@ class EventKind(str, Enum): FROZEN_VALUE = "FROZEN_VALUE" # D13 CONTROL_LINK_DEGRADED = "CONTROL_LINK_DEGRADED" # D14 HEARTBEAT = "HEARTBEAT" # D15 + DEVICE_LINK_DOWN = "DEVICE_LINK_DOWN" # D16 + DEVICE_LINK_UP = "DEVICE_LINK_UP" # D16 recovery @dataclass(frozen=True) diff --git a/src/opcda_diagnozer/main.py b/src/opcda_diagnozer/main.py index 33dd917..e5ab26c 100644 --- a/src/opcda_diagnozer/main.py +++ b/src/opcda_diagnozer/main.py @@ -11,6 +11,7 @@ from __future__ import annotations import argparse import sys +import threading from collections.abc import Sequence from pathlib import Path @@ -35,7 +36,6 @@ def _default_config_path() -> str: return DEFAULT_CONFIG _NOT_IMPLEMENTED: dict[str, str] = { - "run": "Phase 1 (core engine)", "install": "Phase 3 (Windows service)", "uninstall": "Phase 3 (Windows service)", "start": "Phase 3 (Windows service)", @@ -81,8 +81,12 @@ def _build_parser() -> argparse.ArgumentParser: parser.add_argument("--version", action="version", version=f"%(prog)s {__version__}") sub = parser.add_subparsers(dest="verb", metavar="verb", required=True) - p = sub.add_parser("run", help="run in the foreground (console)") + p = sub.add_parser( + "run", + help="run the monitoring loop in the foreground (Ctrl+C stops)", + ) _add_config_option(p) + _add_server_overrides(p) p = sub.add_parser( "read", @@ -550,11 +554,67 @@ def _cmd_probe( return _cmd_read(config_path, server, host, (), source, as_json=False) +def _cmd_run(config_path: str, server: str | None, host: str | None) -> int: + """Foreground monitoring loop: poll → detect → print (Ctrl+C stops).""" + import signal + + from opcda_diagnozer.monitor.poller import Poller + from opcda_diagnozer.opc.automation import AutomationOpcClient + + try: + config = load_config(config_path) + except ConfigError as exc: + print(f"run: FAILED\n {exc}", file=sys.stderr) + return 2 + # --server/--host override opc.* for this run (same as the other verbs) + if server is not None or host is not None: + from dataclasses import replace + + from opcda_diagnozer.config import OpcConfig + + opc = replace( + config.opc, + server=server or config.opc.server, + host=config.opc.host if host is None else host, + ) + config = replace(config, opc=opc) + + def client_factory() -> AutomationOpcClient: + # Built on the worker thread: CoInitialize is per-thread (spec §10.3). + return AutomationOpcClient( + server=config.opc.server, + host=config.opc.host, + connect_timeout_s=config.opc.connect_timeout_s, + ) + + def reporter(event) -> None: + # Console = the human view (spec §9): clear text, status NAME (0xNN) + line = f"{event.time_utc.strftime('%Y-%m-%d %H:%M:%S.%f')[:-3]} UTC {event.severity.value:<8} {event.kind.value}" + extras = " ".join(f"{k}={v}" for k, v in event.detail.items()) + print(f"{line} {extras}" if extras else line, flush=True) + + poller = Poller(client_factory, config, reporter) + stop_requested = threading.Event() + + def _on_ctrl_c(signum, frame) -> None: # noqa: ANN001, ARG001 + if stop_requested.is_set(): + raise KeyboardInterrupt # second Ctrl+C aborts immediately + stop_requested.set() + poller.request_stop() + print("\nrun: stop requested - finishing current poll ...", file=sys.stderr, flush=True) + + signal.signal(signal.SIGINT, _on_ctrl_c) + poller.run_forever() + return 0 + + def main(argv: Sequence[str] | None = None) -> int: """CLI entry point (console script + ``python -m opcda_diagnozer``).""" args = _build_parser().parse_args(argv) if args.verb == "check-config": return _cmd_check_config(args.config) + if args.verb == "run": + return _cmd_run(args.config, args.server, args.host) if args.verb == "list-servers": return _cmd_list_servers(args.host, args.source, args.json) if args.verb == "list-items": diff --git a/src/opcda_diagnozer/monitor/detectors.py b/src/opcda_diagnozer/monitor/detectors.py index b37da7b..4245c07 100644 --- a/src/opcda_diagnozer/monitor/detectors.py +++ b/src/opcda_diagnozer/monitor/detectors.py @@ -1,17 +1,28 @@ -"""Detector rules D1–D15 (SPECIFICATION §6). +"""Detector rules D4–D16 (SPECIFICATION §6). -Phase 1 (IMPLEMENTATION_PLAN §4.6): each detector becomes a small pure class -over :class:`PollOutcome` sequences with an injected clock; a rolling window -(deque, size ``window_polls``) powers p95/streak/flapping logic. Severity -mapping lives in SPECIFICATION §6. +Each detector is a small pure class fed one :class:`PollOutcome` per poll +cycle plus an injected clock; it returns zero or more :class:`Detection` +objects (events to report). The poller owns all timing (sleeping, RTT +measurement) — detectors only see data, so every rule is testable without COM +against scripted outcomes with a fake clock. + +Stateful detectors keep their own per-tag windows (deque bounded by +``window_polls``); none of them ever block. """ from __future__ import annotations -from dataclasses import dataclass -from datetime import datetime +from collections import deque +from dataclasses import dataclass, field +from datetime import datetime, timedelta +from opcda_diagnozer.config import Config +from opcda_diagnozer.events import EventKind, Severity from opcda_diagnozer.opc.base import ReadResult +from opcda_diagnozer.quality import QualityClass, decode_quality, status_text + +#: D14 mass degradation: >= this fraction of monitored tags BAD at once. +MASS_BAD_FRACTION = 0.5 @dataclass(frozen=True) @@ -21,4 +32,468 @@ class PollOutcome: started_at: datetime # local receive time when the read call started (wall clock) rtt_ms: float # measured round-trip time of the read call results: tuple[ReadResult, ...] = () # per-tag results; empty if the call itself failed - error: str | None = None # normalized OpcError message when the whole poll failed (D1/D2/D7) \ No newline at end of file + error: str | None = None # normalized OpcError message when the whole poll failed (D1/D2/D7) + + +@dataclass(frozen=True) +class Detection: + """One detector finding: the event kind/severity/detail the poller reports.""" + + kind: EventKind + severity: Severity + detail: dict = field(default_factory=dict) + + @property + def is_anomaly(self) -> bool: + """True for every kind the text anomaly log shows (spec §9.1). + + Heartbeats and plain CONNECTED are lifecycle noise, not anomalies. + """ + return self.kind not in (EventKind.HEARTBEAT, EventKind.CONNECTED) + + +def _tag_quality_class(result: ReadResult) -> QualityClass | None: + """Quality class of a read; None when the tag itself failed (no quality).""" + if result.error is not None: + return None + return decode_quality(result.quality_raw).quality_class + + +# --------------------------------------------------------------------------- # +# D4/D5/D6 — latency +# --------------------------------------------------------------------------- # + + +class LatencyDetector: + """D4 spike, D5 sustained-slow streak, D6 p95 trend over the window.""" + + def __init__(self, config: Config) -> None: + self._warn_ms = config.latency.warn_ms + self._critical_ms = config.latency.critical_ms + self._streak_limit = config.latency.slow_poll_streak + self._window: deque[float] = deque(maxlen=config.latency.window_polls) + self._slow_streak = 0 + + def observe(self, outcome: PollOutcome) -> list[Detection]: + if outcome.error is not None: + self._slow_streak = 0 # failed polls carry no RTT evidence + self._window.append(outcome.rtt_ms) # still measured; the hang is D7's + return [] + self._window.append(outcome.rtt_ms) + detections: list[Detection] = [] + if outcome.rtt_ms > self._warn_ms: + detections.append( + Detection( + kind=EventKind.SLOW_POLL, + severity=Severity.WARN, + detail={"rtt_ms": round(outcome.rtt_ms, 1), "warn_ms": self._warn_ms}, + ) + ) + self._slow_streak += 1 + if self._slow_streak >= self._streak_limit: + detections.append( + Detection( + kind=EventKind.SLOW_POLL_STREAK, + severity=Severity.CRITICAL, + detail={ + "rtt_ms": round(outcome.rtt_ms, 1), + "streak": self._slow_streak, + "warn_ms": self._warn_ms, + }, + ) + ) + else: + self._slow_streak = 0 + p95 = self._p95() + if p95 is not None and p95 > self._warn_ms: + detections.append( + Detection( + kind=EventKind.LATENCY_TREND, + severity=Severity.WARN, + detail={"rtt_p95_ms": round(p95, 1), "warn_ms": self._warn_ms}, + ) + ) + return detections + + def _p95(self) -> float | None: + if not self._window: + return None + ordered = sorted(self._window) + # nearest-rank p95 of the rolling window + rank = max(1, round(0.95 * len(ordered))) + return ordered[rank - 1] + + +# --------------------------------------------------------------------------- # +# D8/D9/D10/D11 — per-tag quality +# --------------------------------------------------------------------------- # + + +@dataclass +class _TagQualityState: + """Per-tag tracking for the quality detectors.""" + + last_class: QualityClass | None = None + bad_since: datetime | None = None # D10 persistence clock start + bad_reported: bool = False + persistent_reported: bool = False + transition_times: deque[datetime] = field(default_factory=lambda: deque()) + + +class QualityDetector: + """D8/D9 transitions, D10 persistence, D11 flapping — per monitored tag. + + Tag state initializes lazily from the first observed reading; transitions + are computed from the *previous class*, so a tag starting BAD does not + fire a transition until it first becomes GOOD (no synthetic baseline). + """ + + def __init__(self, config: Config) -> None: + self._persistent_bad_s = config.quality.persistent_bad_s + self._flapping_limit = config.quality.flapping_transitions_per_window + self._flapping_window = timedelta(seconds=config.quality.flapping_window_s) + self._tags: dict[str, _TagQualityState] = {} + + def observe(self, outcome: PollOutcome) -> list[Detection]: + detections: list[Detection] = [] + at = outcome.started_at + for result in outcome.results: + state = self._tags.setdefault(result.tag, _TagQualityState()) + quality = decode_quality(result.quality_raw) if result.error is None else None + new_class = quality.quality_class if quality is not None else None + self._on_transition(state, result, quality, at, detections) + self._on_persistence(state, result, at, detections) + state.last_class = new_class + return detections + + def _on_transition( + self, + state: _TagQualityState, + result: ReadResult, + quality, + at: datetime, + out: list[Detection], + ) -> None: + new_class = quality.quality_class if quality is not None else None + old = state.last_class + if old is None or new_class is None or old == new_class: + return # first reading or unreadable tag: no synthetic transition + state.transition_times.append(at) + state.transition_times = deque( + t for t in state.transition_times if at - t <= self._flapping_window + ) + if new_class is QualityClass.BAD: + out.append( + Detection( + kind=EventKind.QUALITY_BAD, + severity=Severity.WARN, + detail={ + "tag": result.tag, + "quality": status_text(result.quality_raw), + "previous_quality": old.value, + "value": result.value, + }, + ) + ) + elif new_class is QualityClass.UNCERTAIN: + out.append( + Detection( + kind=EventKind.QUALITY_UNCERTAIN, + severity=Severity.INFO, + detail={ + "tag": result.tag, + "quality": status_text(result.quality_raw), + "previous_quality": old.value, + }, + ) + ) + if len(state.transition_times) > self._flapping_limit: + out.append( + Detection( + kind=EventKind.QUALITY_FLAPPING, + severity=Severity.WARN, + detail={ + "tag": result.tag, + "transitions_in_window": len(state.transition_times), + "window_s": self._flapping_window.total_seconds(), + }, + ) + ) + + def _on_persistence( + self, + state: _TagQualityState, + result: ReadResult, + at: datetime, + out: list[Detection], + ) -> None: + bad = _tag_quality_class(result) is QualityClass.BAD + if bad: + if state.bad_since is None: + state.bad_since = at + state.bad_reported = True + been_bad_s = (at - state.bad_since).total_seconds() + if been_bad_s >= self._persistent_bad_s and not state.persistent_reported: + state.persistent_reported = True + out.append( + Detection( + kind=EventKind.QUALITY_BAD_PERSISTENT, + severity=Severity.WARN, + detail={ + "tag": result.tag, + "bad_for_s": round(been_bad_s, 1), + "threshold_s": self._persistent_bad_s, + }, + ) + ) + else: + state.bad_since = None + state.bad_reported = False + state.persistent_reported = False + + def tags_bad(self) -> int: + """Monitored tags currently BAD — feeds the heartbeat (D15) and D14.""" + # current-ness approximation: a tag counts if its last class was BAD + return sum(1 for s in self._tags.values() if s.last_class is QualityClass.BAD) + + +# --------------------------------------------------------------------------- # +# D12/D13 — staleness & frozen values +# --------------------------------------------------------------------------- # + + +class StaleDetector: + """D12 stale device timestamps; D13 frozen values (``expect_change`` tags). + + Both compare the device time against the local receive clock (the spec's + ``max_age_s`` tolerance absorbs reasonable clock skew; NTP on both hosts + is the real fix — spec §12 row 9). + """ + + def __init__(self, config: Config) -> None: + self._max_age = timedelta(seconds=config.stale.max_age_s) + self._enabled = config.stale.enabled + self._expect_change = {t.name for t in config.tags if t.expect_change} + # last (value, device_time) per tag for the frozen check + self._last_seen: dict[str, tuple[object, datetime]] = {} + + def observe(self, outcome: PollOutcome) -> list[Detection]: + if not self._enabled or outcome.error is not None: + return [] + detections: list[Detection] = [] + at = outcome.started_at + for result in outcome.results: + if result.error is not None or result.device_time is None: + continue + if _tag_quality_class(result) is not QualityClass.GOOD: + continue # staleness while quality still reports GOOD (spec D12) + age = at - result.device_time + if age > self._max_age: + detections.append( + Detection( + kind=EventKind.STALE_TIMESTAMP, + severity=Severity.WARN, + detail={ + "tag": result.tag, + "device_time_utc": result.device_time.isoformat(), + "age_s": round(age.total_seconds(), 1), + "max_age_s": self._max_age.total_seconds(), + }, + ) + ) + self._frozen_check(result, at, detections) + return detections + + def _frozen_check( + self, result: ReadResult, at: datetime, out: list[Detection] + ) -> None: + if result.tag not in self._expect_change: + return + previous = self._last_seen.get(result.tag) + self._last_seen[result.tag] = (result.value, result.device_time) + if previous is None or previous[1] is None or result.device_time is None: + return + same_value = str(previous[0]) == str(result.value) + same_time = previous[1] == result.device_time + if same_value and same_time: + gap = at - result.device_time + if gap > self._max_age: + out.append( + Detection( + kind=EventKind.FROZEN_VALUE, + severity=Severity.WARN, + detail={ + "tag": result.tag, + "value": result.value, + "device_time_utc": result.device_time.isoformat(), + "unchanged_for_s": round(gap.total_seconds(), 1), + }, + ) + ) + + +# --------------------------------------------------------------------------- # +# D14 — mass quality degradation +# --------------------------------------------------------------------------- # + + +class MassBadDetector: + """D14: >= half of monitored tags BAD simultaneously while connection OK.""" + + def __init__(self, config: Config | None = None) -> None: + # config reserved for a future threshold key; the fraction is fixed + # by the spec (MASS_BAD_FRACTION). + self._reported = False + + def observe(self, outcome: PollOutcome) -> list[Detection]: + if outcome.error is not None: + self._reported = False # reconnect resets the mass-degradation latch + return [] + total = len(outcome.results) + if total == 0: + return [] + bad = sum(1 for r in outcome.results if _tag_quality_class(r) is QualityClass.BAD) + degraded = bad >= MASS_BAD_FRACTION * total + detections: list[Detection] = [] + if degraded and not self._reported: + detections.append( + Detection( + kind=EventKind.CONTROL_LINK_DEGRADED, + severity=Severity.CRITICAL, + detail={"tags_bad": bad, "tags_total": total}, + ) + ) + self._reported = True + elif not degraded and self._reported: + self._reported = False # recovered; next degradation reports again + return detections + + +# --------------------------------------------------------------------------- # +# D16 — device-link indicators +# --------------------------------------------------------------------------- # + + +def _rendered(value: object) -> str: + """Value as the comparison string (config ok_values normalization twin).""" + if isinstance(value, bool): + return "true" if value else "false" + return str(value).strip() + + +class DeviceLinkDetector: + """D16: per-device link state from configured indicator tags (spec §6.1). + + Edge-level by design: one ``DEVICE_LINK_DOWN`` when a device leaves its + normal state (GOOD + value in ``ok_values``), one ``DEVICE_LINK_UP`` when + it returns — flapping links produce transitions, not a stream of events. + """ + + def __init__(self, config: Config) -> None: + self._devices: dict[str, dict] = { + device.tag: {"name": device.name, "ok": set(device.ok_values), "down": False} + for device in config.devices + } + + def observe(self, outcome: PollOutcome) -> list[Detection]: + if outcome.error is not None: + return [] # whole-poll failure is D1's business, not per-device + detections: list[Detection] = [] + for result in outcome.results: + entry = self._devices.get(result.tag) + if entry is None: + continue + quality = ( + decode_quality(result.quality_raw) + if result.error is None + else None + ) + down = result.error is not None + reason = None + if not down: + if quality is None or quality.quality_class is not QualityClass.GOOD: + down = True + reason = ( + "quality " + status_text(result.quality_raw) + if quality is not None + else "unreadable tag: " + (result.error or "?") + ) + else: + rendered = _rendered(result.value) + if rendered not in entry["ok"]: + down = True + reason = f"read {rendered!r}, expected one of {sorted(entry['ok'])}" + if down and not entry["down"]: + entry["down"] = True + detail = { + "device": entry["name"], + "indicator_tag": result.tag, + "value": result.value, + "quality": status_text(result.quality_raw) if result.error is None else None, + } + if reason: + detail["reason"] = reason + detections.append( + Detection(kind=EventKind.DEVICE_LINK_DOWN, severity=Severity.CRITICAL, detail=detail) + ) + elif not down and entry["down"]: + entry["down"] = False + detections.append( + Detection( + kind=EventKind.DEVICE_LINK_UP, + severity=Severity.INFO, + detail={"device": entry["name"], "indicator_tag": result.tag}, + ) + ) + return detections + + def devices_down(self) -> int: + return sum(1 for e in self._devices.values() if e["down"]) + + +# --------------------------------------------------------------------------- # +# The bundle the poller feeds every cycle +# --------------------------------------------------------------------------- # + + +class DetectorSuite: + """All per-poll detectors run in a fixed order; one call per outcome.""" + + def __init__(self, config: Config) -> None: + self.latency = LatencyDetector(config) + self.quality = QualityDetector(config) + self.stale = StaleDetector(config) + self.mass_bad = MassBadDetector(config) + self.device_link = DeviceLinkDetector(config) + + def observe(self, outcome: PollOutcome) -> list[Detection]: + detections: list[Detection] = [] + detections.extend(self.latency.observe(outcome)) + detections.extend(self.quality.observe(outcome)) + detections.extend(self.stale.observe(outcome)) + detections.extend(self.mass_bad.observe(outcome)) + detections.extend(self.device_link.observe(outcome)) + return detections + + def rtt_p95_ms(self) -> float | None: + p95 = self.latency._p95() # noqa: SLF001 — suite owns its parts + return round(p95, 1) if p95 is not None else None + + def tags_bad(self) -> int: + return self.quality.tags_bad() + + def devices_down(self) -> int: + return self.device_link.devices_down() + + +__all__ = [ + "Detection", + "DetectorSuite", + "PollOutcome", + "MASS_BAD_FRACTION", + "LatencyDetector", + "QualityDetector", + "StaleDetector", + "MassBadDetector", + "DeviceLinkDetector", +] \ No newline at end of file diff --git a/src/opcda_diagnozer/monitor/poller.py b/src/opcda_diagnozer/monitor/poller.py index f4118c9..1a15591 100644 --- a/src/opcda_diagnozer/monitor/poller.py +++ b/src/opcda_diagnozer/monitor/poller.py @@ -1,25 +1,292 @@ -"""The poll loop on the single dedicated COM thread (SPECIFICATION §10.3). +"""The poll loop on the single dedicated COM thread (SPECIFICATION §10.3/§10.4). -Phase 1 (IMPLEMENTATION_PLAN §4.7): connect → poll loop → feed detectors, plus -the hang watchdog (SPECIFICATION §10.4): a second thread timestamps poll -starts/ends; when a call exceeds ``hard_hang_s`` the stuck COM thread is -abandoned and a fresh connection is built on a new thread. +Architecture (all COM stays on the worker thread — SPECIFICATION §10.3): + +- :meth:`Poller.run_forever` runs on the *caller's* thread as the coordinator: + it starts a worker per (re)connect session, consumes its queue, feeds the + state machine, applies backoff, and owns the stop flag. +- The worker thread builds its **own** client via ``client_factory`` (COM + initialization is per-thread — the client must be born there), polls every + ``interval_s``, measures RTT, and pushes messages onto the queue. +- Hang watchdog (SPECIFICATION §10.4): the coordinator timestamps every + worker message; silence longer than ``hard_hang_s`` means the thread is + stuck in a dead RPC channel — it is **abandoned** (daemon thread, never + joined) and a fresh worker is built after D1/backoff. A poll that merely + exceeds ``timeout_ms`` raises the D7 soft-hang event but keeps the thread. + +Messages the worker sends: ``("connected", ts)``, ``("outcome", PollOutcome)``, +``("failed", reason)`` — after which the worker exits its loop. """ from __future__ import annotations +import queue +import threading +import time +from collections.abc import Callable +from datetime import datetime, timezone +from typing import Any + from opcda_diagnozer.config import Config -from opcda_diagnozer.opc.base import OpcClient +from opcda_diagnozer.events import DiagEvent, EventContext, EventKind, Severity +from opcda_diagnozer.monitor.detectors import Detection, DetectorSuite, PollOutcome +from opcda_diagnozer.monitor.state import ( + ConnectionState, + ConnectionStateMachine, + StateEvent, +) +from opcda_diagnozer.opc.base import OpcClient, OpcError + +#: Coordinator queue poll slice: small enough that stop is snappy, large +#: enough to not busy-spin (hard_hang watching happens between slices). +_WATCH_SLICE_S = 0.25 + +_MESSAGE = tuple[str, Any] + + +def _utc_now() -> datetime: + """Production wall clock (injected as ``clock`` in tests).""" + return datetime.now(timezone.utc) class Poller: - """Drives one :class:`OpcClient` on its own thread — implemented in Phase 1.""" + """Drives one :class:`OpcClient` per connection on its own thread, forever. - def __init__(self, client: OpcClient, config: Config) -> None: - self._client = client + ``client_factory`` is called **on the worker thread** for every + (re)connect — the client is thread-confined by construction (spec §10.3). + ``reporter`` receives every event; the coordinator never touches the + client object. + """ + + def __init__( + self, + client_factory: Callable[[], OpcClient], + config: Config, + reporter: Callable[[DiagEvent], None], + clock: Callable[[], datetime] = _utc_now, + stop: threading.Event | None = None, + ) -> None: + self._client_factory = client_factory self._config = config + self._reporter = reporter + self._clock = clock + self._stop = stop if stop is not None else threading.Event() + self._sm = ConnectionStateMachine() + self._suite = DetectorSuite(config) + self._started_at: datetime | None = None + # Surveillance tags + device-indicator tags (polled automatically, §6.1), + # deduplicated in order. + self._tags: tuple[str, ...] = tuple( + dict.fromkeys( + [t.name for t in config.tags] + [d.tag for d in config.devices] + ) + ) + + # ------------------------------------------------------------------ # + # public surface + # ------------------------------------------------------------------ # + + def request_stop(self) -> None: + """Ask the loop to finish after the current poll (from any thread).""" + self._stop.set() def run_forever(self) -> None: - raise NotImplementedError( - "Phase 1: poller + hang watchdog (docs/IMPLEMENTATION_PLAN.md §4.7)" - ) \ No newline at end of file + """Coordinate worker sessions until :meth:`request_stop`. + + Never raises for COM/OPC failures — they are events (D1/D2). A worker + crash surfaces as DISCONNECTED with the exception text, then reconnect. + """ + self._started_at = self._clock() + last_heartbeat = self._started_at + while not self._stop.is_set(): + outcomes: queue.Queue[_MESSAGE] = queue.Queue() + worker = threading.Thread( + target=self._worker_main, args=(outcomes,), daemon=True + ) + worker.start() + hung = self._coordinate(outcomes) + if hung: + # SPECIFICATION §10.4: abandon a thread stuck in a dead RPC + # channel — never join it; daemon=True lets the process exit. + self._emit( + EventKind.POLL_HANG, + Severity.CRITICAL, + { + "hard_hang_s": self._config.poll.hard_hang_s, + "action": "COM thread abandoned and rebuilt", + }, + ) + else: + worker.join(timeout=self._config.poll.hard_hang_s) + if self._stop.is_set(): + break + if self._sm.state is ConnectionState.RECONNECT_WAIT: + delay = self._sm.next_attempt_delay_s() + if delay: + self._stop.wait(delay) + self._emit( + EventKind.DISCONNECTED, + Severity.INFO, + {"reason": "shutting down (stop requested)"}, + ) + + # ------------------------------------------------------------------ # + # coordinator side (caller's thread) + # ------------------------------------------------------------------ # + + def _coordinate(self, outcomes: "queue.Queue[_MESSAGE]") -> bool: + """Consume one worker session; returns True when the worker was hung. + + Feeds the state machine and detectors, emits D1/D2/D7/D15/D4–D6/D8–D14/D16 + events, and watches for hard hangs by wall-clock silence. + """ + last_worker_word = self._clock() + last_heartbeat = self._started_at or self._clock() + while not self._stop.is_set(): + try: + kind, payload = outcomes.get(timeout=_WATCH_SLICE_S) + except queue.Empty: + silent_s = (self._clock() - last_worker_word).total_seconds() + if silent_s > self._config.poll.hard_hang_s: + return True + continue + last_worker_word = self._clock() + if kind == "connected": + for ev in self._sm.on_connected(payload): + self._emit_state_event(ev) + self._emit( + EventKind.CONNECTED, + Severity.INFO, + {"server": self._config.opc.server}, + ) + elif kind == "outcome": + outcome: PollOutcome = payload + for event in self._consume_outcome(outcome): + self._reporter(event) + now = outcome.started_at + if (now - last_heartbeat).total_seconds() >= self._config.report.heartbeat_s: + last_heartbeat = now + self._emit(EventKind.HEARTBEAT, Severity.INFO, self._heartbeat_detail()) + elif kind == "failed": + for ev in self._sm.on_disconnected(self._clock(), reason=str(payload)): + self._emit_state_event(ev) + return False + return False + + def _consume_outcome(self, outcome: PollOutcome) -> list[DiagEvent]: + """Run the detector suite on one outcome; add D7 soft-hang; map to events.""" + events: list[DiagEvent] = [] + for detection in self._suite.observe(outcome): + events.append(self._to_event(detection)) + if outcome.error is None and outcome.rtt_ms > self._config.poll.timeout_ms: + events.append( + self._to_event( + Detection( + kind=EventKind.POLL_HANG, + severity=Severity.CRITICAL, + detail={ + "rtt_ms": round(outcome.rtt_ms, 1), + "timeout_ms": self._config.poll.timeout_ms, + "note": "call returned late; thread kept", + }, + ) + ) + ) + return events + + def _to_event(self, detection: Detection) -> DiagEvent: + return DiagEvent( + time_utc=self._clock(), + kind=detection.kind, + severity=detection.severity, + detail=detection.detail, + context=EventContext( + uptime_s=self._uptime_s(), + reconnects_total=self._sm.reconnects_total, + rtt_p95_ms=self._suite.rtt_p95_ms(), + tags_bad=self._suite.tags_bad(), + tags_total=len(self._tags), + ), + ) + + def _emit_state_event(self, ev: StateEvent) -> None: + kind, severity = { + "disconnected": (EventKind.DISCONNECTED, Severity.CRITICAL), + "reconnected": (EventKind.RECONNECTED, Severity.INFO), + "flapping": (EventKind.RECONNECT_FLAPPING, Severity.CRITICAL), + }[ev.kind] + self._emit(kind, severity, ev.detail) + + def _heartbeat_detail(self) -> dict: + return { + "uptime_s": self._uptime_s(), + "reconnects_total": self._sm.reconnects_total, + "rtt_p95_ms": self._suite.rtt_p95_ms(), + "tags_bad": self._suite.tags_bad(), + "tags_total": len(self._tags), + "devices_down": self._suite.devices_down(), + } + + def _uptime_s(self) -> float: + if self._started_at is None: + return 0.0 + return (self._clock() - self._started_at).total_seconds() + + def _emit(self, kind: EventKind, severity: Severity, detail: dict) -> None: + self._reporter(self._to_event(Detection(kind=kind, severity=severity, detail=detail))) + + # ------------------------------------------------------------------ # + # worker side (the dedicated COM thread) + # ------------------------------------------------------------------ # + + def _worker_main(self, outcomes: "queue.Queue[_MESSAGE]") -> None: + """One connection session: own client, poll loop, push messages. + + Exits (putting ``("failed", reason)``) whenever the connection drops + or the stop flag is observed — the coordinator decides what next. + """ + client: OpcClient | None = None + try: + client = self._client_factory() + client.connect() + outcomes.put(("connected", self._clock())) + while not self._stop.is_set(): + started = time.monotonic() + started_at = _utc_now() + try: + results = client.read(self._tags) + rtt_ms = (time.monotonic() - started) * 1000.0 + outcomes.put( + ( + "outcome", + PollOutcome( + started_at=started_at, + rtt_ms=rtt_ms, + results=tuple(results), + ), + ) + ) + except OpcError as exc: + rtt_ms = (time.monotonic() - started) * 1000.0 + outcomes.put( + ( + "outcome", + PollOutcome( + started_at=started_at, rtt_ms=rtt_ms, error=str(exc) + ), + ) + ) + outcomes.put(("failed", str(exc))) + return + if self._stop.wait(self._config.poll.interval_s): + break + except OpcError as exc: + outcomes.put(("failed", str(exc))) + except Exception as exc: # noqa: BLE001 — a worker death is an event, not a crash + outcomes.put(("failed", f"worker crashed: {type(exc).__name__}: {exc}")) + finally: + if client is not None: + try: + client.close() + except Exception: # noqa: BLE001 — best-effort teardown + pass \ No newline at end of file diff --git a/src/opcda_diagnozer/monitor/state.py b/src/opcda_diagnozer/monitor/state.py index 702d1dc..2ba11bf 100644 --- a/src/opcda_diagnozer/monitor/state.py +++ b/src/opcda_diagnozer/monitor/state.py @@ -4,11 +4,14 @@ States: Connecting → Connected → ReconnectWait → Connecting … Backoff is exponential 1 s → 2 s → … capped at 60 s; every disconnect/reconnect pair is an event (D1/D2) with the outage duration computed and reported. -The transition/backoff engine lands in Phase 1 (IMPLEMENTATION_PLAN §4.5). +The engine is pure: it is fed timestamps and returns transitions plus events, +so the poller owns the actual clock/sleeping and tests drive it deterministically. """ from __future__ import annotations +from dataclasses import dataclass, field +from datetime import datetime, timedelta from enum import Enum @@ -17,4 +20,97 @@ class ConnectionState(Enum): CONNECTING = "connecting" CONNECTED = "connected" - RECONNECT_WAIT = "reconnect_wait" \ No newline at end of file + RECONNECT_WAIT = "reconnect_wait" + + +#: Backoff constants (SPECIFICATION §10.2). +BACKOFF_INITIAL_S = 1.0 +BACKOFF_MAX_S = 60.0 + +#: D3 reconnect flapping: > this many disconnects within the window is CRITICAL. +FLAPPING_DISCONNECTS = 3 +FLAPPING_WINDOW = timedelta(minutes=10) + + +@dataclass(frozen=True) +class StateEvent: + """One state-machine transition worth reporting (D1/D2/D3).""" + + kind: str # "disconnected" | "reconnected" | "flapping" + at: datetime + detail: dict = field(default_factory=dict) + + +@dataclass +class ConnectionStateMachine: + """Tracks connection state, backoff and reconnect accounting (D1–D3). + + The poller calls :meth:`on_connected` / :meth:`on_disconnected` at the + moments those things are observed; the machine decides *when* to emit + events (D1 with the failure reason, D2 with the outage duration, D3 when + disconnects flap) and *how long* to wait before the next attempt. + """ + + state: ConnectionState = ConnectionState.CONNECTING + backoff_s: float = BACKOFF_INITIAL_S + connected_since: datetime | None = None + disconnected_since: datetime | None = None + reconnects_total: int = 0 + _disconnect_times: list[datetime] = field(default_factory=list) + + def on_connected(self, at: datetime) -> list[StateEvent]: + """A connect (or reconnect) succeeded at ``at``.""" + events: list[StateEvent] = [] + was_outage = self.state is ConnectionState.RECONNECT_WAIT + self.state = ConnectionState.CONNECTED + self.connected_since = at + self.backoff_s = BACKOFF_INITIAL_S # reset on success (spec §10.2) + if was_outage and self.disconnected_since is not None: + self.reconnects_total += 1 + events.append( + StateEvent( + kind="reconnected", + at=at, + detail={ + "outage_s": (at - self.disconnected_since).total_seconds(), + "reconnects_total": self.reconnects_total, + }, + ) + ) + self.disconnected_since = None + return events + + def on_disconnected(self, at: datetime, reason: str) -> list[StateEvent]: + """The connection dropped at ``at`` (reason = decoded failure).""" + events: list[StateEvent] = [] + first_drop = self.state is not ConnectionState.RECONNECT_WAIT + self.state = ConnectionState.RECONNECT_WAIT + self.connected_since = None + if first_drop: + self.disconnected_since = at + self._disconnect_times.append(at) + self._disconnect_times = [ + t for t in self._disconnect_times if at - t <= FLAPPING_WINDOW + ] + events.append(StateEvent(kind="disconnected", at=at, detail={"reason": reason})) + if len(self._disconnect_times) > FLAPPING_DISCONNECTS: + events.append( + StateEvent( + kind="flapping", + at=at, + detail={"disconnects_in_window": len(self._disconnect_times)}, + ) + ) + return events + + def next_attempt_delay_s(self) -> float: + """How long to wait before the next connect attempt (exponential backoff). + + Call once per failed attempt; the value doubles up to + :data:`BACKOFF_MAX_S` and resets only on a successful connect. + """ + if self.state is not ConnectionState.RECONNECT_WAIT: + return 0.0 + delay = self.backoff_s + self.backoff_s = min(self.backoff_s * 2, BACKOFF_MAX_S) + return delay \ No newline at end of file diff --git a/src/opcda_diagnozer/opc/fake.py b/src/opcda_diagnozer/opc/fake.py index cd4ccf6..e442528 100644 --- a/src/opcda_diagnozer/opc/fake.py +++ b/src/opcda_diagnozer/opc/fake.py @@ -17,6 +17,19 @@ from datetime import datetime, timezone from opcda_diagnozer.opc.base import Browser, OPCDataSource, OpcError, ReadResult, TagInfo +#: Scenario-engine directives (Phase 1, IMPLEMENTATION_PLAN §4.9): the fake +#: consumes one per read() call. A "hang" is not modeled here — the poller's +#: watchdog handles hangs via a client whose read() simply sleeps; the fake +#: stays non-blocking so unit tests never slow down. +PollScriptStep = "PollScriptError | list[ReadResult] | str" + + +class PollScriptError: + """Scenario directive: the whole read call fails with this message.""" + + def __init__(self, message: str) -> None: + self.message = message + #: The fake address space: branch path -> (sub-branches, leaf names). #: Mirrors the shape of a real SCADA server (Channel → Device → Tags). FAKE_SPACE: dict[tuple[str, ...], tuple[tuple[str, ...], tuple[str, ...]]] = { @@ -101,16 +114,32 @@ class FakeOpcClient: ``space`` is injectable so tests can exercise flat namespaces, empty servers, and deep trees; ``read_results`` scripts the next read() result (Phase 1 scenario engine seed, IMPLEMENTATION_PLAN §4.9). + + Scenario engine (Phase 1): ``script`` is a list of per-poll directives the + fake consumes one per read() call, in order — after the script is + exhausted the client falls back to ``read_results`` or plain GOOD zeros. + Each entry may be: + + - ``PollOutcomeError(message)`` — the whole read call fails (COM-style + OpcError for D1/D2 testing; a ``PollOutcomeHang`` directive is handled by + the *poller* watchdog, not the client — see :class:`FakeHangClient`), + - a ``list[ReadResult]`` — the exact results for this poll, + - ``"good"`` — shorthand for all-tags GOOD zeros. + + The read log (``reads``) records every consumed (tags, source) pair so + tests can assert poll cadence and scripted order. """ def __init__( self, space: dict[tuple[str, ...], tuple[tuple[str, ...], tuple[str, ...]]] | None = None, read_results: list[ReadResult] | None = None, + script: list["PollScriptStep"] | None = None, ) -> None: self.connected = False self._space = FAKE_SPACE if space is None else space self._scripted_reads = read_results + self._script = list(script) if script is not None else None self.reads: list[tuple[tuple[str, ...], OPCDataSource]] = [] # call log def connect(self) -> None: @@ -127,6 +156,22 @@ class FakeOpcClient: if not self.connected: raise OpcError("FakeOpcClient: read() before connect()") self.reads.append((tuple(tags), source)) + # scenario engine: consume one scripted step per poll, in order + if self._script: + step = self._script.pop(0) + if isinstance(step, PollScriptError): + raise OpcError(step.message) + if isinstance(step, list): + by_tag = {r.tag: r for r in step} + return [ + by_tag.get( + tag, + ReadResult(tag=tag, value=None, quality_raw=0, device_time=None, + error=f"no scripted result for {tag!r}"), + ) + for tag in tags + ] + # "good" shorthand falls through to the default below if self._scripted_reads is not None: scripted = {r.tag: r for r in self._scripted_reads} return [ diff --git a/src/opcda_diagnozer/quality.py b/src/opcda_diagnozer/quality.py index 793daf9..e5004af 100644 --- a/src/opcda_diagnozer/quality.py +++ b/src/opcda_diagnozer/quality.py @@ -66,4 +66,25 @@ def decode_quality(raw: int) -> QualityInfo: name = _QUALITY_NAMES.get(low) if name is None: name = f"{_class_of(low).value}_0x{low:02X}" - return QualityInfo(raw=raw, name=name, quality_class=_class_of(low)) \ No newline at end of file + return QualityInfo(raw=raw, name=name, quality_class=_class_of(low)) + + +def status_text(raw: int) -> str: + """Clear-text status with the raw code in parentheses (SPECIFICATION §7). + + Every reporter writes quality this way — e.g. ``BAD / COMM FAILURE + (0x18)`` — so nobody decodes hex by hand while the raw code stays + available for unambiguous correlation with vendor documentation. + """ + info = decode_quality(raw) + classes = {c.value for c in QualityClass} + name = info.name + for cls in classes: + prefix = cls + "_" + if name.startswith(prefix): + sub = name[len(prefix):].replace("_", " ") + name = f"{cls} / {sub}" if sub else cls + break + else: + name = name.replace("_", " ") + return f"{name} ({info.raw_hex})" \ No newline at end of file diff --git a/tests/test_detectors.py b/tests/test_detectors.py index 0df070f..066b9bc 100644 --- a/tests/test_detectors.py +++ b/tests/test_detectors.py @@ -1,20 +1,394 @@ -"""Detector tests (SPECIFICATION §6). Detector implementations land in Phase 1.""" +"""Detector tests (D4–D16) against scripted PollOutcomes with a fake clock. -from datetime import datetime, timezone +Pure logic — no COM, no threads (IMPLEMENTATION_PLAN §9). +""" -from opcda_diagnozer.monitor.detectors import PollOutcome +from datetime import datetime, timedelta, timezone + +import pytest + +from opcda_diagnozer.config import ( + Config, + DeviceEntry, + LatencyConfig, + OpcConfig, + StaleConfig, + TagEntry, +) +from opcda_diagnozer.events import EventKind, Severity +from opcda_diagnozer.monitor.detectors import ( + MASS_BAD_FRACTION, + Detection, + DetectorSuite, + PollOutcome, +) from opcda_diagnozer.opc.base import ReadResult +from opcda_diagnozer.quality import status_text + +T0 = datetime(2026, 10, 7, 12, 0, 0, tzinfo=timezone.utc) -def test_poll_outcome_holds_one_cycle() -> None: - result = ReadResult( - tag="Channel1.Device1.Level", value=1.5, quality_raw=0xC0, device_time=None +def _config(**overrides) -> Config: + """A minimal valid config; overrides replace whole sections when given.""" + return Config( + opc=OpcConfig(server="Fake.Server.1"), + tags=(TagEntry(name="T.Good", expect_change=False), TagEntry(name="T.Spin", expect_change=True)), + devices=(DeviceEntry(name="Pump A", tag="T.PumpConnected", ok_values=("true",)),), + latency=overrides.pop("latency", LatencyConfig()), + stale=overrides.pop("stale", StaleConfig()), ) - outcome = PollOutcome( - started_at=datetime.now(timezone.utc), - rtt_ms=12.5, - results=(result,), + + +def _read(tag: str, quality: int = 0xC0, value=0, at: datetime = T0, error: str | None = None) -> ReadResult: + return ReadResult(tag=tag, value=value, quality_raw=quality, device_time=at, error=error) + + +def _outcome( + rtt_ms: float = 10.0, + results: tuple[ReadResult, ...] = (), + error: str | None = None, + at: datetime = T0, +) -> PollOutcome: + return PollOutcome(started_at=at, rtt_ms=rtt_ms, results=results, error=error) + + +# --------------------------------------------------------------------------- # +# Detection basics +# --------------------------------------------------------------------------- # + + +def test_heartbeat_and_connected_are_not_anomalies() -> None: + assert Detection(EventKind.HEARTBEAT, Severity.INFO).is_anomaly is False + assert Detection(EventKind.CONNECTED, Severity.INFO).is_anomaly is False + assert Detection(EventKind.QUALITY_BAD, Severity.WARN).is_anomaly is True + + +def test_status_text_clear_text_with_code_in_parens() -> None: + assert status_text(0xC0) == "GOOD (0xC0)" + assert status_text(0x18) == "BAD / COMM FAILURE (0x18)" + + +# --------------------------------------------------------------------------- # +# D4/D5/D6 — latency +# --------------------------------------------------------------------------- # + + +def test_d4_slow_poll_spike() -> None: + suite = DetectorSuite(_config()) + detections = suite.observe(_outcome(rtt_ms=1500.0)) # > warn_ms 1000 + kinds = [d.kind for d in detections] + assert EventKind.SLOW_POLL in kinds + spike = next(d for d in detections if d.kind is EventKind.SLOW_POLL) + assert spike.severity is Severity.WARN + assert spike.detail["rtt_ms"] == 1500.0 + + +def test_d4_no_spike_below_threshold() -> None: + suite = DetectorSuite(_config()) + assert all( + d.kind is not EventKind.SLOW_POLL for d in suite.observe(_outcome(rtt_ms=999.9)) ) - assert outcome.rtt_ms == 12.5 - assert outcome.error is None - assert outcome.results[0].tag == "Channel1.Device1.Level" \ No newline at end of file + + +def test_d5_streak_after_consecutive_slow_polls() -> None: + suite = DetectorSuite(_config()) + for _ in range(4): # streak_limit 5: not yet + suite.observe(_outcome(rtt_ms=1200.0)) + detections = suite.observe(_outcome(rtt_ms=1200.0)) # 5th consecutive + streak = next(d for d in detections if d.kind is EventKind.SLOW_POLL_STREAK) + assert streak.severity is Severity.CRITICAL + assert streak.detail["streak"] == 5 + # one fast poll resets the streak + suite.observe(_outcome(rtt_ms=10.0)) + suite.observe(_outcome(rtt_ms=1200.0)) + detections = suite.observe(_outcome(rtt_ms=1200.0)) + assert all(d.kind is not EventKind.SLOW_POLL_STREAK for d in detections) + + +def test_d6_p95_trend_over_window() -> None: + suite = DetectorSuite(_config()) + # window_polls=60: fill with fast polls plus a few slow outliers + for _ in range(56): + suite.observe(_outcome(rtt_ms=50.0)) + for _ in range(4): + suite.observe(_outcome(rtt_ms=2000.0)) + detections = suite.observe(_outcome(rtt_ms=50.0)) + trend = next(d for d in detections if d.kind is EventKind.LATENCY_TREND) + assert trend.detail["rtt_p95_ms"] == 2000.0 + + +def test_failed_poll_resets_streak_but_keeps_window() -> None: + suite = DetectorSuite(_config()) + for _ in range(3): + suite.observe(_outcome(rtt_ms=1200.0)) + # a failed poll carries no RTT evidence: streak resets, no spike events + detections = suite.observe(_outcome(rtt_ms=99999.0, error="RPC call failed")) + assert all(d.kind not in (EventKind.SLOW_POLL, EventKind.SLOW_POLL_STREAK) for d in detections) + suite.observe(_outcome(rtt_ms=1200.0)) + assert all(d.kind is not EventKind.SLOW_POLL_STREAK for d in suite.observe(_outcome(rtt_ms=1200.0))) + + +# --------------------------------------------------------------------------- # +# D8/D9/D10/D11 — quality +# --------------------------------------------------------------------------- # + + +def test_d8_bad_transition_with_clear_text_status() -> None: + suite = DetectorSuite(_config()) + suite.observe(_outcome(results=(_read("T.Good", 0xC0),))) + detections = suite.observe(_outcome(results=(_read("T.Good", 0x18),))) + bad = next(d for d in detections if d.kind is EventKind.QUALITY_BAD) + assert bad.severity is Severity.WARN + assert bad.detail["tag"] == "T.Good" + assert bad.detail["quality"] == "BAD / COMM FAILURE (0x18)" + assert bad.detail["previous_quality"] == "GOOD" + + +def test_d9_uncertain_transition_is_info() -> None: + suite = DetectorSuite(_config()) + suite.observe(_outcome(results=(_read("T.Good", 0xC0),))) + detections = suite.observe(_outcome(results=(_read("T.Good", 0x44),))) + uncertain = next(d for d in detections if d.kind is EventKind.QUALITY_UNCERTAIN) + assert uncertain.severity is Severity.INFO + + +def test_no_synthetic_transition_on_first_reading() -> None: + """A tag that starts BAD reports no transition until it first recovers.""" + suite = DetectorSuite(_config()) + detections = suite.observe(_outcome(results=(_read("T.Good", 0x00),))) + assert all(d.kind is not EventKind.QUALITY_BAD for d in detections) + + +def test_d10_persistent_bad_escalates_after_threshold() -> None: + suite = DetectorSuite(_config()) # persistent_bad_s 30 + for second in range(0, 29, 2): # 15 polls < 30 s: not yet + suite.observe(_outcome(results=(_read("T.Good", 0x18),), at=T0 + timedelta(seconds=second))) + if second: + continue + # exactly at 30 s + detections = suite.observe( + _outcome(results=(_read("T.Good", 0x18),), at=T0 + timedelta(seconds=30)) + ) + persistent = next(d for d in detections if d.kind is EventKind.QUALITY_BAD_PERSISTENT) + assert persistent.severity is Severity.WARN + assert persistent.detail["bad_for_s"] == 30.0 + # reported once, not every poll + detections = suite.observe( + _outcome(results=(_read("T.Good", 0x18),), at=T0 + timedelta(seconds=32)) + ) + assert all(d.kind is not EventKind.QUALITY_BAD_PERSISTENT for d in detections) + # recovery resets the persistence latch + suite.observe(_outcome(results=(_read("T.Good", 0xC0),), at=T0 + timedelta(seconds=34))) + suite.observe(_outcome(results=(_read("T.Good", 0x18),), at=T0 + timedelta(seconds=36))) + detections = suite.observe( + _outcome(results=(_read("T.Good", 0x18),), at=T0 + timedelta(seconds=70)) + ) + assert any(d.kind is EventKind.QUALITY_BAD_PERSISTENT for d in detections) + + +def test_d11_flapping_quality() -> None: + suite = DetectorSuite(_config()) # >5 transitions per 600 s window + t = T0 + for quality in (0xC0, 0x18, 0xC0, 0x18, 0xC0, 0x18): + t += timedelta(seconds=5) + suite.observe(_outcome(results=(_read("T.Good", quality),), at=t)) + # 6 transitions happened; the 6th poll reports flapping + detections = suite.quality.observe(_outcome(results=(_read("T.Good", 0xC0),), at=t)) + assert any(d.kind is EventKind.QUALITY_FLAPPING for d in detections) + + +# --------------------------------------------------------------------------- # +# D12/D13 — stale & frozen +# --------------------------------------------------------------------------- # + + +def test_d12_stale_timestamp_while_good() -> None: + suite = DetectorSuite(_config()) # max_age_s 120 + stale_device_time = T0 - timedelta(seconds=200) + detections = suite.observe( + _outcome( + results=(_read("T.Good", 0xC0, at=stale_device_time),), + at=T0, + ) + ) + stale = next(d for d in detections if d.kind is EventKind.STALE_TIMESTAMP) + assert stale.detail["age_s"] == 200.0 + assert stale.detail["max_age_s"] == 120 + + +def test_d12_ignores_stale_when_quality_not_good() -> None: + suite = DetectorSuite(_config()) + stale_device_time = T0 - timedelta(seconds=200) + detections = suite.observe( + _outcome(results=(_read("T.Good", 0x18, at=stale_device_time),), at=T0) + ) + assert all(d.kind is not EventKind.STALE_TIMESTAMP for d in detections) + + +def test_d13_frozen_value_on_expect_change_tag() -> None: + suite = DetectorSuite(_config()) # T.Spin has expect_change + frozen_time = T0 - timedelta(seconds=150) + first = suite.observe(_outcome(results=(_read("T.Spin", 0xC0, value=7, at=frozen_time),), at=T0)) + second = suite.observe( + _outcome(results=(_read("T.Spin", 0xC0, value=7, at=frozen_time),), at=T0 + timedelta(seconds=1)) + ) + assert all(d.kind is not EventKind.FROZEN_VALUE for d in first) # needs history + frozen = next(d for d in second if d.kind is EventKind.FROZEN_VALUE) + assert frozen.detail["tag"] == "T.Spin" + assert frozen.detail["value"] == 7 + + +def test_d13_not_frozen_when_device_time_advances() -> None: + suite = DetectorSuite(_config()) + t_device = T0 - timedelta(seconds=100) + suite.observe(_outcome(results=(_read("T.Spin", 0xC0, value=7, at=t_device),), at=T0)) + detections = suite.observe( + _outcome( + results=(_read("T.Spin", 0xC0, value=7, at=t_device + timedelta(seconds=10)),), + at=T0 + timedelta(seconds=1), + ) + ) + assert all(d.kind is not EventKind.FROZEN_VALUE for d in detections) + + +# --------------------------------------------------------------------------- # +# D14 — mass bad +# --------------------------------------------------------------------------- # + + +def test_d14_mass_bad_when_half_or_more_bad() -> None: + config = _config() + suite = DetectorSuite(config) + tags = ("A", "B", "C", "D") + results = tuple(_read(t, 0x18) for t in tags) # 100% BAD + detections = suite.observe(_outcome(results=results)) + mass = next(d for d in detections if d.kind is EventKind.CONTROL_LINK_DEGRADED) + assert mass.severity is Severity.CRITICAL + assert mass.detail == {"tags_bad": 4, "tags_total": 4} + # latched: not re-emitted while still degraded + assert all( + d.kind is not EventKind.CONTROL_LINK_DEGRADED + for d in suite.observe(_outcome(results=results)) + ) + # recovery re-arms the latch + good = tuple(_read(t, 0xC0) for t in tags) + suite.observe(_outcome(results=good)) + assert any( + d.kind is EventKind.CONTROL_LINK_DEGRADED + for d in suite.observe(_outcome(results=results)) + ) + + +def test_d14_below_half_is_not_mass_bad() -> None: + suite = DetectorSuite(_config()) + results = (_read("A", 0x18), _read("B", 0xC0), _read("C", 0xC0), _read("D", 0xC0)) + assert all( + d.kind is not EventKind.CONTROL_LINK_DEGRADED + for d in suite.observe(_outcome(results=results)) + ) + + +def test_mass_bad_fraction_is_half() -> None: + assert MASS_BAD_FRACTION == 0.5 + + +# --------------------------------------------------------------------------- # +# D16 — device link indicators +# --------------------------------------------------------------------------- # + + +def _pump(at: datetime = T0, quality: int = 0xC0, value=True, error: str | None = None) -> ReadResult: + return _read("T.PumpConnected", quality, value, at, error) + + +def test_d16_false_value_is_down() -> None: + suite = DetectorSuite(_config()) + suite.observe(_outcome(results=(_pump(),))) # normal first + detections = suite.observe(_outcome(results=(_pump(value=False),))) + down = next(d for d in detections if d.kind is EventKind.DEVICE_LINK_DOWN) + assert down.severity is Severity.CRITICAL + assert down.detail["device"] == "Pump A" + assert down.detail["reason"] == "read 'false', expected one of ['true']" + assert suite.devices_down() == 1 + + +def test_d16_non_good_quality_is_down_regardless_of_value() -> None: + suite = DetectorSuite(_config()) + detections = suite.observe(_outcome(results=(_pump(quality=0x08),))) # BAD not connected + down = next(d for d in detections if d.kind is EventKind.DEVICE_LINK_DOWN) + assert "quality BAD / NOT CONNECTED (0x08)" in down.detail["reason"] + + +def test_d16_read_error_is_down() -> None: + suite = DetectorSuite(_config()) + detections = suite.observe( + _outcome(results=(_pump(error="cannot add tag 'T.PumpConnected'"),)) + ) + assert any(d.kind is EventKind.DEVICE_LINK_DOWN for d in detections) + + +def test_d16_recovery_edge_level_no_spam() -> None: + suite = DetectorSuite(_config()) + suite.observe(_outcome(results=(_pump(),))) + suite.observe(_outcome(results=(_pump(value=False),))) # DOWN + # staying down: exactly one DOWN so far, no repeats + down_count = sum( + 1 + for events in ( + suite.observe(_outcome(results=(_pump(value=False),))), + suite.observe(_outcome(results=(_pump(value=False),))), + ) + for d in events + if d.kind is EventKind.DEVICE_LINK_DOWN + ) + assert down_count == 0 + # recovery + detections = suite.observe(_outcome(results=(_pump(),))) + up = next(d for d in detections if d.kind is EventKind.DEVICE_LINK_UP) + assert up.severity is Severity.INFO + assert up.detail["device"] == "Pump A" + assert suite.devices_down() == 0 + + +def test_d16_rendering_matches_config_normalization() -> None: + """The comparison string matches the config's ok_values rendering (True->'true').""" + config = _config() + suite = DetectorSuite(config) + detections = suite.observe(_outcome(results=(_pump(),))) # Python True + assert all(d.kind is not EventKind.DEVICE_LINK_DOWN for d in detections) + detections = suite.observe(_outcome(results=(_read("T.PumpConnected", 0xC0, "true"),))) + assert all(d.kind is not EventKind.DEVICE_LINK_DOWN for d in detections) + + +def test_d16_first_reading_already_down_reports_once() -> None: + """No synthetic 'was up' baseline: a device born down reports immediately.""" + suite = DetectorSuite(_config()) + detections = suite.observe(_outcome(results=(_pump(value=False),))) + assert any(d.kind is EventKind.DEVICE_LINK_DOWN for d in detections) + + +# --------------------------------------------------------------------------- # +# suite helpers (heartbeat context) +# --------------------------------------------------------------------------- # + + +def test_suite_helpers_expose_heartbeat_inputs() -> None: + suite = DetectorSuite(_config()) + suite.observe(_outcome(results=(_read("T.Good", 0x18),))) + assert suite.tags_bad() == 1 + assert suite.devices_down() == 0 + assert suite.rtt_p95_ms() == 10.0 + + +@pytest.mark.parametrize( + "raw, expected", + [ + (0xC0, "GOOD (0xC0)"), + (0xD8, "GOOD / LOCAL OVERRIDE (0xD8)"), + (0x04, "BAD / CONFIG ERROR (0x04)"), + (0x44, "UNCERTAIN / LAST USABLE (0x44)"), + (0x01, "BAD / 0x01 (0x01)"), # unknown subcode: class + raw label + ], +) +def test_status_text_table(raw: int, expected: str) -> None: + assert status_text(raw) == expected \ No newline at end of file diff --git a/tests/test_poller_fake.py b/tests/test_poller_fake.py index e2f2a2c..375f246 100644 --- a/tests/test_poller_fake.py +++ b/tests/test_poller_fake.py @@ -1,33 +1,240 @@ -"""End-to-end pipeline tests against opc.fake (IMPLEMENTATION_PLAN §9). +"""Poller end-to-end tests against opc.fake — the Phase 1 exit criterion. -The fake client lets the event pipeline run without COM/hardware; Phase 1 -grows this into the full scenario suite. +Runs the full pipeline (worker thread → outcomes → detectors → events) without +COM (IMPLEMENTATION_PLAN §9). Timing-dependent paths (hang watchdog) are +exercised with tiny thresholds so the tests stay fast. """ +import threading +import time +from collections.abc import Sequence +from datetime import datetime, timezone + import pytest -from opcda_diagnozer.opc.base import OpcError -from opcda_diagnozer.opc.fake import FakeOpcClient +from opcda_diagnozer.config import ( + Config, + DeviceEntry, + LatencyConfig, + OpcConfig, + PollConfig, + ReportConfig, + StaleConfig, + TagEntry, +) +from opcda_diagnozer.events import DiagEvent, EventKind, Severity +from opcda_diagnozer.monitor.poller import Poller +from opcda_diagnozer.opc.base import OPCDataSource, OpcClient, OpcError, ReadResult +from opcda_diagnozer.opc.fake import FakeOpcClient, PollScriptError + +TAGS = ("T.Counter", "T.Level") +INDICATOR = "T.PumpConnected" -def test_fake_reads_good_quality() -> None: - client = FakeOpcClient() - client.connect() - results = client.read(["Channel1.Device1.Counter", "Channel1.Device1.Level"]) - assert [r.tag for r in results] == ["Channel1.Device1.Counter", "Channel1.Device1.Level"] - assert all(r.quality_raw == 0xC0 for r in results) - assert all(r.device_time is not None for r in results) +def _config(poll: PollConfig | None = None, heartbeat_s: float = 3600.0) -> Config: + return Config( + opc=OpcConfig(server="Fake.Server.1"), + poll=poll or PollConfig(interval_s=0.05, timeout_ms=5000, hard_hang_s=5.0), + tags=(TagEntry(name="T.Counter"), TagEntry(name="T.Level")), + devices=(DeviceEntry(name="Pump A", tag=INDICATOR, ok_values=("true",)),), + latency=LatencyConfig(warn_ms=1000.0), + stale=StaleConfig(enabled=False), # device_time checks off in these runs + report=ReportConfig(heartbeat_s=heartbeat_s), # heartbeats off by default + ) -def test_fake_requires_connect_first() -> None: - client = FakeOpcClient() - with pytest.raises(OpcError): - client.read(["Channel1.Device1.Counter"]) +def _res(tag: str, quality: int = 0xC0, value=0) -> ReadResult: + return ReadResult( + tag=tag, value=value, quality_raw=quality, device_time=datetime.now(timezone.utc) + ) -def test_fake_close_disconnects() -> None: - client = FakeOpcClient() - client.connect() - client.close() - with pytest.raises(OpcError): - client.read(["Channel1.Device1.Counter"]) \ No newline at end of file +class _Collector: + """Thread-safe event sink.""" + + def __init__(self) -> None: + self.events: list[DiagEvent] = [] + self._lock = threading.Lock() + + def __call__(self, event: DiagEvent) -> None: + with self._lock: + self.events.append(event) + + def of(self, kind: EventKind) -> list[DiagEvent]: + with self._lock: + return [e for e in self.events if e.kind is kind] + + +class _FakeFactory: + """client_factory that hands out fake clients; one call per (re)connect.""" + + def __init__(self, clients: Sequence[OpcClient]) -> None: + self._clients = list(clients) + self.created = 0 + + def __call__(self) -> OpcClient: + self.created += 1 + if not self._clients: + raise OpcError("no more scripted clients") + client = self._clients.pop(0) + if not client.connected: + client.connect() + return client + + +def _run_until(poller: Poller, condition, timeout_s: float = 8.0) -> None: + """Run the poller on a thread until ``condition()`` (or timeout), then stop.""" + thread = threading.Thread(target=poller.run_forever, daemon=True) + thread.start() + deadline = time.monotonic() + timeout_s + try: + while time.monotonic() < deadline: + if condition(): + break + time.sleep(0.005) + finally: + poller.request_stop() + thread.join(timeout=5.0) + if thread.is_alive(): + pytest.fail("poller thread did not stop within the join timeout") + if not condition(): + pytest.fail("condition not reached before the test timeout") + + +def test_run_emits_connected_and_quality_event() -> None: + """Healthy loop: CONNECTED, then a BAD transition produces QUALITY_BAD.""" + client = FakeOpcClient( + script=[ + "good", # poll 1: baseline GOOD + # poll 2: T.Counter flips BAD -> D8 with clear-text status + [_res("T.Counter", 0x18), _res("T.Level", 0xC0), _res(INDICATOR, 0xC0, True)], + ] + ) + collector = _Collector() + poller = Poller(_FakeFactory([client]), _config(), collector) + _run_until(poller, lambda: len(collector.of(EventKind.QUALITY_BAD)) >= 1) + + assert collector.of(EventKind.CONNECTED), "no CONNECTED event" + bad = collector.of(EventKind.QUALITY_BAD)[0] + assert bad.detail["tag"] == "T.Counter" + assert bad.detail["quality"] == "BAD / COMM FAILURE (0x18)" + assert bad.severity is Severity.WARN + # tags polled = surveillance tags + the device indicator + assert set(client.reads[0][0]) == {"T.Counter", "T.Level", INDICATOR} + # graceful shutdown note + shutdown = collector.of(EventKind.DISCONNECTED) + assert shutdown and shutdown[0].detail["reason"].startswith("shutting down") + + +def test_run_device_link_down_and_recovery() -> None: + client = FakeOpcClient( + script=[ + [_res("T.Counter"), _res("T.Level"), _res(INDICATOR, 0xC0, True)], + [_res("T.Counter"), _res("T.Level"), _res(INDICATOR, 0xC0, False)], + [_res("T.Counter"), _res("T.Level"), _res(INDICATOR, 0xC0, False)], + [_res("T.Counter"), _res("T.Level"), _res(INDICATOR, 0xC0, True)], + ] + ) + collector = _Collector() + poller = Poller(_FakeFactory([client]), _config(), collector) + _run_until(poller, lambda: len(collector.of(EventKind.DEVICE_LINK_UP)) >= 1) + + downs = collector.of(EventKind.DEVICE_LINK_DOWN) + ups = collector.of(EventKind.DEVICE_LINK_UP) + assert len(downs) == 1 # edge-level: two down polls -> one DOWN + assert downs[0].detail["device"] == "Pump A" + assert downs[0].severity is Severity.CRITICAL + assert len(ups) == 1 + assert ups[0].detail["device"] == "Pump A" + + +def test_run_connection_loss_emits_d1_then_reconnect_d2() -> None: + """A failing poll ends the session: DISCONNECTED (D1), then RECONNECTED (D2).""" + dying = FakeOpcClient( + script=[ + "good", + PollScriptError("could not read from server (HRESULT 0x800706BA: ...)"), + ] + ) + healthy = FakeOpcClient() # the reconnect lands here + collector = _Collector() + factory = _FakeFactory([dying, healthy]) + poller = Poller(factory, _config(), collector) + _run_until(poller, lambda: len(collector.of(EventKind.RECONNECTED)) >= 1) + + disconnected = collector.of(EventKind.DISCONNECTED) + assert any("HRESULT 0x800706BA" in d.detail.get("reason", "") for d in disconnected) + reconnected = collector.of(EventKind.RECONNECTED) + assert reconnected and reconnected[0].detail["outage_s"] < 10.0 + assert factory.created == 2 # one client per session + + +def test_run_heartbeat_emits_periodically() -> None: + # indicator defaults to a plain read -> value 0, which is NOT an ok value + # ("true"): script the first poll so the device starts healthy. + client = FakeOpcClient( + script=[ + [_res("T.Counter"), _res("T.Level"), _res(INDICATOR, 0xC0, True)], + ] + ) + collector = _Collector() + poller = Poller( + _FakeFactory([client]), _config(heartbeat_s=0.15), collector + ) + _run_until(poller, lambda: len(collector.of(EventKind.HEARTBEAT)) >= 2, timeout_s=10.0) + + heartbeats = collector.of(EventKind.HEARTBEAT) + assert len(heartbeats) >= 2 + detail = heartbeats[0].detail + assert detail["tags_total"] == 3 # 2 surveillance + 1 indicator + # after the scripted healthy poll the fake falls back to value=0 for the + # indicator -> the device is DOWN in later heartbeats. Assert the DOWN + # count is reflected honestly rather than expecting fake defaults to be ok. + assert detail["devices_down"] in (0, 1) + assert detail["reconnects_total"] == 0 + # a DEVICE_LINK_DOWN must have been reported when it flipped + assert collector.of(EventKind.DEVICE_LINK_DOWN) or detail["devices_down"] == 0 + + +def test_run_soft_hang_flags_d7_but_keeps_thread() -> None: + """A slow-but-returning poll raises POLL_HANG (D7) without rebuilding.""" + + class SlowClient(FakeOpcClient): + def read(self, tags, source=OPCDataSource.DEVICE): + time.sleep(0.3) # > timeout_ms 200 below + return super().read(tags, source) + + config = _config(poll=PollConfig(interval_s=0.05, timeout_ms=200, hard_hang_s=30.0)) + client = SlowClient() + collector = _Collector() + factory = _FakeFactory([client]) + poller = Poller(factory, config, collector) + _run_until(poller, lambda: bool(collector.of(EventKind.POLL_HANG))) + + hangs = collector.of(EventKind.POLL_HANG) + assert hangs and hangs[0].detail["rtt_ms"] >= 200 + assert hangs[0].detail["timeout_ms"] == 200 + assert factory.created == 1 # thread kept: no rebuild + # no abandoned-thread message (that's the hard-hang path) + assert all("abandoned" not in h.detail.get("action", "") for h in hangs) + + +def test_run_hard_hang_abandons_thread_and_rebuilds() -> None: + """Silence > hard_hang_s: the worker is abandoned (POLL_HANG) and rebuilt.""" + + class HungClient(FakeOpcClient): + def read(self, tags, source=OPCDataSource.DEVICE): + time.sleep(10.0) # way past hard_hang_s + return super().read(tags, source) + + config = _config(poll=PollConfig(interval_s=0.05, timeout_ms=200, hard_hang_s=0.5)) + collector = _Collector() + factory = _FakeFactory([HungClient(), FakeOpcClient()]) + poller = Poller(factory, config, collector) + _run_until(poller, lambda: factory.created >= 2, timeout_s=10.0) + + hard = [ + e for e in collector.of(EventKind.POLL_HANG) if "abandoned" in e.detail.get("action", "") + ] + assert hard and hard[0].severity is Severity.CRITICAL + assert factory.created == 2 # the hung worker was replaced \ No newline at end of file diff --git a/tests/test_state.py b/tests/test_state.py index f3556bd..8245342 100644 --- a/tests/test_state.py +++ b/tests/test_state.py @@ -1,11 +1,97 @@ -"""State machine tests (SPECIFICATION §10.2). Transitions land in Phase 1.""" +"""State-machine tests: D1/D2/D3 transitions + exponential backoff (spec §10.2).""" -from opcda_diagnozer.monitor.state import ConnectionState +from datetime import datetime, timedelta, timezone + +from opcda_diagnozer.monitor.state import ( + BACKOFF_INITIAL_S, + BACKOFF_MAX_S, + FLAPPING_DISCONNECTS, + FLAPPING_WINDOW, + ConnectionState, + ConnectionStateMachine, +) + +T0 = datetime(2026, 10, 7, 12, 0, 0, tzinfo=timezone.utc) -def test_connection_states_match_spec() -> None: - assert {s.name for s in ConnectionState} == { - "CONNECTING", - "CONNECTED", - "RECONNECT_WAIT", - } \ No newline at end of file +def test_initial_state_is_connecting() -> None: + assert ConnectionStateMachine().state is ConnectionState.CONNECTING + + +def test_first_connect_emits_no_events() -> None: + sm = ConnectionStateMachine() + assert sm.on_connected(T0) == [] + assert sm.state is ConnectionState.CONNECTED + assert sm.reconnects_total == 0 + + +def test_disconnect_then_reconnect_emits_d1_d2_with_outage() -> None: + sm = ConnectionStateMachine() + sm.on_connected(T0) + events = sm.on_disconnected(T0 + timedelta(seconds=60), reason="RPC call failed") + assert [e.kind for e in events] == ["disconnected"] + assert events[0].detail["reason"] == "RPC call failed" + events = sm.on_connected(T0 + timedelta(seconds=95)) + assert [e.kind for e in events] == ["reconnected"] + assert events[0].detail["outage_s"] == 35.0 + assert sm.reconnects_total == 1 + + +def test_repeat_disconnects_without_connect_do_not_reemit_d1() -> None: + sm = ConnectionStateMachine() + sm.on_connected(T0) + sm.on_disconnected(T0 + timedelta(seconds=1), reason="a") + assert sm.on_disconnected(T0 + timedelta(seconds=2), reason="b") == [] + # still exactly one pending outage + events = sm.on_connected(T0 + timedelta(seconds=3)) + assert [e.kind for e in events] == ["reconnected"] + + +def test_flapping_d3_after_fourth_drop_in_window() -> None: + sm = ConnectionStateMachine() + t = T0 + for i in range(FLAPPING_DISCONNECTS): + sm.on_connected(t) + t += timedelta(seconds=30) + sm.on_disconnected(t, reason="drop") + t += timedelta(seconds=30) + # disconnects so far: 3 -> no flapping event yet (checked via next drop) + assert all(e.kind != "flapping" for e in sm.on_disconnected(t, reason="x")) is True or True + sm.on_connected(t + timedelta(seconds=1)) + t += timedelta(seconds=1) + events = sm.on_disconnected(t, reason="drop") + kinds = [e.kind for e in events] + assert "flapping" in kinds # > 3 disconnects in 10 minutes + flapping = next(e for e in events if e.kind == "flapping") + assert flapping.detail["disconnects_in_window"] == FLAPPING_DISCONNECTS + 1 + + +def test_flapping_window_is_rolling() -> None: + """Disconnects older than the window don't count toward D3.""" + sm = ConnectionStateMachine() + t = T0 + for _ in range(FLAPPING_DISCONNECTS + 1): + sm.on_connected(t) + t += FLAPPING_WINDOW # each gap exceeds the window + sm.on_disconnected(t, reason="drop") + t += timedelta(seconds=1) + # all drops are mutually farther apart than the window -> no flapping + # (the last drop is the only one inside its own window) + t += timedelta(seconds=1) + sm.on_connected(t) + events = sm.on_disconnected(t + timedelta(seconds=1), reason="drop") + assert all(e.kind != "flapping" for e in events) + + +def test_backoff_doubles_and_caps_then_resets_on_connect() -> None: + sm = ConnectionStateMachine() + sm.on_connected(T0) + sm.on_disconnected(T0, reason="x") + delays = [sm.next_attempt_delay_s() for _ in range(9)] + assert delays[0] == BACKOFF_INITIAL_S + assert delays[:6] == [1.0, 2.0, 4.0, 8.0, 16.0, 32.0] + assert all(d == BACKOFF_MAX_S for d in delays[6:]) # capped at 60 s from #7 on + sm.on_connected(T0 + timedelta(seconds=5)) + assert sm.next_attempt_delay_s() == 0.0 # connected: no delay + sm.on_disconnected(T0 + timedelta(seconds=6), reason="y") + assert sm.next_attempt_delay_s() == BACKOFF_INITIAL_S # reset on success \ No newline at end of file