c36ca10edd
All checks were successful
CI / build (pull_request) Successful in 1m22s
- Remove scripts/ble_client.py (uses old UUIDs, incompatible) - Move test_ota_notify.py to tools/ directory - Make serial port configurable via SYLDA_CLIENT_PORT env var
123 lines
4.3 KiB
Python
123 lines
4.3 KiB
Python
#!/usr/bin/env python3
|
|
"""Test OTA notify flow: writes 0x02 to server's CLIENT_OTA_STATUS via BLE,
|
|
then checks client serial for peripheral-mode entry log."""
|
|
import sys, time, asyncio, serial, threading, os
|
|
from bleak import BleakScanner, BleakClient
|
|
|
|
SERVICE_UUID = "4fafc201-1fb5-459e-8fcc-c5c9c331914b"
|
|
STATUS_UUID = "beb5483e-36e1-4688-b7f5-ea07361f26df"
|
|
SERVER_NAME = "SylDa_Server"
|
|
|
|
CLIENT_PORT = os.environ.get("SYLDA_CLIENT_PORT", "/dev/serial/by-id/usb-Espressif_USB_JTAG_serial_debug_unit_44:1B:F6:F5:D1:E0-if00")
|
|
TIMEOUT_S = 25
|
|
|
|
class TestResult:
|
|
def __init__(self):
|
|
self.pass_marker = False
|
|
self.fail_marker = False
|
|
self.lines = []
|
|
|
|
def monitor_serial(stop_event, result):
|
|
try:
|
|
ser = serial.Serial(CLIENT_PORT, 115200, timeout=0.5)
|
|
print(f"[TEST] Serial opened: {CLIENT_PORT}")
|
|
while not stop_event.is_set():
|
|
try:
|
|
data = ser.read(1024)
|
|
if data:
|
|
text = data.decode("ascii", errors="replace")
|
|
for line in text.splitlines():
|
|
if line.strip():
|
|
result.lines.append(line)
|
|
if "peripheral" in line.lower():
|
|
result.pass_marker = True
|
|
stop_event.set()
|
|
print(f"[TEST] PASS: {line}")
|
|
if "OTA: received trigger" in line or "entering peripheral" in line:
|
|
result.pass_marker = True
|
|
stop_event.set()
|
|
print(f"[TEST] PASS: {line}")
|
|
except serial.SerialException:
|
|
result.fail_marker = True
|
|
stop_event.set()
|
|
print("[TEST] FAIL: serial exception")
|
|
except Exception as e:
|
|
result.fail_marker = True
|
|
stop_event.set()
|
|
print(f"[TEST] FAIL: {e}")
|
|
finally:
|
|
try: ser.close()
|
|
except: pass
|
|
|
|
async def ble_write_02(result):
|
|
print("[TEST] Scanning for SylDa_Server...")
|
|
try:
|
|
device = await BleakScanner.find_device_by_name(SERVER_NAME, timeout=10)
|
|
except Exception as e:
|
|
print(f"[TEST] FAIL: BLE scan error: {e}")
|
|
result.fail_marker = True
|
|
return
|
|
|
|
if device is None:
|
|
print("[TEST] FAIL: SylDa_Server not found via BLE scan")
|
|
result.fail_marker = True
|
|
return
|
|
|
|
print(f"[TEST] Found: {device.name} ({device.address}), connecting...")
|
|
try:
|
|
async with BleakClient(device, timeout=10) as client:
|
|
print(f"[TEST] Connected, writing 0x02 to {STATUS_UUID}...")
|
|
svc = client.services.get_service(SERVICE_UUID)
|
|
if svc is None:
|
|
print("[TEST] FAIL: service not found")
|
|
result.fail_marker = True
|
|
return
|
|
char = svc.get_characteristic(STATUS_UUID)
|
|
if char is None:
|
|
print("[TEST] FAIL: CLIENT_OTA_STATUS characteristic not found")
|
|
result.fail_marker = True
|
|
return
|
|
await client.write_gatt_char(char, b'\x02', response=True)
|
|
print("[TEST] Wrote 0x02 OK, waiting for client to enter peripheral mode...")
|
|
except Exception as e:
|
|
print(f"[TEST] FAIL: BLE write error: {e}")
|
|
result.fail_marker = True
|
|
return
|
|
|
|
async def main():
|
|
result = TestResult()
|
|
stop = threading.Event()
|
|
|
|
serial_thread = threading.Thread(target=monitor_serial, args=(stop, result), daemon=True)
|
|
serial_thread.start()
|
|
|
|
await asyncio.sleep(2)
|
|
|
|
await ble_write_02(result)
|
|
|
|
if not stop.is_set():
|
|
print(f"[TEST] Waiting up to {TIMEOUT_S}s for client response...")
|
|
stop.wait(TIMEOUT_S)
|
|
|
|
stop.set()
|
|
serial_thread.join(timeout=3)
|
|
|
|
if result.pass_marker:
|
|
print("[TEST] ===== PASS =====")
|
|
return 0
|
|
elif result.fail_marker:
|
|
print("[TEST] ===== FAIL =====")
|
|
print("[TEST] Serial lines captured:")
|
|
for line in result.lines[-20:]:
|
|
print(f" {line}")
|
|
return 1
|
|
else:
|
|
print("[TEST] ===== INDETERMINATE (timeout, no pass/fail marker) =====")
|
|
print("[TEST] Serial lines captured:")
|
|
for line in result.lines[-20:]:
|
|
print(f" {line}")
|
|
return 2
|
|
|
|
if __name__ == "__main__":
|
|
rc = asyncio.run(main())
|
|
sys.exit(rc)
|