Initial commit: UlHub workspace

test/test_ws_mqtt.py

@@ -0,0 +1,84 @@
+#!/usr/bin/env python3
+"""Test MQTT over WebSocket (wss) via nginx proxy at /mqtt.
+Usage:
+  pip install -r requirements.txt
+  python test_ws_mqtt.py
+
+Environment:
+  MQTT_HOST (default: dev.hub.umagul.net)
+  MQTT_PORT (default: 443)
+  MQTT_PATH (default: /mqtt)
+  MQTT_TOPIC (default: devices/#)
+  MQTT_USERNAME, MQTT_PASSWORD (optional)
+  TLS_VERIFY (true/false, default: true)
+"""
+import os
+import ssl
+import time
+import json
+import sys
+
+import paho.mqtt.client as mqtt
+
+HOST = os.getenv("MQTT_HOST", "dev.hub.umagul.net")
+PORT = int(os.getenv("MQTT_PORT", "443"))
+PATH = os.getenv("MQTT_PATH", "/mqtt")
+TOPIC = os.getenv("MQTT_TOPIC", "devices/#")
+USERNAME = os.getenv("MQTT_USERNAME", "testuser")
+PASSWORD = os.getenv("MQTT_PASSWORD", "testpass")
+TLS_VERIFY = os.getenv("TLS_VERIFY", "true").lower() in ("1", "true", "yes")
+
+received = 0
+
+def on_connect(client, userdata, flags, rc):
+    if rc == 0:
+        print(f"connected to wss://{HOST}{PATH} (rc={rc})")
+        client.subscribe(TOPIC)
+        print(f"subscribed to {TOPIC}")
+    else:
+        print("connect failed rc=", rc)
+
+
+def on_message(client, userdata, msg):
+    global received
+    received += 1
+    payload = msg.payload.decode(errors="ignore")
+    print(f"[{received}] {msg.topic} -> {payload}")
+
+
+def on_disconnect(client, userdata, rc):
+    print("disconnected", rc)
+
+
+def main():
+    client = mqtt.Client(transport="websockets")
+    if USERNAME:
+        client.username_pw_set(USERNAME, PASSWORD)
+
+    client.ws_set_options(path=PATH)
+
+    if TLS_VERIFY:
+        client.tls_set()  # default system CA
+    else:
+        client.tls_set(cert_reqs=ssl.CERT_NONE)
+        client.tls_insecure_set(True)
+
+    client.on_connect = on_connect
+    client.on_message = on_message
+    client.on_disconnect = on_disconnect
+
+    try:
+        client.connect(HOST, PORT, keepalive=60)
+    except Exception as e:
+        print("connect error:", e)
+        sys.exit(1)
+
+    try:
+        client.loop_forever()
+    except KeyboardInterrupt:
+        print("exiting")
+        client.disconnect()
+
+
+if __name__ == "__main__":
+    main()