david/nr-flow-validator
1
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

doorbell: fake zigbee probe device + E2E validation (no real ring)

Browse Source
This commit is contained in:
David Gwilliam
2026-08-28 12:56:25 -07:00
parent 595796b027
commit d17c8dea98
5 changed files with 126 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files
infra/compose.yaml
+18
View File
@@ -106,11 +106,29 @@ services:
MQTT_HOST: 192.168.81.147
MQTT_PORT: "1883"
PLUG_TOPIC: "zigbee2mqtt/Squiggle/set"
PROBE_PLUG_TOPIC: "zigbee2mqtt/DoorbellProbe/set"
POLL_INTERVAL: "5"
MAX_AGE_SECONDS: "180"
FLASH_INTERVAL_SECONDS: "1"
FLASH_DURATION_SECONDS: "7"
TZ: America/Los_Angeles
# Fake Zigbee smart-plug used by the doorbell E2E probe. Echoes TOGGLE
# commands back on its state topic so the chain can be validated without
# flashing a real lamp. No production traffic is routed here.
doorbell-probe:
build: ./doorbell-probe
container_name: doorbell-probe
restart: unless-stopped
network_mode: host
depends_on:
mosquitto:
condition: service_healthy
environment:
MQTT_HOST: 192.168.81.147
MQTT_PORT: "1883"
SET_TOPIC: "zigbee2mqtt/DoorbellProbe/set"
STATE_TOPIC: "zigbee2mqtt/DoorbellProbe"
volumes:
librespot_cache:
infra/doorbell-listener/doorbell-listener.py
+13 -5
View File
@@ -44,6 +44,10 @@ NTFY_TOPICS = [t.strip() for t in os.environ.get("NTFY_TOPICS", DEFAULT_TOPICS).
MQTT_HOST = os.environ.get("MQTT_HOST", "mosquitto")
MQTT_PORT = int(os.environ.get("MQTT_PORT", "1883"))
PLUG_TOPIC = os.environ.get("PLUG_TOPIC", "zigbee2mqtt/Sideboard Lamp/set")
# Canary/"probe" alerts (tagged doorbell-probe) are routed to a fake zigbee
# device instead of the real plug, so E2E tests don't flash a real lamp.
PROBE_PLUG_TOPIC = os.environ.get("PROBE_PLUG_TOPIC", "zigbee2mqtt/DoorbellProbe/set")
PROBE_TAG = os.environ.get("PROBE_TAG", "doorbell-probe")
STATE_DIR = Path(os.environ.get("STATE_DIR", "/data/doorbell-listener"))
POLL_INTERVAL = int(os.environ.get("POLL_INTERVAL", "30"))
# Drop messages older than this many seconds (replay protection after restart).
@@ -85,7 +89,7 @@ def save_last_id(topic: str, msg_id: str) -> None:
last_id_path(topic).write_text(msg_id)
def start_flash(client: mqtt.Client, count: int, interval: float) -> None:
def start_flash(client: mqtt.Client, count: int, interval: float, topic: str) -> None:
"""Spawn a thread that publishes `count` TOGGLEs spaced `interval` seconds.
Each alert gets its own thread; overlapping flashes will publish toggles
@@ -93,11 +97,11 @@ def start_flash(client: mqtt.Client, count: int, interval: float) -> None:
regardless of how many threads are running.
"""
def run() -> None:
log(f"flash: starting {count} toggles @ {interval}s")
log(f"flash: starting {count} toggles @ {interval}s -> {topic}")
for i in range(count):
payload = json.dumps({"state": "TOGGLE"})
client.publish(PLUG_TOPIC, payload)
log(f"flash {i + 1}/{count} -> {PLUG_TOPIC}")
client.publish(topic, payload)
log(f"flash {i + 1}/{count} -> {topic}")
if i < count - 1:
time.sleep(interval)
log("flash: complete")
@@ -146,7 +150,11 @@ def poll_topic(client: mqtt.Client, topic: str, last_id: str) -> str:
continue
log(f"[{topic}] alert: id={msg_id} age={age}s title={title!r} message={message!r}")
start_flash(client, n_flashes, FLASH_INTERVAL_SECONDS)
tags = event.get("tags") or []
is_probe = PROBE_TAG in tags or event.get("title") == PROBE_TAG
target = PROBE_PLUG_TOPIC if is_probe else PLUG_TOPIC
log(f"[{topic}] probe={is_probe} -> {target}")
start_flash(client, n_flashes, FLASH_INTERVAL_SECONDS, target)
return newest_id
infra/doorbell-probe/Dockerfile
+5
View File
@@ -0,0 +1,5 @@
FROM ghcr.io/astral-sh/uv:python3.12-bookworm-slim
WORKDIR /app
COPY fake-device.py .
CMD ["uv", "run", "--no-cache", "fake-device.py"]
infra/doorbell-probe/e2e-check.sh
+27
View File
@@ -0,0 +1,27 @@
#!/bin/bash
# Validate the full doorbell chain WITHOUT flashing a real lamp:
# ntfy.sh -> doorbell-listener -> MQTT -> fake zigbee device
# Publishes a probe canary (tagged "doorbell-probe") to the ntfy test topic,
# then waits for the fake device to echo a TOGGLE on its state topic.
set -u
NTFY_TOPIC="${NTFY_TOPIC:-ALERT_klubhaus_topic_test}"
PROBE_STATE="${PROBE_STATE:-zigbee2mqtt/DoorbellProbe}"
MQTT_HOST="${MQTT_HOST:-127.0.0.1}"
TIMEOUT="${TIMEOUT:-25}"
marker="e2e-$(date +%s)-$$"
echo "publishing probe canary to ntfy.sh/$NTFY_TOPIC"
curl -fsS -m 10 -o /dev/null -X POST "https://ntfy.sh/$NTFY_TOPIC" \
-H "Title: doorbell-probe" \
-H "Tags: doorbell-probe" \
-d "$marker" || { echo "E2E FAIL: ntfy publish failed"; exit 2; }
echo "waiting for TOGGLE echo on $PROBE_STATE (<=${TIMEOUT}s)"
if timeout "$TIMEOUT" docker exec mosquitto mosquitto_sub -h 127.0.0.1 -t "$PROBE_STATE" -C 1 -W "$TIMEOUT" 2>/dev/null | grep -q TOGGLE; then
echo "E2E OK"
exit 0
else
echo "E2E FAIL: no TOGGLE echo on $PROBE_STATE within ${TIMEOUT}s"
exit 1
fi
infra/doorbell-probe/fake-device.py
+63
View File
@@ -0,0 +1,63 @@
#!/usr/bin/env python3
# /// script
# requires-python = ">=3.10"
# dependencies = [
# "paho-mqtt>=2.0",
# ]
# ///
"""Fake Zigbee device for doorbell E2E validation.
Subscribes to a "set" topic and echoes a TOGGLE back to a "state" topic
(retained), emulating a Zigbee smart plug at the MQTT layer. This lets the
doorbell E2E probe confirm the full chain (ntfy.sh -> listener -> MQTT -> device)
without toggling a real lamp.
Env vars:
MQTT_HOST default "mosquitto"
MQTT_PORT default 1883
SET_TOPIC default "zigbee2mqtt/DoorbellProbe/set"
STATE_TOPIC default "zigbee2mqtt/DoorbellProbe"
"""
import json
import os
import paho.mqtt.client as mqtt
MQTT_HOST = os.environ.get("MQTT_HOST", "mosquitto")
MQTT_PORT = int(os.environ.get("MQTT_PORT", "1883"))
SET_TOPIC = os.environ.get("SET_TOPIC", "zigbee2mqtt/DoorbellProbe/set")
STATE_TOPIC = os.environ.get("STATE_TOPIC", "zigbee2mqtt/DoorbellProbe")
def log(msg: str) -> None:
print(msg, flush=True)
def on_connect(client, userdata, flags, reason_code, properties=None):
log(f"connected to {MQTT_HOST}:{MQTT_PORT} (rc={reason_code})")
client.subscribe(SET_TOPIC)
def on_message(client, userdata, msg):
log(f"echo -> {STATE_TOPIC} (from {msg.topic})")
client.publish(STATE_TOPIC, json.dumps({"state": "TOGGLE", "echo": True}), retain=True)
def main() -> None:
log(f"fake-device: {SET_TOPIC} -> {STATE_TOPIC}")
client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2)
client.on_connect = on_connect
client.on_message = on_message
while True:
try:
client.connect(MQTT_HOST, MQTT_PORT, 60)
break
except Exception as e:
log(f"connect failed: {e}; retry in 5s")
import time
time.sleep(5)
client.loop_forever()
if __name__ == "__main__":
main()