f2ba81bcc1
An "unknown"/empty node_name silently produced evt-unknown-* event files. One landed in events/piha/ and, being lexically greater than every evt-piha-* name, pinned the observer's (then lexical) checkpoint and blocked PIHA for ~34 days. Defense in depth, two layers, both refuse rather than emit poison: - config.py: node_name field_validator rejects ""/"unknown" (case-insensitive, trimmed); validate_default=True so the "unknown" default itself is rejected when NODE_NAME never reaches the process. main() catches ValidationError and exits 1 with a clear FATAL message instead of a raw traceback. - event_emitter.py: EventEmitter.__init__ guards node_name at the exact poison site (node_name is embedded in the filename), so no future call path can reintroduce evt-unknown-*. Precedence unchanged and correct: Settings.load() uses os.environ.setdefault, so env NODE_NAME wins over YAML; YAML only supplies node_name when env is absent. api.py never writes event files (only /health, /trigger), so its _node_name global cannot produce poison — verified, left as-is. Tests (services/ha-diag-agent/tests/): new test_config.py (env load, env>YAML precedence, fail-fast on unknown/empty/whitespace/default/YAML-unknown) and EventEmitter guard tests. Full unit suite: 139 passed, 0 regressions. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
104 lines
3.8 KiB
Python
104 lines
3.8 KiB
Python
"""Tests for EventEmitter."""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from ha_diag.event_emitter import EventEmitter
|
|
|
|
|
|
def test_emit_creates_json_file(tmp_events_dir: Path, emitter: EventEmitter):
|
|
event_id = emitter.emit(
|
|
event_type="ha_websocket_dead",
|
|
severity="error",
|
|
service="homeassistant",
|
|
message="HA unreachable",
|
|
payload={"error": "timeout"},
|
|
)
|
|
files = list(tmp_events_dir.glob("*.json"))
|
|
assert len(files) == 1
|
|
assert files[0].name == f"{event_id}.json"
|
|
|
|
|
|
def test_emit_event_schema(tmp_events_dir: Path, emitter: EventEmitter):
|
|
event_id = emitter.emit(
|
|
event_type="ha_websocket_dead",
|
|
severity="error",
|
|
service="homeassistant",
|
|
message="HA unreachable",
|
|
payload={"error": "timeout"},
|
|
)
|
|
data = json.loads((tmp_events_dir / f"{event_id}.json").read_text())
|
|
assert data["id"] == event_id
|
|
assert data["type"] == "ha_websocket_dead"
|
|
assert data["severity"] == "error"
|
|
assert data["node"] == "test-node"
|
|
assert data["service"] == "homeassistant"
|
|
assert data["message"] == "HA unreachable"
|
|
assert data["payload"] == {"error": "timeout"}
|
|
assert "timestamp" in data
|
|
assert "date" in data
|
|
|
|
|
|
def test_emit_multiple_events_unique_files(tmp_events_dir: Path, emitter: EventEmitter):
|
|
ids = [
|
|
emitter.emit("ha_websocket_dead", "error", "homeassistant", f"msg {i}")
|
|
for i in range(3)
|
|
]
|
|
assert len(set(ids)) == 3
|
|
assert len(list(tmp_events_dir.glob("*.json"))) == 3
|
|
|
|
|
|
def test_emit_no_tmp_file_left(tmp_events_dir: Path, emitter: EventEmitter):
|
|
emitter.emit("ha_websocket_dead", "error", "homeassistant", "msg")
|
|
assert not list(tmp_events_dir.glob("*.tmp"))
|
|
|
|
|
|
def test_emitter_creates_events_dir(tmp_path: Path):
|
|
new_dir = tmp_path / "nested" / "events"
|
|
emitter = EventEmitter(new_dir, "my-node")
|
|
assert new_dir.exists()
|
|
|
|
|
|
def test_location_tag_included_in_payload(tmp_events_dir: Path):
|
|
emitter = EventEmitter(tmp_events_dir, node_name="piha", location_tag="ken")
|
|
event_id = emitter.emit("ha_websocket_dead", "error", "homeassistant", "msg")
|
|
data = json.loads((tmp_events_dir / f"{event_id}.json").read_text())
|
|
assert data["payload"]["location_tag"] == "ken"
|
|
|
|
|
|
def test_location_tag_empty_not_in_payload(tmp_events_dir: Path):
|
|
emitter = EventEmitter(tmp_events_dir, node_name="piha", location_tag="")
|
|
event_id = emitter.emit("ha_websocket_dead", "error", "homeassistant", "msg")
|
|
data = json.loads((tmp_events_dir / f"{event_id}.json").read_text())
|
|
assert "location_tag" not in data["payload"]
|
|
|
|
|
|
@pytest.mark.parametrize("bad_node", ["unknown", "UNKNOWN", "", " "])
|
|
def test_emitter_refuses_unknown_or_empty_node(tmp_events_dir: Path, bad_node: str):
|
|
# No emit path may ever produce evt-unknown-* poison files (see d5139c9).
|
|
with pytest.raises(ValueError):
|
|
EventEmitter(tmp_events_dir, node_name=bad_node)
|
|
|
|
|
|
def test_emitter_never_writes_unknown_node_in_filename(tmp_events_dir: Path):
|
|
emitter = EventEmitter(tmp_events_dir, node_name="piha")
|
|
event_id = emitter.emit("ha_websocket_dead", "error", "homeassistant", "msg")
|
|
assert "unknown" not in event_id
|
|
data = json.loads((tmp_events_dir / f"{event_id}.json").read_text())
|
|
assert data["node"] == "piha"
|
|
|
|
|
|
def test_location_tag_does_not_override_explicit_payload_key(tmp_events_dir: Path):
|
|
emitter = EventEmitter(tmp_events_dir, node_name="piha", location_tag="ken")
|
|
event_id = emitter.emit(
|
|
"ha_websocket_dead", "error", "homeassistant", "msg",
|
|
payload={"location_tag": "override", "other": "value"},
|
|
)
|
|
data = json.loads((tmp_events_dir / f"{event_id}.json").read_text())
|
|
# Explicit payload key wins over the emitter's location_tag
|
|
assert data["payload"]["location_tag"] == "override"
|
|
assert data["payload"]["other"] == "value"
|