fix: improve field diagnostics and mirror logs to USB CDC

tools/test_read_usb_log.py

@@ -0,0 +1,62 @@
+"""Test the Mac USB reader against a temporary pseudo-terminal, not hardware."""
+from pathlib import Path
+import os
+import pty
+import select
+import signal
+import subprocess
+import sys
+import tempfile
+import time
+
+ROOT = Path(__file__).resolve().parents[1]
+
+
+def exercise(diag_only):
+    master, slave = pty.openpty()
+    port = os.ttyname(slave)
+    process = None
+    try:
+        with tempfile.TemporaryDirectory(prefix='banban-usb-reader-') as directory:
+            output = Path(directory) / 'capture.log'
+            command = [sys.executable, str(ROOT / 'tools/read_usb_log.py'), '--port', port, '--output', str(output)]
+            if diag_only:
+                command.append('--diag-only')
+            process = subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+            notice = b''
+            deadline = time.monotonic() + 5
+            while '已连接'.encode() not in notice:
+                assert time.monotonic() < deadline, notice.decode(errors='replace')
+                if select.select([process.stderr], [], [], 0.1)[0]:
+                    notice += os.read(process.stderr.fileno(), 4096)
+                assert process.poll() is None, notice.decode(errors='replace')
+            payload = '[现场诊断] 中文连续测试\r\nordinary application log\r\n[USB日志] 在线\n[现场诊断] 最后一行'.encode()
+            os.write(master, payload[:3])  # Deliberately split a UTF-8 character.
+            time.sleep(0.02)
+            os.write(master, payload[3:])
+            deadline = time.monotonic() + 5
+            while not output.exists() or '[USB日志] 在线' not in output.read_text():
+                assert time.monotonic() < deadline
+                time.sleep(0.02)
+            process.send_signal(signal.SIGINT)
+            stdout, stderr = process.communicate(timeout=5)
+            assert process.returncode == 0, stderr.decode()
+            saved = output.read_text()
+            assert '\ufffd' not in saved
+            assert '[现场诊断] 中文连续测试' in saved
+            assert '[USB日志] 在线' in saved
+            assert '[现场诊断] 最后一行' in saved
+            assert ('ordinary application log' in saved) != diag_only
+            assert '[现场诊断] 中文连续测试' in stdout.decode()
+    finally:
+        if process and process.poll() is None:
+            process.kill()
+            process.communicate()
+        os.close(master)
+        os.close(slave)
+
+
+if __name__ == '__main__':
+    exercise(False)
+    exercise(True)
+    print('PASS: pseudo-terminal USB capture, split UTF-8, diagnostic filter, Ctrl+C and final-line flush')