5a1067263a
- patterns/: sample dynamic pattern modules for OTA - esp32/msg.json: example bridge message shape - models/http_driver.py, wifi_peer.py: Wi-Fi driver HTTP poll helpers - tests: pattern OTA send script and UDP discovery echo server - Submodule led-driver: http_poll and test utilities Made-with: Cursor
100 lines
2.9 KiB
Python
100 lines
2.9 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Manual test helper for pattern OTA send flow.
|
|
|
|
Examples:
|
|
python tests/test_pattern_ota_send.py --base-url http://led.local --pattern blink
|
|
python tests/test_pattern_ota_send.py --base-url http://127.0.0.1:8080 --pattern blink --device-id 102030405060
|
|
"""
|
|
|
|
import argparse
|
|
import json
|
|
import sys
|
|
from urllib import request, error
|
|
|
|
|
|
def _http_json(method, url, payload=None):
|
|
data = None
|
|
headers = {"Accept": "application/json"}
|
|
if payload is not None:
|
|
data = json.dumps(payload).encode("utf-8")
|
|
headers["Content-Type"] = "application/json"
|
|
req = request.Request(url, data=data, method=method, headers=headers)
|
|
try:
|
|
with request.urlopen(req, timeout=15) as resp:
|
|
body = resp.read().decode("utf-8")
|
|
return resp.status, json.loads(body) if body else {}
|
|
except error.HTTPError as e:
|
|
body = e.read().decode("utf-8")
|
|
try:
|
|
parsed = json.loads(body) if body else {}
|
|
except Exception:
|
|
parsed = {"raw": body}
|
|
return e.code, parsed
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description="Test /patterns/<name>/send OTA flow.")
|
|
parser.add_argument(
|
|
"--base-url",
|
|
default="http://127.0.0.1",
|
|
help="Controller base URL (default: http://127.0.0.1)",
|
|
)
|
|
parser.add_argument(
|
|
"--pattern",
|
|
required=True,
|
|
help="Pattern name (without .py), e.g. blink",
|
|
)
|
|
parser.add_argument(
|
|
"--device-id",
|
|
default="",
|
|
help="Optional device id (MAC). If omitted, sends to all Wi-Fi devices.",
|
|
)
|
|
args = parser.parse_args()
|
|
|
|
base = args.base_url.rstrip("/")
|
|
pattern = args.pattern.strip()
|
|
if not pattern:
|
|
print("Pattern name is required.")
|
|
return 2
|
|
|
|
# Quick visibility before send.
|
|
status, patterns = _http_json("GET", f"{base}/patterns")
|
|
print(f"GET /patterns -> {status}")
|
|
if status != 200:
|
|
print(patterns)
|
|
return 1
|
|
if pattern not in patterns:
|
|
print(f"Pattern {pattern!r} not found in /patterns list.")
|
|
return 1
|
|
|
|
status, devices = _http_json("GET", f"{base}/devices")
|
|
print(f"GET /devices -> {status}")
|
|
if status != 200:
|
|
print(devices)
|
|
return 1
|
|
wifi_ids = [
|
|
did
|
|
for did, d in (devices or {}).items()
|
|
if isinstance(d, dict) and str(d.get("transport", "")).lower() == "wifi"
|
|
]
|
|
print(f"Wi-Fi devices in registry: {len(wifi_ids)}")
|
|
if wifi_ids:
|
|
print(" - " + "\n - ".join(wifi_ids))
|
|
|
|
payload = {"device_id": args.device_id} if args.device_id else {}
|
|
status, result = _http_json(
|
|
"POST", f"{base}/patterns/{pattern}/send", payload=payload
|
|
)
|
|
print(f"POST /patterns/{pattern}/send -> {status}")
|
|
print(json.dumps(result, indent=2))
|
|
|
|
if status != 200:
|
|
return 1
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|
|
|