matt/meshai
1
0
Fork 0
mirror of https://github.com/zvx-echo6/meshai.git synced 2026-05-21 23:24:44 +02:00
Code Issues Projects Releases Packages Wiki Activity

feat(notifications): Phase 2.5b per-channel-type renderers

Browse source 
Adds dedicated renderer classes per channel type:

- MeshRenderer produces 1+ chunks <=200 chars with (k/N) counters
  when the payload overflows. Reuses the toggle-label vocabulary
  from the digest. Mesh channels skip re-chunking when the payload
  already carries chunk_index metadata (digest path).
- EmailRenderer produces {subject, body} with structured context
  lines. Plain text only; HTML body is a future polish.
- WebhookRenderer produces a JSON-serializable dict with stable
  schema_version 1.0. Optional fields omitted (not nulled) for
  compactness. Designed for reuse by Phase 2.6.5's MQTT event
  publisher.
- All four channel implementations (MeshBroadcast, MeshDM, Email,
  Webhook) now call their renderer in deliver() before transport.
- New renderer tests cover each renderer in isolation; new channel
  integration tests confirm channels actually call their renderer.

Renderers are pure functions of the payload - no network, no
state, fully testable without mocking I/O. The future MQTT
publisher will instantiate WebhookRenderer directly.

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
K7ZVX 2026-05-15 04:25:44 +00:00
commit b2bb7f7a95
8 changed files with 898 additions and 30 deletions
Download patch file Download diff file Expand all files Collapse all files

55
meshai/notifications/channels.py
View file

@ -17,6 +17,8 @@ if TYPE_CHECKING:
from ..config import NotificationRuleConfig
from .events import NotificationPayload
from meshai.notifications.renderers import MeshRenderer, EmailRenderer, WebhookRenderer
logger = logging.getLogger(__name__)
@ -61,6 +63,7 @@ class MeshBroadcastChannel(NotificationChannel):
def __init__(self, connector: "MeshConnector", channel_index: int = 0):
self._connector = connector
self._channel = channel_index
self._renderer = MeshRenderer()
async def deliver(self, alert: "NotificationPayload", rule: "NotificationRuleConfig") -> bool:
"""Send alert to mesh channel."""
@ -69,13 +72,25 @@ class MeshBroadcastChannel(NotificationChannel):
return False
try:
message = alert.message or ""
# If payload already has chunk metadata (from digest), use message directly
if alert.chunk_index is not None:
self._connector.send_message(
text=message,
text=alert.message or "",
destination=None,
channel=self._channel,
)
logger.info("Broadcast alert to channel %d", self._channel)
logger.info("Broadcast pre-chunked alert to channel %d", self._channel)
return True
# Render to chunks for single-event delivery
chunks = self._renderer.render(alert)
for chunk in chunks:
self._connector.send_message(
text=chunk,
destination=None,
channel=self._channel,
)
logger.info("Broadcast %d chunk(s) to channel %d", len(chunks), self._channel)
return True
except Exception as e:
logger.error("Failed to broadcast alert: %s", e)
@ -159,16 +174,23 @@ class MeshDMChannel(NotificationChannel):
def __init__(self, connector: "MeshConnector", node_ids: list[str]):
self._connector = connector
self._node_ids = node_ids
self._renderer = MeshRenderer()
async def deliver(self, alert: "NotificationPayload", rule: "NotificationRuleConfig") -> bool:
"""Send alert via DM to configured nodes."""
if not self._connector:
return False
message = alert.message or ""
success = True
# If payload already has chunk metadata (from digest), use message directly
if alert.chunk_index is not None:
messages = [alert.message or ""]
else:
# Render to chunks for single-event delivery
messages = self._renderer.render(alert)
success = True
for node_id in self._node_ids:
for message in messages:
try:
node_id = str(node_id)
self._connector.send_message(text=message, destination=node_id, channel=0)
@ -287,19 +309,17 @@ class EmailChannel(NotificationChannel):
self._tls = smtp_tls
self._from = from_address
self._recipients = recipients
self._renderer = EmailRenderer()
async def deliver(self, alert: "NotificationPayload", rule: "NotificationRuleConfig") -> bool:
"""Send alert via email."""
if not self._recipients:
return False
alert_type = alert.event_type or "alert"
severity = (alert.severity or "routine").upper()
message = alert.message or ""
subject = "[MeshAI %s] %s" % (severity, alert_type.replace("_", " ").title())
body = "MeshAI Alert\n\nType: %s\nSeverity: %s\nTime: %s\n\n%s\n\n---\nAutomated message from MeshAI." % (
alert_type, severity, time.strftime("%Y-%m-%d %H:%M:%S"), message
)
# Use renderer for subject and body
rendered = self._renderer.render(alert)
subject = rendered["subject"]
body = rendered["body"]
try:
loop = asyncio.get_event_loop()
@ -515,17 +535,12 @@ class WebhookChannel(NotificationChannel):
def __init__(self, url: str, headers: Optional[dict] = None):
self._url = url
self._headers = headers or {}
self._renderer = WebhookRenderer()
async def deliver(self, alert: "NotificationPayload", rule: "NotificationRuleConfig") -> bool:
"""POST alert to webhook URL."""
payload = {
"type": alert.event_type,
"severity": alert.severity or "routine",
"message": alert.message or "",
"timestamp": alert.timestamp or time.time(),
"node_name": alert.node_name,
"region": alert.region,
}
# Use renderer for generic JSON payload
payload = self._renderer.render(alert)
# Discord/Slack format
if "discord.com" in self._url or "slack.com" in self._url:

