"""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