fix: remove outdated ble_client.py, add test_ota_notify.py to tools
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
This commit is contained in:
eddy 2026-08-04 18:33:35 +02:00
commit c36ca10edd
2 changed files with 123 additions and 87 deletions

123
tools/test_ota_notify.py Normal file
View file

@ -0,0 +1,123 @@
#!/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)