22
meshai/notifications/renderers/__init__.py Normal file
View file

@ -0,0 +1,22 @@
"""Channel-type-aware renderers.
Each renderer takes a NotificationPayload and produces a
channel-type-appropriate output: mesh chunks, email subject/body,
or webhook JSON dict.
Renderers are reusable beyond channels.py — the future MQTT event
publisher (Phase 2.6.5) uses WebhookRenderer for its on-the-wire
format.
"""
from meshai.notifications.renderers.base import Renderer
from meshai.notifications.renderers.mesh import MeshRenderer
from meshai.notifications.renderers.email import EmailRenderer
from meshai.notifications.renderers.webhook import WebhookRenderer
__all__ = [
"Renderer",
"MeshRenderer",
"EmailRenderer",
"WebhookRenderer",
]

28
meshai/notifications/renderers/base.py Normal file
View file

@ -0,0 +1,28 @@
"""Abstract base for channel-type-aware renderers.
Each renderer takes a NotificationPayload and produces a
channel-type-appropriate output (string, list, dict, etc.).
Renderers are pure functions of their input — no network, no
state, no side effects beyond logging. The channel that owns a
renderer calls render() and then handles delivery.
"""
from abc import ABC, abstractmethod
from typing import Any
from meshai.notifications.events import NotificationPayload
class Renderer(ABC):
"""Base class for all channel-type renderers."""
@abstractmethod
def render(self, payload: NotificationPayload) -> Any:
"""Produce the channel-type-appropriate output.
Subclasses define the concrete return type:
- MeshRenderer returns list[str]
- EmailRenderer returns dict (subject, body)
- WebhookRenderer returns dict (JSON-serializable)
"""
raise NotImplementedError

78
meshai/notifications/renderers/email.py Normal file
View file

