367 lines
14 KiB
Python
367 lines
14 KiB
Python
"""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 |