van-router-mqtt: subscribe and publish discovery in on_connect
Clean sessions lose subscriptions on reconnect and a refused first connect loses the retained discovery, so both now happen on every successful connect. Adds reason codes 132-136 and behavioural tests. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01KqhjP7nud5gdBg25zCiRk9
This commit is contained in:
@@ -561,7 +561,8 @@ def on_message(mqc, userdata, msg):
|
|||||||
REFUSAL_REASONS = {
|
REFUSAL_REASONS = {
|
||||||
1: "unacceptable protocol version", 2: "client identifier rejected",
|
1: "unacceptable protocol version", 2: "client identifier rejected",
|
||||||
3: "server unavailable", 4: "bad username or password", 5: "not authorised",
|
3: "server unavailable", 4: "bad username or password", 5: "not authorised",
|
||||||
134: "bad username or password", 135: "not authorised",
|
132: "unsupported protocol version", 133: "client identifier rejected",
|
||||||
|
134: "bad username or password", 135: "not authorised", 136: "server unavailable",
|
||||||
}
|
}
|
||||||
_mqtt_state = None # None | "connected" | "refused" | "down": log changes only
|
_mqtt_state = None # None | "connected" | "refused" | "down": log changes only
|
||||||
|
|
||||||
@@ -597,6 +598,16 @@ def note_disconnect(rc):
|
|||||||
|
|
||||||
def on_connect(mqc, userdata, flags, reason_code, properties=None):
|
def on_connect(mqc, userdata, flags, reason_code, properties=None):
|
||||||
note_connect(reason_code)
|
note_connect(reason_code)
|
||||||
|
if getattr(reason_code, "value", reason_code) != 0:
|
||||||
|
return
|
||||||
|
# (Re)subscribe and republish discovery on EVERY successful connect: a clean
|
||||||
|
# session drops subscriptions on reconnect and a refused first connect loses them.
|
||||||
|
try:
|
||||||
|
for topic in COMMANDS:
|
||||||
|
mqc.subscribe(topic)
|
||||||
|
publish_discovery(mqc)
|
||||||
|
except Exception as e:
|
||||||
|
log(f"MQTT on_connect setup failed: {type(e).__name__}: {e}")
|
||||||
|
|
||||||
|
|
||||||
def on_disconnect(mqc, userdata, disconnect_flags, reason_code, properties=None):
|
def on_disconnect(mqc, userdata, disconnect_flags, reason_code, properties=None):
|
||||||
@@ -611,10 +622,7 @@ def main():
|
|||||||
mqc.on_connect = on_connect
|
mqc.on_connect = on_connect
|
||||||
mqc.on_disconnect = on_disconnect
|
mqc.on_disconnect = on_disconnect
|
||||||
mqc.connect(broker, mqtt_port, keepalive=60)
|
mqc.connect(broker, mqtt_port, keepalive=60)
|
||||||
for topic in COMMANDS:
|
|
||||||
mqc.subscribe(topic)
|
|
||||||
mqc.loop_start()
|
mqc.loop_start()
|
||||||
publish_discovery(mqc)
|
|
||||||
log(f"van-router-mqtt started: broker={broker}:{mqtt_port} poll={poll_interval_s}s")
|
log(f"van-router-mqtt started: broker={broker}:{mqtt_port} poll={poll_interval_s}s")
|
||||||
while True:
|
while True:
|
||||||
publish_state(mqc)
|
publish_state(mqc)
|
||||||
|
|||||||
@@ -733,13 +733,43 @@ with contextlib.redirect_stdout(buf):
|
|||||||
open(cp, "w").write("{not json"); r = m.load_credentials(cfgp, {"broker": "b"}); check("invalid JSON -> unchanged", r == {"broker": "b"})
|
open(cp, "w").write("{not json"); r = m.load_credentials(cfgp, {"broker": "b"}); check("invalid JSON -> unchanged", r == {"broker": "b"})
|
||||||
open(cp, "w").write("[1]"); r = m.load_credentials(cfgp, {"broker": "b"}); check("non-object -> unchanged", r == {"broker": "b"})
|
open(cp, "w").write("[1]"); r = m.load_credentials(cfgp, {"broker": "b"}); check("non-object -> unchanged", r == {"broker": "b"})
|
||||||
open(cp, "w").write('{"username": "", "password": "x"}'); r = m.load_credentials(cfgp, {"broker": "b"}); check("empty username -> anonymous", "username" not in r and "password" not in r)
|
open(cp, "w").write('{"username": "", "password": "x"}'); r = m.load_credentials(cfgp, {"broker": "b"}); check("empty username -> anonymous", "username" not in r and "password" not in r)
|
||||||
os.chmod(cp, 0); r = m.load_credentials(cfgp, {"broker": "b"}); check("unreadable -> unchanged", r == {"broker": "b"} or os.geteuid() == 0)
|
if os.geteuid() == 0:
|
||||||
|
res.append(("unreadable -> unchanged (SKIPPED: running as root)", True))
|
||||||
|
else:
|
||||||
|
os.chmod(cp, 0); r = m.load_credentials(cfgp, {"broker": "b"}); check("unreadable -> unchanged", r == {"broker": "b"})
|
||||||
os.chmod(cp, 0o600); open(cp, "w").write('{"username": "u", "password": "topsecretpw"}')
|
os.chmod(cp, 0o600); open(cp, "w").write('{"username": "u", "password": "topsecretpw"}')
|
||||||
m.load_credentials(cfgp, {})
|
m.load_credentials(cfgp, {})
|
||||||
check("credentials path beside config", m.credentials_path("/etc/van-router-mqtt/config.json") == "/etc/van-router-mqtt/credentials.json")
|
check("credentials path beside config", m.credentials_path("/etc/van-router-mqtt/config.json") == "/etc/van-router-mqtt/credentials.json")
|
||||||
check("reason rc4", "bad username or password" in m.refusal_reason(4))
|
check("reason rc4", "bad username or password" in m.refusal_reason(4))
|
||||||
check("reason rc5", "not authorised" in m.refusal_reason(5))
|
check("reason rc5", "not authorised" in m.refusal_reason(5))
|
||||||
check("reason unknown", "unknown" in m.refusal_reason(99) and "99" in m.refusal_reason(99))
|
check("reason unknown", "unknown" in m.refusal_reason(99) and "99" in m.refusal_reason(99))
|
||||||
|
check("reason 132-136", "unsupported protocol" in m.refusal_reason(132) and "identifier" in m.refusal_reason(133))
|
||||||
|
check("reason 134", "bad username or password" in m.refusal_reason(134))
|
||||||
|
check("reason 135", "not authorised" in m.refusal_reason(135))
|
||||||
|
check("reason 136", "server unavailable" in m.refusal_reason(136))
|
||||||
|
class Fake:
|
||||||
|
def __init__(s): s.subs = []; s.pubs = []
|
||||||
|
def subscribe(s, t): s.subs.append(t)
|
||||||
|
def publish(s, t, *a, **k): s.pubs.append(t)
|
||||||
|
m.COMMANDS.update({"c/one": None, "c/two": None})
|
||||||
|
m._mqtt_state = None
|
||||||
|
f = Fake(); m.on_connect(f, None, {}, 135)
|
||||||
|
check("refused on_connect subscribes/publishes nothing", not f.subs and not f.pubs)
|
||||||
|
f = Fake(); m.on_connect(f, None, {}, 0)
|
||||||
|
check("ok on_connect subscribes every COMMANDS topic", sorted(f.subs) == sorted(m.COMMANDS) and len(f.subs) >= 2)
|
||||||
|
check("ok on_connect publishes discovery", len(f.pubs) > 0 and all("/config" in t for t in f.pubs))
|
||||||
|
m.on_disconnect(f, None, None, 7)
|
||||||
|
m.on_connect(f, None, {}, 0)
|
||||||
|
check("reconnect subscribes again", len(f.subs) == 2 * len(m.COMMANDS))
|
||||||
|
class Boom(Fake):
|
||||||
|
def subscribe(s, t): raise RuntimeError("boom")
|
||||||
|
try: m.on_connect(Boom(), None, {}, 0); check("exception in on_connect caught", True)
|
||||||
|
except Exception: check("exception in on_connect caught", False)
|
||||||
|
try: m.on_connect(None, None, {}, 135); m.on_disconnect(None, None, None, 7); check("callbacks with paho-2 signatures do not raise", True)
|
||||||
|
except Exception: check("callbacks with paho-2 signatures do not raise", False)
|
||||||
|
src = open("router-status/van-router-mqtt").read()
|
||||||
|
check("main wires callbacks", "mqc.on_connect = on_connect" in src and "mqc.on_disconnect = on_disconnect" in src)
|
||||||
|
m._mqtt_state = None
|
||||||
mark = buf.tell()
|
mark = buf.tell()
|
||||||
m.note_connect(5); m.note_connect(5); m.note_disconnect(5); m.note_connect(5)
|
m.note_connect(5); m.note_connect(5); m.note_disconnect(5); m.note_connect(5)
|
||||||
refused = buf.getvalue()[mark:]
|
refused = buf.getvalue()[mark:]
|
||||||
|
|||||||
Reference in New Issue
Block a user