@ -0,0 +1,78 @@
"""Email channel renderer.
Produces a dict with subject and body for SMTP delivery. Plain
text body for now; HTML body is a future polish.
Subject format: "[MeshAI] <Severity> — <Event Type>"
Body format: multi-line, with the message as the lead, followed
by structured context fields (severity, region, node, timestamp,
source category).
"""
import logging
from datetime import datetime
from typing import Optional
from meshai.notifications.events import NotificationPayload
from meshai.notifications.renderers.base import Renderer
class EmailRenderer(Renderer):
"""Produce email subject and body for a single payload."""
def __init__(self):
self._logger = logging.getLogger("meshai.renderers.email")
def render(self, payload: NotificationPayload) -> dict:
"""Render the payload as {subject, body}."""
return {
"subject": self._build_subject(payload),
"body": self._build_body(payload),
}
def _build_subject(self, p: NotificationPayload) -> str:
sev = (p.severity or "routine").upper()
type_label = self._type_label(p.event_type) or "Alert"
return f"[MeshAI] {sev} — {type_label}"
def _build_body(self, p: NotificationPayload) -> str:
lines: list[str] = []
# Lead line
lines.append(p.message or "(no message)")
lines.append("") # blank separator
# Structured context
lines.append(f"Severity: {p.severity or 'routine'}")
if p.event_type:
lines.append(f"Category: {p.event_type}")
if p.region:
lines.append(f"Region: {p.region}")
if p.node_name:
lines.append(f"Node: {p.node_name}")
elif p.node_id:
lines.append(f"Node: {p.node_id}")
if p.timestamp:
try:
ts = datetime.fromtimestamp(p.timestamp)
lines.append(f"Time: {ts.strftime('%Y-%m-%d %H:%M:%S')}")
except (ValueError, OverflowError):
pass
# Optional source event detail (if present)
if p.source_event is not None:
ev = p.source_event
if hasattr(ev, "source") and ev.source:
lines.append(f"Source: {ev.source}")
if hasattr(ev, "title") and ev.title and ev.title != p.message:
lines.append(f"Title: {ev.title}")
lines.append("")
lines.append("--")
lines.append("MeshAI notification")
return "\n".join(lines)
def _type_label(self, event_type: Optional[str]) -> Optional[str]:
"""Title-cased label for the event type (for subject line)."""
if not event_type:
return None
return event_type.replace("_", " ").title()

127
meshai/notifications/renderers/mesh.py Normal file
View file

@ -0,0 +1,127 @@
"""Mesh channel renderer.
Produces a list of short strings (each ≤200 chars by default)
suitable for mesh radio broadcast. Reuses the digest's chunking
pattern: never split a single "line" across chunks; add (k/N)
counters when the rendered output produces more than one chunk.
The mesh renderer is symmetric with the digest accumulator's
mesh chunk renderer — same chunk-packing algorithm, same char
limit semantics.
"""
import logging
from typing import Optional
from meshai.notifications.events import NotificationPayload
from meshai.notifications.renderers.base import Renderer
class MeshRenderer(Renderer):
"""Produce mesh-compact chunks for a single payload."""
def __init__(self, char_limit: int = 200):
self._limit = char_limit
self._logger = logging.getLogger("meshai.renderers.mesh")
def render(self, payload: NotificationPayload) -> list[str]:
"""Render the payload as 1+ mesh-compact chunks.
Algorithm:
- Build the full message line (event_type, severity hint,
and message text — see _format_one_line)
- If the line fits in self._limit chars, return [line].
- If the line exceeds self._limit, split across multiple
chunks at word boundaries when possible, and add
"(k/N)" counters at the start of each chunk.
- If a single word exceeds the chunk limit, hard-split
mid-word (rare — long URLs etc.)
"""
line = self._format_one_line(payload)
if len(line) <= self._limit:
return [line]
return self._chunk_long_line(line)
def _format_one_line(self, p: NotificationPayload) -> str:
"""Build the headline for a payload.
Default format:
"[<EventTypeTitle>] <message>"
where EventTypeTitle is a short label derived from
p.event_type (e.g. "weather_warning" → "Weather"). If
p.event_type is None, omit the prefix.
Truncates the message at the limit only if the prefix
is short enough; otherwise lets the chunker handle it.
"""
prefix = self._toggle_label(p.event_type)
if prefix:
return f"[{prefix}] {p.message}"
return p.message
def _toggle_label(self, event_type: Optional[str]) -> Optional[str]:
"""Map an event category to a short toggle label.
Looks up the toggle the category belongs to and returns
the toggle's display label. If unknown, returns None
(no prefix added).
"""
if not event_type:
return None
from meshai.notifications.categories import get_toggle
toggle = get_toggle(event_type)
if not toggle:
return None
# Same label set used by the digest renderer
TOGGLE_LABELS = {
"mesh_health": "Mesh",
"weather": "Weather",
"fire": "Fire",
"rf_propagation": "RF",
"roads": "Roads",
"avalanche": "Avalanche",
"seismic": "Seismic",
"tracking": "Tracking",
"other": "Other",
}
return TOGGLE_LABELS.get(toggle, toggle.title())
def _chunk_long_line(self, line: str) -> list[str]:
"""Split a long line into chunks ≤ self._limit each.
Reserves ~10 chars per chunk for the "(k/N)" counter
suffix. Splits at word boundaries; falls back to
mid-word split if a single word exceeds the budget.
"""
# Reserve space for " (k/N)" — generous to 10 chars
body_budget = self._limit - 10
if body_budget <= 0:
body_budget = self._limit # extreme small limit; ignore counter
words = line.split(" ")
chunks: list[str] = []
current = ""
for word in words:
# Hard-split words longer than body_budget (rare)
while len(word) > body_budget:
if current:
chunks.append(current)
current = ""
chunks.append(word[:body_budget])
word = word[body_budget:]
# Add word to current chunk if it fits
tentative = (current + " " + word) if current else word
if len(tentative) <= body_budget:
current = tentative
else:
chunks.append(current)
current = word
if current:
chunks.append(current)
# Add counters: "DIGEST"-style "(k/N)" suffix
total = len(chunks)
if total == 1:
return chunks
return [f"{chunk} ({i+1}/{total})" for i, chunk in enumerate(chunks)]

