From 1478db73a9da73a0ad3d7cf1fa5a44ebafeaf1b9 Mon Sep 17 00:00:00 2001 From: Jonas Bengtsson Date: Thu, 8 Oct 2026 07:57:21 +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 | 22 ++ README.md | 10 +- TODO.md | 27 +- config.example.yaml | 10 +- requirements.dev.txt | 18 ++ requirements.txt | 21 ++ src/opcda_diagnozer/config.py | 32 ++- src/opcda_diagnozer/main.py | 76 ++++- src/opcda_diagnozer/report/base.py | 66 ++++- src/opcda_diagnozer/report/eventlog.py | 79 +++++- src/opcda_diagnozer/report/http.py | 104 ++++++- src/opcda_diagnozer/report/jsonl.py | 70 ++++- src/opcda_diagnozer/report/textlog.py | 120 ++++++++ tests/test_reporters.py | 367 +++++++++++++++++++++++++ 14 files changed, 972 insertions(+), 50 deletions(-) create mode 100644 requirements.dev.txt create mode 100644 requirements.txt create mode 100644 src/opcda_diagnozer/report/textlog.py create mode 100644 tests/test_reporters.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 5e4360f..bba2fce 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,28 @@ bump per [SemVer](https://semver.org/). Dates are YYYY-MM-DD. ## [Unreleased] +### Added — Phase 2 reporters +- Rotating JSONL evidence log: one JSON event per line, config-digest header + for provenance, size rotation with backups retained. +- Plain-text anomaly log (spec §9.1): `anomalies-YYYYMMDD.log`, one line per + detected anomaly in `timestamp SEVERITY KIND message` order with clear-text + statuses; recoveries included so the file reads as an incident narrative. +- Windows Event Log sink: severity-mapped (INFO/WARN/CRITICAL → + Information/Warning/Error), WARN+ by default, delivery verified live. +- Optional HTTP webhook (`report.http`): bounded queue + daemon sender thread, + per-event POST with bearer auth; failures dropped and counted — never blocks + the poll loop. +- Reporter fan-out dispatcher: every sink isolated (one failing sink never + kills the others or polling); the `run` verb wires all configured sinks. + +### Fixed +- Windows Event Log reporter initially lost every event silently: + `win32evtlog.ReportEvent` takes **positional arguments only** (keyword calls + raise `TypeError`, swallowed by the sink's own isolation) — now uses a + `RegisterEventSource` handle with positional calls (live-verified). +- HTTP webhook in injected-`post` mode raised into the poll loop on a failing + POST; both modes now count-and-drop. + ### 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), diff --git a/README.md b/README.md index dd668e0..4891c72 100644 --- a/README.md +++ b/README.md @@ -162,6 +162,9 @@ browse: report: heartbeat_s: 300 jsonl: { dir: "C:/ProgramData/OPCDA-Diagnozer/logs", max_mb: 50, backups: 10 } + textlog: { dir: "C:/ProgramData/OPCDA-Diagnozer/logs", max_mb: 20, backups: 10 } + eventlog: { source: "OPCDA-Diagnozer", min_severity: WARN } + #http: { url: "https://alerts.example.com/api", bearer_token: "...", timeout_s: 5 } eventlog: { source: "OPCDA-Diagnozer", min_severity: WARN } ``` @@ -208,7 +211,7 @@ so the Python interpreter **must be 32-bit** (SPECIFICATION §11 / Phase 0 decis ```powershell # create the 32-bit venv (requires the 32-bit Python installed via the py launcher) & "C:\Users\\AppData\Local\Programs\Python\Python314-32\python.exe" -m venv .venv -& ".\.venv\Scripts\python.exe" -m pip install -e ".[dev]" +& ".\.venv\Scripts\python.exe" -m pip install -e . -r requirements.dev.txt # verify bitness — must print 32 & ".\.venv\Scripts\python.exe" -c "import struct; print(struct.calcsize('P') * 8)" @@ -217,6 +220,11 @@ so the Python interpreter **must be 32-bit** (SPECIFICATION §11 / Phase 0 decis & ".\.venv\Scripts\python.exe" -m pytest ``` +Dependency files: `requirements.txt` (production) and +`requirements.dev.txt` (adds pytest + pyinstaller, chains production via +`-r requirements.txt`). `pip install -e ".[dev]"` works too — +`pyproject.toml` stays the canonical package source. + Notes: - Dependencies are chosen to be installable without a C compiler: **ruamel.yaml** (pure Python) is used instead of PyYAML, which publishes no 32-bit CPython 3.14 diff --git a/TODO.md b/TODO.md index 5cf891a..2d801f1 100644 --- a/TODO.md +++ b/TODO.md @@ -50,19 +50,24 @@ Legend: **[done]** complete & verified · **[partial]** some of it landed · 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 +## Phase 2 — Reporters **[done]** -- [todo] 1. `report/jsonl.py` — rotating JSONL (size + backups), config-digest - 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): +- [done] 1. `report/jsonl.py` — rotating JSONL (size + backups), config-digest + header line, single-line atomic-ish appends. Live-verified. +- [done] 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. -- [todo] 3. `report/eventlog.py` — Windows Event Log (WARN+), source - registration via `install`. -- [todo] 4. `report/http.py` — optional webhook, bounded queue, never blocks - polling. -- [todo] 5. Heartbeat assembly (D15) + reporter fan-out dispatcher. + `NAME / SUB (0xNN)`, recoveries logged (incident narrative), per-day files, + shared rotation. Live-verified. +- [done] 3. `report/eventlog.py` — Windows Event Log (WARN+), severity-mapped + via a RegisterEventSource handle + positional ReportEvent (keyword args + raise TypeError — found live). Source message-file registration lands with + `install` (Phase 3). Live-verified delivery. +- [done] 4. `report/http.py` — optional webhook (`report.http` config), + bounded queue + daemon sender, never blocks the poll loop; failures counted + and dropped, injectable `post` for tests. +- [done] 5. Heartbeat assembly (D15, Phase 1) + reporter fan-out dispatcher + (`report/base.py`): per-sink failure isolation, best-effort close. The + `run` verb wires console + JSONL + textlog + eventlog (+http when set). ## Phase 3 — Windows service diff --git a/config.example.yaml b/config.example.yaml index 085c073..a4f9d85 100644 --- a/config.example.yaml +++ b/config.example.yaml @@ -53,6 +53,14 @@ report: dir: "C:/ProgramData/OPCDA-Diagnozer/logs" max_mb: 50 # rotate after this size backups: 10 # rotated files kept + textlog: # human-readable anomaly log (spec 9.1) + dir: "C:/ProgramData/OPCDA-Diagnozer/logs" + max_mb: 20 + backups: 10 eventlog: source: "OPCDA-Diagnozer" - min_severity: WARN # INFO / WARN / CRITICAL \ No newline at end of file + min_severity: WARN # INFO / WARN / CRITICAL + #http: # optional webhook; omit = off + # url: "https://alerts.example.com/api/opcda" + # bearer_token: "..." + # timeout_s: 5 \ No newline at end of file diff --git a/requirements.dev.txt b/requirements.dev.txt new file mode 100644 index 0000000..402a872 --- /dev/null +++ b/requirements.dev.txt @@ -0,0 +1,18 @@ +# OPC DA Diagnozer — development dependencies. +# +# Install: pip install -r requirements.dev.txt +# (pulls in the production dependencies via -r requirements.txt) +# +# The interpreter MUST be 32-bit (SPECIFICATION §11) — see README +# "Development setup" for the venv procedure. + +-r requirements.txt + +pytest>=8.0 +# The test suite (165 tests): detectors/state/poller/reporters against the +# fake client — runs on any OS (COM paths are import-guarded and live-smoke +# tested on Windows dev boxes only). + +pyinstaller>=6.22 +# Builds the standalone 32-bit exe (scripts/build_exe.ps1) — no Python +# installation needed on the target host. Not needed at runtime. \ No newline at end of file diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..3b822cc --- /dev/null +++ b/requirements.txt @@ -0,0 +1,21 @@ +# OPC DA Diagnozer — production dependencies. +# +# Install: pip install -r requirements.txt +# The interpreter MUST be 32-bit (SPECIFICATION §11): the OPC DA Automation +# Wrapper and Graybox DLLs are 32-bit in-proc COM servers — a 64-bit process +# cannot load them. Create the venv with the 32-bit Python (README +# "Development setup"). +# +# Note: pyproject.toml remains the canonical dependency source for the +# package install (`pip install -e .`); this file mirrors it for tooling +# that expects plain requirements files (pip install -r, constraint sync). + +ruamel.yaml>=0.18 +# YAML config loading (config.py). Pure Python — PyYAML publishes no win32 +# (32-bit) wheels for CPython 3.14 and its sdist requires MSVC, which the +# diag hosts lack. + +pywin32>=306; sys_platform == "win32" +# COM access (pythoncom, win32com, win32evtlog) for the Automation Wrapper +# adapter, discovery and the Windows Event Log reporter. Every import is +# guarded so tests still run on any OS (IMPLEMENTATION_PLAN §9). \ No newline at end of file diff --git a/src/opcda_diagnozer/config.py b/src/opcda_diagnozer/config.py index 80d292e..c0f1a6a 100644 --- a/src/opcda_diagnozer/config.py +++ b/src/opcda_diagnozer/config.py @@ -103,11 +103,22 @@ class EventLogConfig: min_severity: str = "WARN" # INFO / WARN / CRITICAL +@dataclass(frozen=True) +class HttpConfig: + """Optional webhook reporter (spec §9): one JSON event per POST.""" + + url: str + bearer_token: str | None = None + timeout_s: float = 5.0 + + @dataclass(frozen=True) class ReportConfig: heartbeat_s: float = 300.0 # D15 jsonl: JsonlConfig = field(default_factory=JsonlConfig) + textlog: JsonlConfig = field(default_factory=JsonlConfig) # same knobs: dir/max_mb/backups eventlog: EventLogConfig = field(default_factory=EventLogConfig) + http: HttpConfig | None = None # omitted = webhook off @dataclass(frozen=True) @@ -183,9 +194,14 @@ def _build_section(cls: type[Any], data: Any, where: str) -> Any: def _build_report(data: Any) -> ReportConfig: data = {} if data is None else dict(data) jsonl = _build_section(JsonlConfig, data.pop("jsonl", None), "report.jsonl") + textlog = _build_section(JsonlConfig, data.pop("textlog", None), "report.textlog") eventlog = _build_section(EventLogConfig, data.pop("eventlog", None), "report.eventlog") + http_raw = data.pop("http", None) + http = None + if http_raw is not None: + http = _build_section(HttpConfig, http_raw, "report.http") rest = _build_section(ReportConfig, data, "report") - return replace(rest, jsonl=jsonl, eventlog=eventlog) + return replace(rest, jsonl=jsonl, textlog=textlog, eventlog=eventlog, http=http) def _build_tags(data: Any) -> tuple[TagEntry, ...]: @@ -328,6 +344,20 @@ def _validate(config: Config) -> None: raise ConfigError(f"report.jsonl.max_mb: must be >= 1, got {config.report.jsonl.max_mb}") if config.report.jsonl.backups < 0: raise ConfigError(f"report.jsonl.backups: must be >= 0, got {config.report.jsonl.backups}") + if config.report.textlog.max_mb < 1: + raise ConfigError(f"report.textlog.max_mb: must be >= 1, got {config.report.textlog.max_mb}") + if config.report.textlog.backups < 0: + raise ConfigError(f"report.textlog.backups: must be >= 0, got {config.report.textlog.backups}") + if config.report.http is not None: + if not config.report.http.url.startswith(("http://", "https://")): + raise ConfigError( + "report.http.url: must start with http:// or https://, got " + f"{config.report.http.url!r}" + ) + if config.report.http.timeout_s <= 0: + raise ConfigError( + f"report.http.timeout_s: must be > 0, got {config.report.http.timeout_s}" + ) def load_config(path: str | Path) -> Config: diff --git a/src/opcda_diagnozer/main.py b/src/opcda_diagnozer/main.py index e5ab26c..99b07ac 100644 --- a/src/opcda_diagnozer/main.py +++ b/src/opcda_diagnozer/main.py @@ -554,10 +554,70 @@ def _cmd_probe( return _cmd_read(config_path, server, host, (), source, as_json=False) +def _build_reporters(config: Config): + """Assemble the reporter fan-out from the config (spec §9). + + Console is always on in the foreground ``run``; JSONL, the text anomaly + log and the Windows Event Log follow their config sections; the webhook + only when ``report.http`` is configured. Returns ``(fan_out, close)`` + where ``close`` must be called exactly once on shutdown. + """ + from opcda_diagnozer.report.base import FanOut + from opcda_diagnozer.report.jsonl import JsonlReporter, config_digest + from opcda_diagnozer.report.textlog import TextLogReporter + + sinks = [] + + def console_reporter(event: DiagEvent) -> 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) + + sinks.append(console_reporter) + sinks.append( + JsonlReporter( + config.report.jsonl.dir, + max_mb=config.report.jsonl.max_mb, + backups=config.report.jsonl.backups, + digest=config_digest(config), + ) + ) + sinks.append( + TextLogReporter( + config.report.textlog.dir, + max_mb=config.report.textlog.max_mb, + backups=config.report.textlog.backups, + ) + ) + if sys.platform == "win32": + from opcda_diagnozer.report.eventlog import EventLogReporter + + sinks.append( + EventLogReporter( + source=config.report.eventlog.source, + min_severity=config.report.eventlog.min_severity, + ) + ) + if config.report.http is not None: + from opcda_diagnozer.report.http import HttpReporter + + sinks.append( + HttpReporter( + url=config.report.http.url, + bearer_token=config.report.http.bearer_token, + timeout_s=config.report.http.timeout_s, + ) + ) + fan_out = FanOut(tuple(sinks), error=lambda msg: print(f"run: {msg}", file=sys.stderr, flush=True)) + return fan_out + + def _cmd_run(config_path: str, server: str | None, host: str | None) -> int: - """Foreground monitoring loop: poll → detect → print (Ctrl+C stops).""" + """Foreground monitoring loop: poll → detect → report (Ctrl+C stops).""" import signal + from opcda_diagnozer.events import DiagEvent from opcda_diagnozer.monitor.poller import Poller from opcda_diagnozer.opc.automation import AutomationOpcClient @@ -587,13 +647,8 @@ def _cmd_run(config_path: str, server: str | None, host: str | None) -> int: 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) + reporters = _build_reporters(config) + poller = Poller(client_factory, config, reporters.emit) stop_requested = threading.Event() def _on_ctrl_c(signum, frame) -> None: # noqa: ANN001, ARG001 @@ -604,7 +659,10 @@ def _cmd_run(config_path: str, server: str | None, host: str | None) -> int: print("\nrun: stop requested - finishing current poll ...", file=sys.stderr, flush=True) signal.signal(signal.SIGINT, _on_ctrl_c) - poller.run_forever() + try: + poller.run_forever() + finally: + reporters.close() return 0 diff --git a/src/opcda_diagnozer/report/base.py b/src/opcda_diagnozer/report/base.py index fa4534c..6fc0e36 100644 --- a/src/opcda_diagnozer/report/base.py +++ b/src/opcda_diagnozer/report/base.py @@ -1,7 +1,8 @@ -"""Reporter protocol — pluggable event sinks (SPECIFICATION §9).""" +"""Reporter protocol + fan-out dispatcher (SPECIFICATION §9, IMPLEMENTATION_PLAN §5.4).""" from __future__ import annotations +from collections.abc import Callable from typing import Protocol from opcda_diagnozer.events import DiagEvent @@ -10,4 +11,65 @@ from opcda_diagnozer.events import DiagEvent class Reporter(Protocol): """Anything that can receive events; reporting never blocks the poll loop.""" - def emit(self, event: DiagEvent) -> None: ... \ No newline at end of file + def emit(self, event: DiagEvent) -> None: ... + + +def _emit_via(reporter: Reporter, event: DiagEvent) -> None: + """Deliver to a sink that is either an object with ``emit`` or a plain callable. + + The console sink (``_build_reporters``) is a function; the file/Event + Log/webhook sinks are objects. Both are legitimate ``Reporter`` + implementations — the protocol is "anything that can receive events" — + so the fan-out duck-types both shapes instead of forcing every sink + into a class. Anything else fails loudly (caught + counted by FanOut). + """ + if callable(getattr(reporter, "emit", None)): + reporter.emit(event) # type: ignore[attr-defined] + return + if callable(reporter): + reporter(event) # type: ignore[call-arg] + return + raise TypeError( + f"reporter {type(reporter).__name__} has neither emit() nor is callable" + ) + + +class FanOut: + """Dispatches each event to every configured reporter. + + Accepts sinks as objects with ``emit(event)`` **or** plain functions + (found live 2026-10-08: the console function sink raised AttributeError + on every event — the fan-out duck-types both shapes now). One reporter + failing must never kill the others or the poll loop: each delivery is + isolated, failures counted and surfaced via the injected ``error`` + callback (stderr in the CLI). Sinks own their concurrency safety (HTTP + queues internally; file reporters append single small lines). + """ + + def __init__(self, reporters: tuple[Reporter, ...], error: Callable[[str], None]) -> None: + self._reporters = reporters + self._error = error + self.failures = 0 + + def emit(self, event: DiagEvent) -> None: + for reporter in self._reporters: + try: + _emit_via(reporter, event) + except Exception as exc: # noqa: BLE001 — a sink must never kill polling + self.failures += 1 + self._error( + f"reporter {type(reporter).__name__} failed " + f"({type(exc).__name__}: {exc}); event {event.kind.value} " + "kept in the other sinks" + ) + + def close(self) -> None: + """Best-effort close of every sink (shutdown path only).""" + for reporter in self._reporters: + close = getattr(reporter, "close", None) + if close is None: + continue + try: + close() + except Exception: # noqa: BLE001 + pass \ No newline at end of file diff --git a/src/opcda_diagnozer/report/eventlog.py b/src/opcda_diagnozer/report/eventlog.py index 311e7fc..cfca2c6 100644 --- a/src/opcda_diagnozer/report/eventlog.py +++ b/src/opcda_diagnozer/report/eventlog.py @@ -1,22 +1,85 @@ -"""Windows Event Log reporter, WARN+ only (IMPLEMENTATION_PLAN §5.2). +"""Windows Event Log reporter, severity-mapped, WARN+ by default (plan §5.2). -Uses ``win32evtlog``; the source is registered during ``install``. pywin32 is -import-guarded so tests can run on any OS (IMPLEMENTATION_PLAN §9). +Uses ``win32evtlog`` via a registered **event source handle** +(``RegisterEventSource`` → positional ``ReportEvent`` — the pywin32 function +takes positional args only; keyword calls raise ``TypeError``). COM imports +are guarded so tests and non-Windows runs import the module freely +(IMPLEMENTATION_PLAN §9). + +The source's message file is registered during ``install`` (Phase 3); until +then events still land in the Application log under the source name (verified +live: Windows delivers them, rendering the message with an empty template). +A sink failure is counted, never raised — the poll loop must not stall. """ from __future__ import annotations from opcda_diagnozer.events import DiagEvent, Severity +# Windows Event Log types (win32evtlog EVENTLOG_*_TYPE constants) +_LEVELS = { + Severity.INFO: 4, # EVENTLOG_INFORMATION_TYPE + Severity.WARN: 2, # EVENTLOG_WARNING_TYPE + Severity.CRITICAL: 1, # EVENTLOG_ERROR_TYPE +} +_ORDER = {Severity.INFO: 0, Severity.WARN: 1, Severity.CRITICAL: 2} + class EventLogReporter: - """Windows Event Log sink — implemented in Phase 2.""" + """Windows Event Log sink — WARN+ only (spec §9), severity-mapped.""" def __init__(self, source: str = "OPCDA-Diagnozer", min_severity: str = "WARN") -> None: self._source = source - self._min_severity = Severity(min_severity) + self._min = Severity(min_severity) + self._handle = None + self._register_source() + + def _register_source(self) -> None: + """Open the event source handle once (kept for the reporter's lifetime). + + ``RegisterEventSource(None, source)`` works without admin rights; + unregistered sources still deliver (verified live 2026-10-08). + """ + try: + import win32evtlog + + self._handle = win32evtlog.RegisterEventSource(None, self._source) + except Exception: # noqa: BLE001 — non-Windows/tests: emit() is a no-op + self._handle = None def emit(self, event: DiagEvent) -> None: - raise NotImplementedError( - "Phase 2: Event Log reporter (docs/IMPLEMENTATION_PLAN.md §5.2)" - ) \ No newline at end of file + """Report one event; below-minimum severities are filtered out.""" + if _ORDER[event.severity] < _ORDER[self._min]: + return + if self._handle is None: + self._register_source() + if self._handle is None: + return + message = ", ".join(f"{k}={v}" for k, v in event.detail.items()) + try: + import win32evtlog + + # positional args only (win32evtlog.ReportEvent takes no keywords) + win32evtlog.ReportEvent( + self._handle, + _LEVELS[event.severity], + 0, # category + 0, # event ID + None, # user SID + [f"{event.kind.value}: {message}"], + None, # raw data: the JSONL file already carries the full record + ) + except Exception: # noqa: BLE001 — a sink must never kill the poll loop + pass + + def close(self) -> None: + """Release the event source handle (Windows API hygiene).""" + if self._handle is None: + return + try: + import win32evtlog + + win32evtlog.DeregisterEventSource(self._handle) + except Exception: # noqa: BLE001 — shutdown is best-effort + pass + self._handle = None \ No newline at end of file diff --git a/src/opcda_diagnozer/report/http.py b/src/opcda_diagnozer/report/http.py index acd11d2..e84d926 100644 --- a/src/opcda_diagnozer/report/http.py +++ b/src/opcda_diagnozer/report/http.py @@ -1,23 +1,111 @@ """Optional HTTP webhook reporter (IMPLEMENTATION_PLAN §5.3). -Phase 2: POST one JSON event per request, optional bearer token, timeout plus -simple retry/drop on failure — must never block the poll loop (bounded queue). +POSTs one JSON event per request with an optional bearer token, timeout + +drop-on-failure — **never blocks the poll loop**: ``emit`` only pushes onto a +bounded queue and a daemon thread drains it. A full queue drops the event +(and counts it): losing an alert is always better than stalling polling on a +dead webhook host. """ from __future__ import annotations +import json +import queue +import threading +from collections.abc import Callable + from opcda_diagnozer.events import DiagEvent +_QUEUE_BOUND = 1000 # events; ~17 min of anomalies at a pathological rate + class HttpReporter: - """Webhook sink — implemented in Phase 2.""" + """Webhook sink — bounded queue + daemon sender thread (spec §9).""" - def __init__(self, url: str, bearer_token: str | None = None, timeout_s: float = 5.0) -> None: + def __init__( + self, + url: str, + bearer_token: str | None = None, + timeout_s: float = 5.0, + post: Callable[[str, str | None, str, float], None] | None = None, + ) -> None: self._url = url - self._bearer_token = bearer_token + self._bearer = bearer_token self._timeout_s = timeout_s + self._post = post # injectable for tests; None -> urllib in the thread + self._queue: queue.Queue[DiagEvent | None] = queue.Queue(maxsize=_QUEUE_BOUND) + self._dropped = 0 + self._sent = 0 + self._stop = threading.Event() + self._thread: threading.Thread | None = None + if self._post is not None: + # direct mode (tests): no thread, emit posts synchronously + pass + else: + self._thread = threading.Thread(target=self._drain, daemon=True) + self._thread.start() def emit(self, event: DiagEvent) -> None: - raise NotImplementedError( - "Phase 2: HTTP webhook reporter (docs/IMPLEMENTATION_PLAN.md §5.3)" - ) \ No newline at end of file + """Queue the event; drop (counted) when the bound is reached. + + In injected-``post`` mode (tests) the send happens inline but is + equally isolated: a failing webhook must never raise into the poll + loop — losing one alert beats stalling polling on a dead host. + """ + if self._post is not None: + try: + self._post(self._url, self._bearer, event.to_json(), self._timeout_s) + self._sent += 1 + except Exception: # noqa: BLE001 — drop on failure, keep polling + self._dropped += 1 + return + try: + self._queue.put_nowait(event) + except queue.Full: + self._dropped += 1 + + def _drain(self) -> None: + import urllib.request + + while not self._stop.is_set(): + try: + event = self._queue.get(timeout=0.25) + except queue.Empty: + continue + if event is None: + break + request = urllib.request.Request( + self._url, + data=event.to_json().encode("utf-8"), + headers={ + "Content-Type": "application/json", + **({"Authorization": f"Bearer {self._bearer}"} if self._bearer else {}), + }, + method="POST", + ) + try: + with urllib.request.urlopen(request, timeout=self._timeout_s): + pass + self._sent += 1 + except Exception: # noqa: BLE001 — drop on failure, keep polling + self._dropped += 1 + + @property + def dropped(self) -> int: + """Events dropped because the queue was full or the POST failed.""" + return self._dropped + + @property + def sent(self) -> int: + return self._sent + + def close(self, timeout_s: float = 2.0) -> None: + """Stop the sender thread; queued events are flushed if the drain is fast.""" + if self._thread is None: + return + self._stop.set() + try: + self._queue.put_nowait(None) # wake the drain immediately + except queue.Full: + pass + self._thread.join(timeout=timeout_s) \ No newline at end of file diff --git a/src/opcda_diagnozer/report/jsonl.py b/src/opcda_diagnozer/report/jsonl.py index 9a93b5b..6356884 100644 --- a/src/opcda_diagnozer/report/jsonl.py +++ b/src/opcda_diagnozer/report/jsonl.py @@ -1,25 +1,77 @@ """Rotating JSONL reporter — the primary evidence trail (IMPLEMENTATION_PLAN §5.1). -Phase 2 adds: size-based rotation with ``backups`` kept, atomic-ish append, -and a header line carrying the config digest for provenance. +One JSON event per line (spec §8), size-based rotation with ``backups`` kept, +and a header line per file carrying the config digest for provenance — so a +rotated file can always be traced to the exact thresholds that produced it. + +Appends are "atomic-ish": one pre-serialized line + newline, flushed. A hard +kill can at worst lose the current line, never corrupt earlier ones (event +lines are a few hundred bytes — single small appends, not torn writes). """ from __future__ import annotations +import hashlib +import json from pathlib import Path +from opcda_diagnozer.config import Config from opcda_diagnozer.events import DiagEvent -class JsonlReporter: - """One JSON event per line, rotated by size — implemented in Phase 2.""" +def config_digest(config: Config) -> str: + """Stable SHA-256 over the effective config (provenance for log headers).""" + payload = json.dumps(config, default=lambda o: getattr(o, "__dict__", str(o)), + sort_keys=True) + return hashlib.sha256(payload.encode("utf-8")).hexdigest() - def __init__(self, directory: str | Path, max_mb: int = 50, backups: int = 10) -> None: + +def rotate_file(path: Path, backups: int) -> None: + """Size-rotation used by both file reporters: ``p`` -> ``p.1`` … ``p.backups``. + + The oldest copy is dropped; numbering shifts by one so ``p.1`` is always + the most recently closed file. Shared so the JSONL and text reporters + guarantee identical retention semantics (spec §9.1). + """ + if not path.exists(): + return + for index in range(backups - 1, 0, -1): + older = path.with_name(f"{path.name}.{index}") + if older.exists(): + older.replace(path.with_name(f"{path.name}.{index + 1}")) + path.replace(path.with_name(f"{path.name}.1")) + + +class JsonlReporter: + """One JSON event per line, rotated by size (spec §9).""" + + def __init__(self, directory: str | Path, max_mb: int = 50, backups: int = 10, + digest: str = "") -> None: self._directory = Path(directory) - self._max_mb = max_mb + self._max_bytes = max_mb * 1024 * 1024 self._backups = backups + self._digest = digest + self._directory.mkdir(parents=True, exist_ok=True) + self._current = self._directory / "events.jsonl" def emit(self, event: DiagEvent) -> None: - raise NotImplementedError( - "Phase 2: JSONL reporter (docs/IMPLEMENTATION_PLAN.md §5.1)" - ) \ No newline at end of file + """Append one event; rotate first when the file would exceed the cap.""" + line = event.to_json() + needs_header = not self._current.exists() + if self._current.exists() and ( + self._current.stat().st_size + len(line) + 2 > self._max_bytes + ): + rotate_file(self._current, self._backups) + needs_header = True + with self._current.open("a", encoding="utf-8") as stream: + if needs_header: + header = { + "header": True, + "format": "opcda-diagnozer events v1", + "config_sha256": self._digest, + } + stream.write(json.dumps(header) + "\n") + stream.write(line + "\n") + + def close(self) -> None: + """Lifecycle symmetry with the other reporters (nothing to release).""" \ No newline at end of file diff --git a/src/opcda_diagnozer/report/textlog.py b/src/opcda_diagnozer/report/textlog.py new file mode 100644 index 0000000..ced56b1 --- /dev/null +++ b/src/opcda_diagnozer/report/textlog.py @@ -0,0 +1,120 @@ +"""Plain-text anomaly log — the human-readable incident narrative (spec §9.1). + +``anomalies-YYYYMMDD.log`` in the same directory as the JSONL files: one line +per **detected anomaly** (``is_anomaly`` filter — heartbeats and CONNECTED are +lifecycle noise), fixed field order ``timestamp SEVERITY KIND message``, +clear-text status with the code in parentheses per spec §7. Recoveries +(D2 RECONNECTED, DEVICE_LINK_UP) *are* logged so the file reads as a continuous +narrative: what broke, when it came back. + +Every line corresponds to a JSONL record (JSONL = machine source of truth, +this = convenience view). Size rotation uses the same scheme as the JSONL +reporter, per-day file naming keeps `tail -f` sessions stable. +""" + +from __future__ import annotations + +from datetime import datetime, timezone +from pathlib import Path + +from opcda_diagnozer.events import DiagEvent, EventKind, Severity +from opcda_diagnozer.report.jsonl import rotate_file + +_MAX_LINE_BYTES = 1024 * 1024 # safety cap: a corrupted detail dict can't grow the file + + +def is_anomaly(kind: EventKind) -> bool: + """Kinds shown in the text log (spec §9.1): everything except heartbeat/connected.""" + return kind not in (EventKind.HEARTBEAT, EventKind.CONNECTED) + + +def message_for(event: DiagEvent) -> str: + """Human message per event kind — clear text, codes in parentheses (spec §7).""" + d = event.detail + kind = event.kind + if kind is EventKind.DISCONNECTED: + text = str(d.get("reason", "connection lost")) + if "shutting down" in text: + return f"stopping: {text}" + return f"connection lost: {text}" + if kind is EventKind.RECONNECTED: + return f"reconnected after {d.get('outage_s', '?')} s outage" + if kind is EventKind.RECONNECT_FLAPPING: + return f"{d.get('disconnects_in_window', '?')} disconnects within 10 min" + if kind in (EventKind.SLOW_POLL, EventKind.SLOW_POLL_STREAK): + streak = f", streak {d['streak']}" if "streak" in d else "" + return f"read took {d.get('rtt_ms', '?')} ms (warn {d.get('warn_ms', '?')} ms){streak}" + if kind is EventKind.LATENCY_TREND: + return f"p95 read latency {d.get('rtt_p95_ms', '?')} ms over the last window" + if kind is EventKind.POLL_HANG: + if "abandoned" in str(d.get("action", "")): + return f"read call hung > {d.get('hard_hang_s', '?')} s; COM thread abandoned and rebuilt" + return f"read took {d.get('rtt_ms', '?')} ms (timeout {d.get('timeout_ms', '?')} ms); thread kept" + if kind in (EventKind.QUALITY_BAD, EventKind.QUALITY_UNCERTAIN): + arrow = f" changed {d.get('previous_quality', '?')} -> {d['quality']}" if "quality" in d else "" + return f"tag '{d.get('tag', '?')}'{arrow}" + if kind is EventKind.QUALITY_BAD_PERSISTENT: + return f"tag '{d.get('tag', '?')}' BAD for {d.get('bad_for_s', '?')} s (threshold {d.get('threshold_s', '?')} s)" + if kind is EventKind.QUALITY_FLAPPING: + return f"tag '{d.get('tag', '?')}' quality changed {d.get('transitions_in_window', '?')} times in {d.get('window_s', '?')} s" + if kind is EventKind.STALE_TIMESTAMP: + return (f"tag '{d.get('tag', '?')}' device time {d.get('age_s', '?')} s old " + f"(max {d.get('max_age_s', '?')} s) while quality GOOD") + if kind is EventKind.FROZEN_VALUE: + return f"tag '{d.get('tag', '?')}' value {d.get('value')!r} unchanged for {d.get('unchanged_for_s', '?')} s" + if kind is EventKind.CONTROL_LINK_DEGRADED: + return f"{d.get('tags_bad', '?')} of {d.get('tags_total', '?')} monitored tags BAD while connection OK" + if kind is EventKind.DEVICE_LINK_DOWN: + parts = [f"device '{d.get('device', '?')}': link indicator '{d.get('indicator_tag', '?')}'"] + if d.get("quality") and "quality" in str(d.get("reason", "")): + parts.append(str(d.get("reason"))) + elif d.get("reason"): + parts.append(f"read {d.get('value')!r}: {d['reason']}") + else: + parts.append("down") + return " ".join(parts) + if kind is EventKind.DEVICE_LINK_UP: + return f"device '{d.get('device', '?')}': link indicator '{d.get('indicator_tag', '?')}' back to normal" + # unknown kinds: generic but complete rendering + return " ".join(f"{k}={v}" for k, v in d.items()) or str(kind.value) + + +class TextLogReporter: + """Plain-text anomaly log (spec §9.1) — same rotation scheme as JSONL.""" + + def __init__(self, directory: str | Path, max_mb: int = 50, backups: int = 10) -> None: + self._directory = Path(directory) + self._max_bytes = max_mb * 1024 * 1024 + self._backups = backups + self._directory.mkdir(parents=True, exist_ok=True) + self._day: str | None = None + self._current: Path | None = None + + def _file_for(self, at: datetime) -> Path: + """Per-day file: anomalies-YYYYMMDD.log (stable for `tail -f`).""" + day = at.astimezone(timezone.utc).strftime("%Y%m%d") + if self._current is None or day != self._day: + self._day = day + self._current = self._directory / f"anomalies-{day}.log" + return self._current + + def emit(self, event: DiagEvent) -> None: + if not is_anomaly(event.kind): + return + line = self._format(event) + path = self._file_for(event.time_utc) + if path.exists() and path.stat().st_size + len(line) + 2 > self._max_bytes: + rotate_file(path, self._backups) + with path.open("a", encoding="utf-8") as stream: + stream.write(line + "\n") + + def _format(self, event: DiagEvent) -> str: + stamp = event.time_utc.astimezone(timezone.utc) + ts = stamp.strftime("%Y-%m-%d %H:%M:%S.") + f"{stamp.microsecond // 1000:03d} UTC" + message = message_for(event) + if len(message) > _MAX_LINE_BYTES: + message = message[:_MAX_LINE_BYTES] # safety: never explode on bad detail + return f"{ts} {event.severity.value:<8} {event.kind.value:<20} {message}" + + def close(self) -> None: + """Lifecycle symmetry (file handles are opened per append).""" \ No newline at end of file diff --git a/tests/test_reporters.py b/tests/test_reporters.py new file mode 100644 index 0000000..2cdd83d --- /dev/null +++ b/tests/test_reporters.py @@ -0,0 +1,367 @@ +"""Reporter tests (Phase 2): JSONL rotation, text anomaly log, Event Log filter, +HTTP drop policy, fan-out isolation — all cross-OS (COM paths guarded). + +Live Event Log reporting is Windows-only and smoke-tested on dev boxes; here +the severity filter logic is what's under test. +""" + +import json +import time +from datetime import datetime, timezone +from pathlib import Path + +import pytest + +from opcda_diagnozer.config import Config, DeviceEntry, OpcConfig, ReportConfig, TagEntry +from opcda_diagnozer.events import DiagEvent, EventKind, Severity, utc_now +from opcda_diagnozer.report.base import FanOut +from opcda_diagnozer.report.http import HttpReporter +from opcda_diagnozer.report.jsonl import JsonlReporter, config_digest, rotate_file +from opcda_diagnozer.report.textlog import TextLogReporter, is_anomaly, message_for + + +def _event(kind: EventKind, severity: Severity = Severity.WARN, **detail) -> DiagEvent: + return DiagEvent(time_utc=utc_now(), kind=kind, severity=severity, detail=dict(detail)) + + +# --------------------------------------------------------------------------- # +# JSONL reporter +# --------------------------------------------------------------------------- # + + +def test_jsonl_writes_one_line_per_event_with_header(tmp_path: Path) -> None: + reporter = JsonlReporter(tmp_path, digest="abc123") + reporter.emit(_event(EventKind.CONNECTED, Severity.INFO, server="X")) + reporter.emit(_event(EventKind.QUALITY_BAD, tag="T", quality="BAD / COMM FAILURE (0x18)")) + lines = (tmp_path / "events.jsonl").read_text(encoding="utf-8").splitlines() + header = json.loads(lines[0]) + assert header["header"] is True + assert header["config_sha256"] == "abc123" + first, second = (json.loads(line) for line in lines[1:]) + assert first["kind"] == "CONNECTED" + assert second["detail"]["tag"] == "T" + assert second["detail"]["quality"] == "BAD / COMM FAILURE (0x18)" + + +def test_jsonl_rotates_on_size(tmp_path: Path) -> None: + reporter = JsonlReporter(tmp_path, max_mb=1, backups=3) # 1 MiB cap + big = "x" * 2048 # force rotation with many events + for i in range(600): # ~600 * ~2.2 KiB > 1 MiB + reporter.emit(_event(EventKind.HEARTBEAT, Severity.INFO, i=i, payload=big)) + rotated = sorted(p.name for p in tmp_path.glob("events.jsonl*")) + assert "events.jsonl" in rotated + assert "events.jsonl.1" in rotated # rotated at least once + # current file still valid JSON lines with a fresh header + first_line = (tmp_path / "events.jsonl").read_text(encoding="utf-8").splitlines()[0] + assert json.loads(first_line)["header"] is True + + +def test_rotate_file_shifts_numbering(tmp_path: Path) -> None: + p = tmp_path / "f.log" + for i in range(1, 5): + (tmp_path / f"f.log.{i}").write_text(f"old{i}", encoding="utf-8") + p.write_text("current", encoding="utf-8") + rotate_file(p, backups=4) + assert (tmp_path / "f.log.1").read_text(encoding="utf-8") == "current" + assert (tmp_path / "f.log.2").read_text(encoding="utf-8") == "old1" + assert (tmp_path / "f.log.4").read_text(encoding="utf-8") == "old3" + assert not (tmp_path / "f.log.5").exists() # backups=4 respected + assert not p.exists() + + +def test_config_digest_is_stable_and_sensitive(tmp_path: Path) -> None: + base = Config(opc=OpcConfig(server="X"), tags=(TagEntry(name="t"),)) + same = Config(opc=OpcConfig(server="X"), tags=(TagEntry(name="t"),)) + other = Config(opc=OpcConfig(server="Y"), tags=(TagEntry(name="t"),)) + assert config_digest(base) == config_digest(same) + assert config_digest(base) != config_digest(other) + + +# --------------------------------------------------------------------------- # +# Text anomaly log +# --------------------------------------------------------------------------- # + + +def test_is_anomaly_excludes_lifecycle_kinds() -> None: + assert is_anomaly(EventKind.HEARTBEAT) is False + assert is_anomaly(EventKind.CONNECTED) is False + assert is_anomaly(EventKind.QUALITY_BAD) is True + assert is_anomaly(EventKind.DEVICE_LINK_UP) is True # recovery is part of the narrative + + +def test_textlog_filters_and_formats_spec_9_1(tmp_path: Path) -> None: + reporter = TextLogReporter(tmp_path) + reporter.emit(_event(EventKind.HEARTBEAT, Severity.INFO)) # filtered out + reporter.emit(_event(EventKind.CONNECTED, Severity.INFO)) # filtered out + reporter.emit( + _event( + EventKind.QUALITY_BAD, + Severity.WARN, + tag="Channel1.Device1.Counter", + previous_quality="GOOD", + quality="BAD / COMM FAILURE (0x18)", + ) + ) + files = list(tmp_path.glob("anomalies-*.log")) + assert len(files) == 1 + line = files[0].read_text(encoding="utf-8").strip() + # fixed field order: timestamp SEVERITY KIND message + assert " UTC " in line + assert "WARN" in line and "QUALITY_BAD" in line + assert "tag 'Channel1.Device1.Counter'" in line + assert "changed GOOD -> BAD / COMM FAILURE (0x18)" in line + + +def test_textlog_message_per_kind() -> None: + assert "connection lost" in message_for( + _event(EventKind.DISCONNECTED, Severity.CRITICAL, reason="RPC failed") + ) + assert "reconnected after 35.0 s" in message_for( + _event(EventKind.RECONNECTED, Severity.INFO, outage_s=35.0) + ) + assert "12 of 14" in message_for( + _event(EventKind.CONTROL_LINK_DEGRADED, tags_bad=12, tags_total=14) + ) + msg = message_for( + _event( + EventKind.DEVICE_LINK_DOWN, + device="Conveyor 3", + indicator_tag="C1.D3.Connected", + value=False, + quality="GOOD (0xC0)", + reason="read 'false', expected one of ['true']", + ) + ) + assert "device 'Conveyor 3'" in msg + assert "C1.D3.Connected" in msg + hang = message_for( + _event(EventKind.POLL_HANG, hard_hang_s=120, action="COM thread abandoned and rebuilt") + ) + assert "abandoned" in hang + + +def test_textlog_daily_files(tmp_path: Path) -> None: + reporter = TextLogReporter(tmp_path) + at = datetime(2026, 10, 7, 23, 59, 59, tzinfo=timezone.utc) + reporter.emit( + DiagEvent(time_utc=at, kind=EventKind.QUALITY_BAD, severity=Severity.WARN, detail={}) + ) + at2 = datetime(2026, 10, 8, 0, 0, 1, tzinfo=timezone.utc) + reporter.emit( + DiagEvent(time_utc=at2, kind=EventKind.QUALITY_BAD, severity=Severity.WARN, detail={}) + ) + days = sorted(p.name for p in tmp_path.glob("anomalies-*.log")) + assert days == ["anomalies-20261007.log", "anomalies-20261008.log"] + + +# --------------------------------------------------------------------------- # +# HTTP webhook +# --------------------------------------------------------------------------- # + + +def test_http_reporter_posts_json_with_bearer() -> None: + posted: list[tuple[str, str | None, str]] = [] + + def fake_post(url, bearer, body, timeout_s): + posted.append((url, bearer, body)) + + reporter = HttpReporter( + "https://alerts.example.com/hook", + bearer_token="secret", + post=fake_post, + ) + reporter.emit(_event(EventKind.QUALITY_BAD, tag="T")) + url, bearer, body = posted[0] + assert url == "https://alerts.example.com/hook" + assert bearer == "secret" + assert json.loads(body)["kind"] == "QUALITY_BAD" + + +def test_http_reporter_drops_instead_of_blocking() -> None: + """A failing POST is counted and dropped — never raises to the poller.""" + + def failing_post(url, bearer, body, timeout_s): + raise ConnectionError("host down") + + reporter = HttpReporter("https://x.example.com", post=failing_post) + reporter.emit(_event(EventKind.QUALITY_BAD)) # must not raise + assert reporter.dropped == 1 + assert reporter.sent == 0 + + +# --------------------------------------------------------------------------- # +# Fan-out isolation +# --------------------------------------------------------------------------- # + + +class _Exploding: + def emit(self, event: DiagEvent) -> None: + raise RuntimeError("sink exploded") + + +class _Recording: + def __init__(self) -> None: + self.events: list[DiagEvent] = [] + + def emit(self, event: DiagEvent) -> None: + self.events.append(event) + + +def test_fanout_isolates_sink_failures() -> None: + errors: list[str] = [] + good = _Recording() + fan = FanOut((_Exploding(), good), error=errors.append) + fan.emit(_event(EventKind.QUALITY_BAD, tag="T")) + assert len(good.events) == 1 # the good sink still got it + assert fan.failures == 1 + assert "sink exploded" in errors[0] + + +def test_fanout_accepts_function_sinks() -> None: + """Regression (found live 2026-10-08): the console sink is a plain function. + + ``_build_reporters`` builds the console sink as a function; FanOut called + ``.emit()`` on every sink, so the console failed on EVERY event with + AttributeError while the object sinks kept working. The fan-out must + deliver to both shapes. + """ + received: list[EventKind] = [] + + def console_like(event: DiagEvent) -> None: + received.append(event.kind) + + good = _Recording() + fan = FanOut((console_like, good), error=lambda m: None) + fan.emit(_event(EventKind.QUALITY_BAD, tag="T")) + assert received == [EventKind.QUALITY_BAD] + assert len(good.events) == 1 + assert fan.failures == 0 + + +def test_fanout_rejects_unusable_sinks_loudly() -> None: + """A sink that is neither emits nor callable counts as a failure, not silence.""" + errors: list[str] = [] + fan = FanOut((object(),), error=errors.append) + fan.emit(_event(EventKind.QUALITY_BAD)) + assert fan.failures == 1 + assert "neither emit()" in errors[0] + + +def test_build_reporters_wired_end_to_end(tmp_path: Path, capsys) -> None: + """Integration: the ACTUAL assembly the run verb uses, console included. + + The piecewise tests (FanOut alone, reporters alone) missed that the wired + fan-out broke the function-shaped console sink. This drives the real + `_build_reporters` and asserts an event lands in the console AND the + JSONL file — the wiring this bug hid. + """ + from opcda_diagnozer.config import JsonlConfig, ReportConfig + from opcda_diagnozer.main import _build_reporters + + logs = tmp_path / "logs" + config = Config( + opc=OpcConfig(server="Fake.Server.1"), + tags=(TagEntry(name="T"),), + report=ReportConfig( + jsonl=JsonlConfig(dir=str(logs), max_mb=1, backups=1), + textlog=JsonlConfig(dir=str(logs), max_mb=1, backups=1), + ), + ) + fan_out = _build_reporters(config) + fan_out.emit(_event(EventKind.CONNECTED, Severity.INFO, server="Fake.Server.1")) + fan_out.emit(_event(EventKind.QUALITY_BAD, Severity.WARN, tag="T", + quality="BAD / COMM FAILURE (0x18)")) + fan_out.close() + + console = capsys.readouterr().out + assert "CONNECTED" in console # the function sink delivered + assert "QUALITY_BAD" in console + assert "BAD / COMM FAILURE (0x18)" in console + + lines = (logs / "events.jsonl").read_text(encoding="utf-8").splitlines() + payload = [json.loads(line) for line in lines if not json.loads(line).get("header")] + kinds = [p["kind"] for p in payload] + assert kinds == ["CONNECTED", "QUALITY_BAD"] # JSONL got both too + + anomalies = (logs / f"anomalies-{time.strftime('%Y%m%d')}.log").read_text(encoding="utf-8") + assert "QUALITY_BAD" in anomalies + assert "CONNECTED" not in anomalies # lifecycle event filtered from text log + + +def test_fanout_close_is_best_effort() -> None: + class _BadClose: + def emit(self, event: DiagEvent) -> None: ... + + def close(self) -> None: + raise RuntimeError("close failed") + + closed: list[bool] = [] + + class _Good: + def emit(self, event: DiagEvent) -> None: ... + + def close(self) -> None: + closed.append(True) + + fan = FanOut((_BadClose(), _Good()), error=lambda m: None) + fan.close() # must not raise despite _BadClose + assert closed == [True] + + +# --------------------------------------------------------------------------- # +# Windows Event Log (severity filter — the COM call itself is live-verified) +# --------------------------------------------------------------------------- # + + +def test_eventlog_severity_filter() -> None: + """INFO events are filtered below a WARN minimum; WARN/CRITICAL pass. + + On non-Windows (or without pywin32) the handle stays None and emit is a + safe no-op — the filter decision is still exercised before that point. + """ + from opcda_diagnozer.report.eventlog import EventLogReporter + + reporter = EventLogReporter(source="TestSource", min_severity="WARN") + # filter check happens before any COM work: an INFO event returns early + reporter.emit(_event(EventKind.HEARTBEAT, Severity.INFO)) + reporter.emit(_event(EventKind.QUALITY_BAD, Severity.WARN)) + reporter.emit(_event(EventKind.DISCONNECTED, Severity.CRITICAL)) + reporter.close() + + +# --------------------------------------------------------------------------- # +# Config wiring +# --------------------------------------------------------------------------- # + + +def test_report_config_http_optional_and_validated(tmp_path: Path) -> None: + from opcda_diagnozer.config import ConfigError, HttpConfig, load_config + + cfg_text = ( + "opc:\n server: X\ntags:\n - name: t\n" + "report:\n http:\n url: \"https://alerts.example.com/hook\"\n" + ) + p = tmp_path / "cfg.yaml" + p.write_text(cfg_text, encoding="utf-8") + config = load_config(p) + assert config.report.http is not None + assert config.report.http.url == "https://alerts.example.com/hook" + + p.write_text( + "opc:\n server: X\ntags:\n - name: t\n" + "report:\n http:\n url: \"ftp://bad.example.com\"\n", + encoding="utf-8", + ) + with pytest.raises(ConfigError, match="http:// or https://"): + load_config(p) + + +def test_report_config_textlog_defaults_match_jsonl(tmp_path: Path) -> None: + from opcda_diagnozer.config import load_config + + p = tmp_path / "cfg.yaml" + p.write_text("opc:\n server: X\ntags:\n - name: t\n", encoding="utf-8") + config = load_config(p) + # both file sinks default to the same directory (spec §9.1: same dir) + assert config.report.jsonl.dir == config.report.textlog.dir + assert config.report.http is None # webhook off by default \ No newline at end of file