Run coroutine test suite via RPC.
(ctx: RunContext)
| 1734 | |
| 1735 | |
| 1736 | async def _run_coroutine_tests(ctx: RunContext) -> int: |
| 1737 | """Run coroutine test suite via RPC.""" |
| 1738 | upload_port = ctx.upload_port |
| 1739 | assert upload_port is not None |
| 1740 | serial_iface = ctx.serial_iface |
| 1741 | |
| 1742 | print() |
| 1743 | print("=" * 60) |
| 1744 | print("COROUTINE TEST MODE - Task Creation, Stop, Await") |
| 1745 | print("=" * 60) |
| 1746 | print() |
| 1747 | |
| 1748 | client: RpcClient | None = None |
| 1749 | try: |
| 1750 | print(" Connecting to device...", end="", flush=True) |
| 1751 | client = RpcClient(upload_port, timeout=60.0, serial_interface=serial_iface) |
| 1752 | await client.connect(boot_wait=1.0) |
| 1753 | print(f" {Fore.GREEN}ok{Style.RESET_ALL}") |
| 1754 | |
| 1755 | print(" Sending testCoroutineAll RPC...", end="", flush=True) |
| 1756 | response = await client.send_and_match( |
| 1757 | "testCoroutineAll", match_key="passed", retries=3 |
| 1758 | ) |
| 1759 | print(f" {Fore.GREEN}ok{Style.RESET_ALL}") |
| 1760 | print() |
| 1761 | |
| 1762 | total = response.get("total", 0) |
| 1763 | passed_count = response.get("passed", 0) |
| 1764 | failed_count = response.get("failed", 0) |
| 1765 | results = response.get("results", {}) |
| 1766 | |
| 1767 | print(f" Results: {passed_count}/{total} passed", end="") |
| 1768 | if failed_count > 0: |
| 1769 | print(f", {failed_count} FAILED") |
| 1770 | else: |
| 1771 | print() |
| 1772 | |
| 1773 | if isinstance(results, dict): |
| 1774 | for test_name, test_result in results.items(): |
| 1775 | if isinstance(test_result, dict): |
| 1776 | success = test_result.get("success", False) |
| 1777 | duration = test_result.get("durationMs", "?") |
| 1778 | status = ( |
| 1779 | f"{Fore.GREEN}PASS{Style.RESET_ALL}" |
| 1780 | if success |
| 1781 | else f"{Fore.RED}FAIL{Style.RESET_ALL}" |
| 1782 | ) |
| 1783 | print(f" {status} {test_name} ({duration}ms)") |
| 1784 | if not success and "error" in test_result: |
| 1785 | print(f" {test_result['error']}") |
| 1786 | |
| 1787 | print() |
| 1788 | if response.get("success", False): |
| 1789 | print(f"{Fore.GREEN}COROUTINE TEST PASSED ({total} tests){Style.RESET_ALL}") |
| 1790 | return 0 |
| 1791 | else: |
| 1792 | print( |
| 1793 | f"{Fore.RED}COROUTINE TEST FAILED" |
no test coverage detected