67
meshai/notifications/renderers/webhook.py Normal file
View file

@ -0,0 +1,67 @@
"""Webhook channel renderer.
Produces a JSON-serializable dict with stable field names. Also
intended for reuse by the future MQTT event publisher (Phase
2.6.5), which wants the same structured shape on the wire.
Field names use snake_case. Optional fields are omitted from the
output when None, NOT set to null — keeps payloads compact and
avoids ambiguity between "field absent" and "field explicitly
null".
"""
import logging
from typing import Any
from meshai.notifications.events import NotificationPayload
from meshai.notifications.renderers.base import Renderer
# Schema version for the wire format. Bump when making
# breaking changes to the field shape. External consumers
# can use this to handle multiple versions.
WEBHOOK_SCHEMA_VERSION = "1.0"
class WebhookRenderer(Renderer):
"""Produce a JSON-serializable dict for a single payload."""
def __init__(self):
self._logger = logging.getLogger("meshai.renderers.webhook")
def render(self, payload: NotificationPayload) -> dict:
"""Render the payload as a structured dict."""
out: dict[str, Any] = {
"schema_version": WEBHOOK_SCHEMA_VERSION,
"message": payload.message,
"severity": payload.severity or "routine",
"timestamp": payload.timestamp,
}
# Optional fields — omit if None
for src_attr, dst_key in (
("category", "category"),
("event_type", "event_type"),
("node_id", "node_id"),
("node_name", "node_name"),
("region", "region"),
("chunk_index", "chunk_index"),
("chunk_total", "chunk_total"),
):
value = getattr(payload, src_attr, None)
if value is not None:
out[dst_key] = value
# Optional source event detail
if payload.source_event is not None:
ev = payload.source_event
source_event: dict[str, Any] = {}
for attr in ("id", "source", "title", "expires", "group_key"):
if hasattr(ev, attr):
value = getattr(ev, attr)
if value is not None:
source_event[attr] = value
if source_event:
out["source_event"] = source_event
return out

214
tests/test_channel_rendering.py Normal file
View file

