Execute main RPC test loop.
(ctx: RunContext, qctx: QuietContext)
| 1813 | |
| 1814 | |
| 1815 | async def _run_rpc_tests(ctx: RunContext, qctx: QuietContext) -> int: |
| 1816 | """Execute main RPC test loop.""" |
| 1817 | upload_port = ctx.upload_port |
| 1818 | assert upload_port is not None |
| 1819 | serial_iface = ctx.serial_iface |
| 1820 | use_fbuild = ctx.use_fbuild |
| 1821 | |
| 1822 | if upload_port and not use_fbuild: |
| 1823 | kill_port_users(upload_port) |
| 1824 | time.sleep(0.5) |
| 1825 | |
| 1826 | if ctx.effective_tx_pin is not None and ctx.effective_rx_pin is not None: |
| 1827 | qctx.emit(f"PINS tx={ctx.effective_tx_pin} rx={ctx.effective_rx_pin}") |
| 1828 | |
| 1829 | print() |
| 1830 | print("=" * 60) |
| 1831 | print("EXECUTING AUTORESEARCH TESTS VIA RPC") |
| 1832 | print("=" * 60) |
| 1833 | |
| 1834 | test_failed = False |
| 1835 | stop_word_found: str | None = None |
| 1836 | json_rpc_commands = ctx.json_rpc_commands |
| 1837 | crash_decoder = ctx.crash_decoder |
| 1838 | |
| 1839 | client: RpcClient | None = None |
| 1840 | try: |
| 1841 | if ctx.discovery_client: |
| 1842 | client = ctx.discovery_client |
| 1843 | else: |
| 1844 | print(f"\U0001f4e1 Connecting to {upload_port}...") |
| 1845 | client = RpcClient( |
| 1846 | upload_port, |
| 1847 | timeout=ctx.timeout_seconds, |
| 1848 | serial_interface=serial_iface, |
| 1849 | verbose=True, |
| 1850 | crash_decoder=crash_decoder, |
| 1851 | ) |
| 1852 | await client.connect(boot_wait=1.0, drain_boot=True) |
| 1853 | print(f"{Fore.GREEN}\u2713 Connected{Style.RESET_ALL}") |
| 1854 | |
| 1855 | print(f"\n\U0001f527 Executing {len(json_rpc_commands)} RPC command(s)...") |
| 1856 | print("\u2500" * 60) |
| 1857 | |
| 1858 | for i, cmd in enumerate(json_rpc_commands, 1): |
| 1859 | method = cmd.get("method", "unknown") |
| 1860 | params = cmd.get("params", []) |
| 1861 | |
| 1862 | print(f"\n[{i}/{len(json_rpc_commands)}] Calling {method}()...") |
| 1863 | |
| 1864 | # Wrapper-side known-issue skip: synthesise a passing test_data |
| 1865 | # without touching the device. This is used by the OBJECT_FLED-on- |
| 1866 | # Teensy-4 substitution above (FastLED#3059 + #2772). The skipped |
| 1867 | # marker survives the json_rpc_commands transformation because it |
| 1868 | # gets set alongside `method`/`params` on the raw dict. |
| 1869 | if cmd.get("__skip_with_pass"): |
| 1870 | print( |
| 1871 | f"{Fore.YELLOW}⚠️ {cmd.get('__skip_reason', 'skipped')}" |
| 1872 | f"{Style.RESET_ALL}" |
no test coverage detected