Run --net-server autoresearch flow. 1. Send startNetServer RPC to ESP32 2. Connect host to ESP32's WiFi AP 3. Make HTTP requests to ESP32 and validate responses 4. Cleanup
(
client: RpcClient,
wifi: HostWifiManager,
)
| 511 | |
| 512 | |
| 513 | async def _run_net_server_autoresearch( |
| 514 | client: RpcClient, |
| 515 | wifi: HostWifiManager, |
| 516 | ) -> int: |
| 517 | """Run --net-server autoresearch flow. |
| 518 | |
| 519 | 1. Send startNetServer RPC to ESP32 |
| 520 | 2. Connect host to ESP32's WiFi AP |
| 521 | 3. Make HTTP requests to ESP32 and validate responses |
| 522 | 4. Cleanup |
| 523 | """ |
| 524 | import httpx |
| 525 | |
| 526 | print("\n--- Step 1: Start WiFi AP + HTTP Server on ESP32 ---") |
| 527 | response = await client.send("startNetServer", timeout=30.0) |
| 528 | server_info = response.data |
| 529 | |
| 530 | if not isinstance(server_info, dict) or not server_info.get("success"): |
| 531 | error = ( |
| 532 | server_info.get("error", "Unknown error") |
| 533 | if isinstance(server_info, dict) |
| 534 | else str(server_info) |
| 535 | ) |
| 536 | print(f" {Fore.RED}Failed to start net server: {error}{Style.RESET_ALL}") |
| 537 | return 1 |
| 538 | |
| 539 | ssid = server_info.get("ssid", NET_SSID) |
| 540 | password = server_info.get("password", NET_PASSWORD) |
| 541 | ip = server_info.get("ip", NET_AP_IP) |
| 542 | port = server_info.get("port", NET_SERVER_PORT) |
| 543 | print( |
| 544 | f" {Fore.GREEN}ESP32 WiFi AP started: SSID={ssid}, IP={ip}:{port}{Style.RESET_ALL}" |
| 545 | ) |
| 546 | |
| 547 | print("\n--- Step 2: Connect Host to ESP32 WiFi AP ---") |
| 548 | if not wifi.connect(ssid, password): |
| 549 | print(f" {Fore.RED}Failed to connect host to ESP32 AP{Style.RESET_ALL}") |
| 550 | return 1 |
| 551 | |
| 552 | print(f"\n--- Step 3: Validate HTTP Endpoints on {ip}:{port} ---") |
| 553 | base_url = f"http://{ip}:{port}" |
| 554 | tests_passed = 0 |
| 555 | tests_failed = 0 |
| 556 | |
| 557 | # Test 1: GET /ping |
| 558 | print(f"\n Test 1: GET /ping") |
| 559 | try: |
| 560 | r = httpx.get(f"{base_url}/ping", timeout=10.0) |
| 561 | if r.status_code == 200 and r.text == "pong": |
| 562 | print( |
| 563 | f" {Fore.GREEN}PASS{Style.RESET_ALL} - status={r.status_code}, body='{r.text}'" |
| 564 | ) |
| 565 | tests_passed += 1 |
| 566 | else: |
| 567 | print( |
| 568 | f" {Fore.RED}FAIL{Style.RESET_ALL} - status={r.status_code}, body='{r.text}' (expected 200, 'pong')" |
| 569 | ) |
| 570 | tests_failed += 1 |