@ -0,0 +1,214 @@
"""Tests for channel-renderer integration (Phase 2.5b)."""
import asyncio
import time
from unittest.mock import MagicMock, AsyncMock, patch
import pytest
from meshai.notifications.events import NotificationPayload
from meshai.notifications.channels import (
MeshBroadcastChannel,
MeshDMChannel,
EmailChannel,
WebhookChannel,
)
# ============================================================
# MESH CHANNEL RENDERING TESTS
# ============================================================
def test_mesh_channel_uses_mesh_renderer():
"""MeshBroadcastChannel renders long messages to multiple chunks."""
mock_connector = MagicMock()
channel = MeshBroadcastChannel(
connector=mock_connector,
channel_index=0,
)
# Build a long message that will require chunking
long_message = "This is a very long alert message that exceeds the character limit. " * 5
payload = NotificationPayload(
message=long_message,
category="weather_warning",
severity="priority",
timestamp=time.time(),
event_type="weather_warning",
)
asyncio.run(channel.deliver(payload, None))
# Should have called send_message multiple times (once per chunk)
assert mock_connector.send_message.call_count >= 2
# Each call's text should be <= 200 chars
for call in mock_connector.send_message.call_args_list:
text = call.kwargs.get("text", call.args[0] if call.args else "")
assert len(text) <= 200
def test_mesh_channel_uses_payload_message_directly_when_chunk_metadata_set():
"""Pre-chunked payloads (from digest) skip re-rendering."""
mock_connector = MagicMock()
channel = MeshBroadcastChannel(
connector=mock_connector,
channel_index=0,
)
# Payload with chunk metadata set (from digest scheduler)
payload = NotificationPayload(
message="pre-chunked text",
category="digest",
severity="routine",
timestamp=time.time(),
chunk_index=1,
chunk_total=3,
)
asyncio.run(channel.deliver(payload, None))
# Should have called send_message exactly once
assert mock_connector.send_message.call_count == 1
# Should use the message directly
call = mock_connector.send_message.call_args
text = call.kwargs.get("text", call.args[0] if call.args else "")
assert text == "pre-chunked text"
def test_mesh_dm_channel_uses_mesh_renderer():
"""MeshDMChannel renders long messages to chunks for each recipient."""
mock_connector = MagicMock()
channel = MeshDMChannel(
connector=mock_connector,
node_ids=["!node1", "!node2"],
)
long_message = "This is a long DM message that should be chunked. " * 4
payload = NotificationPayload(
message=long_message,
category="test",
severity="routine",
timestamp=time.time(),
)
asyncio.run(channel.deliver(payload, None))
# Should have called send_message multiple times
# (chunks * nodes)
assert mock_connector.send_message.call_count >= 2
def test_mesh_dm_channel_uses_payload_message_directly_when_chunk_metadata_set():
"""Pre-chunked DM payloads skip re-rendering."""
mock_connector = MagicMock()
channel = MeshDMChannel(
connector=mock_connector,
node_ids=["!node1"],
)
payload = NotificationPayload(
message="pre-chunked DM",
category="digest",
severity="routine",
timestamp=time.time(),
chunk_index=2,
chunk_total=5,
)
asyncio.run(channel.deliver(payload, None))
# Should use message directly, once per node
assert mock_connector.send_message.call_count == 1
call = mock_connector.send_message.call_args
text = call.kwargs.get("text", call.args[0] if call.args else "")
assert text == "pre-chunked DM"
# ============================================================
# EMAIL CHANNEL RENDERING TESTS
# ============================================================
def test_email_channel_uses_email_renderer():
"""EmailChannel uses renderer for subject and body."""
channel = EmailChannel(
smtp_host="localhost",
smtp_port=25,
smtp_user="",
smtp_password="",
smtp_tls=False,
from_address="test@example.com",
recipients=["user@example.com"],
)
payload = NotificationPayload(
message="Test alert message",
category="weather_warning",
severity="immediate",
timestamp=time.time(),
event_type="weather_warning",
)
# Mock the _send_email method
with patch.object(channel, "_send_email") as mock_send:
asyncio.run(channel.deliver(payload, None))
# Should have been called with renderer output
mock_send.assert_called_once()
call_args = mock_send.call_args
subject = call_args.args[0]
body = call_args.args[1]
# Renderer format checks
assert "[MeshAI]" in subject
assert "IMMEDIATE" in subject
assert "Test alert message" in body
assert "Severity:" in body
# ============================================================
# WEBHOOK CHANNEL RENDERING TESTS
# ============================================================
def test_webhook_channel_uses_webhook_renderer():
"""WebhookChannel uses renderer for JSON payload."""
channel = WebhookChannel(
url="https://example.com/webhook",
headers={},
)
payload = NotificationPayload(
message="Test webhook message",
category="test",
severity="priority",
timestamp=time.time(),
event_type="battery_warning",
)
# Mock httpx
with patch("meshai.notifications.channels.httpx.AsyncClient") as mock_client_class:
mock_client = MagicMock()
mock_response = MagicMock()
mock_response.status_code = 200
mock_client.post = AsyncMock(return_value=mock_response)
mock_client.__aenter__ = AsyncMock(return_value=mock_client)
mock_client.__aexit__ = AsyncMock(return_value=None)
mock_client_class.return_value = mock_client
asyncio.run(channel.deliver(payload, None))
# Check the POST was called
mock_client.post.assert_called_once()
call_kwargs = mock_client.post.call_args.kwargs
# Should have JSON payload with schema_version
json_payload = call_kwargs.get("json", {})
assert "schema_version" in json_payload
assert json_payload["schema_version"] == "1.0"
assert json_payload["message"] == "Test webhook message"

317
tests/test_renderers.py Normal file
View file

@ -0,0 +1,317 @@
"""Tests for Phase 2.5b per-channel-type renderers."""
import json
import re
import time
import pytest
from meshai.notifications.events import NotificationPayload, make_event, make_payload_from_event
from meshai.notifications.renderers import MeshRenderer, EmailRenderer, WebhookRenderer
# ============================================================
# MESH RENDERER TESTS
# ============================================================
def test_mesh_render_short_message_single_chunk():
"""Short message produces a single chunk."""
payload = NotificationPayload(
message="Test alert",
category="test",
severity="routine",
timestamp=time.time(),
)
renderer = MeshRenderer()
chunks = renderer.render(payload)
assert len(chunks) == 1
assert "Test alert" in chunks[0]
def test_mesh_render_event_type_prefix():
"""Known event type adds toggle label prefix."""
payload = NotificationPayload(
message="Severe storm",
category="weather_warning",
severity="priority",
timestamp=time.time(),
event_type="weather_warning",
)
renderer = MeshRenderer()
chunks = renderer.render(payload)
assert len(chunks) == 1
assert chunks[0].startswith("[Weather]")
def test_mesh_render_unknown_event_type_no_prefix():
"""Unknown event type does not add a prefix."""
payload = NotificationPayload(
message="Hello",
category="made_up_thing",
severity="routine",
timestamp=time.time(),
event_type="made_up_thing",
)
renderer = MeshRenderer()
chunks = renderer.render(payload)
assert len(chunks) == 1
assert not chunks[0].startswith("[")
def test_mesh_render_long_message_chunks():
"""Long message splits into multiple chunks with counters."""
# Build a ~500 char message
long_message = "This is a very long alert message that should definitely exceed the two hundred character limit. " * 5
payload = NotificationPayload(
message=long_message,
category="weather_warning",
severity="priority",
timestamp=time.time(),
event_type="weather_warning",
)
renderer = MeshRenderer()
chunks = renderer.render(payload)
assert len(chunks) >= 2
for chunk in chunks:
assert len(chunk) <= 200, f"Chunk exceeds limit: {len(chunk)} chars"
# Check each chunk ends with "(k/N)" counter
for chunk in chunks:
assert re.search(r"\(\d+/\d+\)$", chunk), f"Missing counter: {chunk}"
def test_mesh_render_chunks_preserve_full_content():
"""Chunking preserves all words from the original message."""
# Build a message with distinct tokens
words = ["alpha", "bravo", "charlie", "delta", "echo", "foxtrot", "golf",
"hotel", "india", "juliet", "kilo", "lima", "mike", "november",
"oscar", "papa", "quebec", "romeo", "sierra", "tango", "uniform",
"victor", "whiskey", "xray", "yankee", "zulu"]
long_message = " ".join(words * 3) # Repeat to span multiple chunks
payload = NotificationPayload(
message=long_message,
category="test",
severity="routine",
timestamp=time.time(),
)
renderer = MeshRenderer()
chunks = renderer.render(payload)
# Join all chunks and remove counters
combined = " ".join(chunks)
combined = re.sub(r"\s*\(\d+/\d+\)", "", combined)
# Every original word should appear
for word in words:
assert word in combined, f"Missing word: {word}"
def test_mesh_render_no_event_type():
"""Payload without event_type has no prefix."""
payload = NotificationPayload(
message="Plain message",
category="test",
severity="routine",
timestamp=time.time(),
event_type=None,
)
renderer = MeshRenderer()
chunks = renderer.render(payload)
assert len(chunks) == 1
assert chunks[0] == "Plain message"
# ============================================================
# EMAIL RENDERER TESTS
# ============================================================
def test_email_render_subject_includes_severity_and_type():
"""Subject line includes severity and event type."""
payload = NotificationPayload(
message="Storm approaching",
category="weather_warning",
severity="immediate",
timestamp=time.time(),
event_type="weather_warning",
)
renderer = EmailRenderer()
rendered = renderer.render(payload)
assert "IMMEDIATE" in rendered["subject"]
assert "Weather Warning" in rendered["subject"]
def test_email_render_body_includes_message_and_context():
"""Body includes message and structured context fields."""
fixed_time = 1700000000.0 # Known timestamp
payload = NotificationPayload(
message="Test alert message",
category="battery_warning",
severity="priority",
timestamp=fixed_time,
event_type="battery_warning",
region="Magic Valley",
node_name="BLD-MTN",
)
renderer = EmailRenderer()
rendered = renderer.render(payload)
body = rendered["body"]
assert "Test alert message" in body
assert "Magic Valley" in body
assert "BLD-MTN" in body
assert "priority" in body
assert "battery_warning" in body
# Check formatted date
assert "2023-11-14" in body # Date part of timestamp
def test_email_render_omits_missing_context():
"""Body omits lines for missing optional fields."""
payload = NotificationPayload(
message="Minimal alert",
category="test",
severity="routine",
timestamp=time.time(),
# No region, no node, no event_type
)
renderer = EmailRenderer()
rendered = renderer.render(payload)
body = rendered["body"]
assert "Minimal alert" in body
assert "Severity: routine" in body
assert "Region:" not in body
assert "Node:" not in body
assert "Category:" not in body
def test_email_render_includes_source_event():
"""Body includes source event details when present."""
event = make_event(
source="weather_adapter",
category="weather_warning",
severity="priority",
title="Severe Storm",
summary="Severe storm expected",
)
payload = make_payload_from_event(event)
renderer = EmailRenderer()
rendered = renderer.render(payload)
body = rendered["body"]
assert "weather_adapter" in body
assert "Severe Storm" in body
# ============================================================
# WEBHOOK RENDERER TESTS
# ============================================================
def test_webhook_render_has_schema_version():
"""Output includes schema_version field."""
payload = NotificationPayload(
message="Test",
category="test",
severity="routine",
timestamp=time.time(),
)
renderer = WebhookRenderer()
rendered = renderer.render(payload)
assert rendered["schema_version"] == "1.0"
def test_webhook_render_omits_none_fields():
"""None optional fields are omitted, not set to null."""
payload = NotificationPayload(
message="Test message",
category="test",
severity="routine",
timestamp=time.time(),
# All optional fields default to None
)
renderer = WebhookRenderer()
rendered = renderer.render(payload)
# Required fields present
assert "message" in rendered
assert "severity" in rendered
assert "timestamp" in rendered
assert "schema_version" in rendered
# Optional fields omitted
assert "node_id" not in rendered
assert "region" not in rendered
assert "chunk_index" not in rendered
def test_webhook_render_includes_source_event_when_present():
"""source_event dict included when payload has source event."""
event = make_event(
source="test_adapter",
category="test",
severity="routine",
title="Test Event",
)
payload = make_payload_from_event(event)
renderer = WebhookRenderer()
rendered = renderer.render(payload)
assert "source_event" in rendered
assert rendered["source_event"]["id"] == event.id
assert rendered["source_event"]["source"] == "test_adapter"
def test_webhook_render_is_json_serializable():
"""Rendered output is JSON-serializable (no datetime objects)."""
event = make_event(
source="test",
category="test",
severity="routine",
title="Test",
)
payload = make_payload_from_event(event)
payload.chunk_index = 1
payload.chunk_total = 3
payload.region = "Test Region"
payload.node_name = "Test-Node"
renderer = WebhookRenderer()
rendered = renderer.render(payload)
# Should not raise
json_str = json.dumps(rendered)
assert isinstance(json_str, str)
# Verify round-trip
parsed = json.loads(json_str)
assert parsed["message"] == payload.message
def test_webhook_render_includes_optional_fields_when_set():
"""Optional fields included when they have values."""
payload = NotificationPayload(
message="Test",
category="test_category",
severity="priority",
timestamp=time.time(),
event_type="battery_warning",
node_id="!abc123",
node_name="Test-Node",
region="Test Region",
chunk_index=2,
chunk_total=5,
)
renderer = WebhookRenderer()
rendered = renderer.render(payload)
assert rendered["category"] == "test_category"
assert rendered["event_type"] == "battery_warning"
assert rendered["node_id"] == "!abc123"
assert rendered["node_name"] == "Test-Node"
assert rendered["region"] == "Test Region"
assert rendered["chunk_index"] == 2
assert rendered["chunk_total"] == 5