test_app_lifecycle.py

"""在真实 app_utils 环境运行示例 App 子进程，不模拟 AppController 的错误处理。"""

import os
import socket
import subprocess
import sys
import time
import unittest
from pathlib import Path

from websockets.sync.client import connect

try:
    from arduino.app_utils.app import AppController
    HAS_APP_UTILS = True
except ImportError:
    HAS_APP_UTILS = False

APP_DIR = Path(__file__).resolve().parents[1]


def child_environment(port):
    """
    @description         : 为独立 App 子进程设置测试端口和本 App 的 Brick 搜索目录
    @param port          : 本机测试端口
    @return              : 不改变父进程环境的字典
    """
    environment = dict(os.environ)
    environment["PYTHONPATH"] = str(APP_DIR / "bricks") + os.pathsep + environment.get("PYTHONPATH", "")
    environment["PYTHONUNBUFFERED"] = "1"
    environment["WEBSOCKET_SERVER_HOST"] = "127.0.0.1"
    environment["WEBSOCKET_SERVER_PORT"] = str(port)
    environment["WEBSOCKET_SERVER_PATH"] = "/ws"
    return environment


@unittest.skipUnless(HAS_APP_UTILS, "requires real arduino.app_utils; run sh tests/run.sh --unit on the board")
class AppLifecycleTests(unittest.TestCase):
    """验证真实生命周期控制器与示例入口的组合行为。"""

    def test_occupied_port_exits_nonzero_with_reason(self):
        """
        @description         : 即使 AppController 捕获 start 异常，示例仍以非零退出并记录具体原因
        @param               : 无参数
        @return              : 无返回值
        """
        with socket.create_server(("127.0.0.1", 0)) as occupied:
            result = subprocess.run(
                [sys.executable, "-B", str(APP_DIR / "python/main.py")],
                env=child_environment(occupied.getsockname()[1]),
                capture_output=True,
                text=True,
                encoding="utf-8",
                timeout=10,
            )
        self.assertEqual(result.returncode, 1, result.stdout + result.stderr)
        logs = result.stdout + result.stderr
        self.assertIn("WebSocket startup failed", logs)
        self.assertIn("WebSocket service unavailable", logs)
        self.assertIn("OSError", logs)

    @unittest.skipIf(sys.platform == "win32", "SIGTERM lifecycle is validated on Linux")
    def test_sigterm_releases_port_and_process_can_restart(self):
        """
        @description         : 真实 App 接收 SIGTERM 后在五秒内退出，并可在相同端口重新启动
        @param               : 无参数
        @return              : 无返回值
        """
        with socket.socket() as reservation:
            reservation.bind(("127.0.0.1", 0))
            port = reservation.getsockname()[1]
        for _ in range(2):
            process = subprocess.Popen(
                [sys.executable, "-B", str(APP_DIR / "python/main.py")],
                env=child_environment(port),
                stdout=subprocess.PIPE,
                stderr=subprocess.STDOUT,
                text=True,
                encoding="utf-8",
            )
            try:
                deadline = time.monotonic() + 5
                while True:
                    try:
                        with connect(f"ws://127.0.0.1:{port}/ws", open_timeout=0.3, close_timeout=0.3) as client:
                            client.send("ready")
                            self.assertEqual(client.recv(timeout=1), "ready")
                        break
                    except OSError:
                        if process.poll() is not None or time.monotonic() >= deadline:
                            self.fail("App did not become ready")
                        time.sleep(0.02)
                stopped_at = time.monotonic()
                process.terminate()
                output, _ = process.communicate(timeout=5)
                self.assertLess(time.monotonic() - stopped_at, 5)
                self.assertEqual(process.returncode, 143, output)
                self.assertNotIn("stop incomplete", output)
                with socket.create_server(("127.0.0.1", port)):
                    pass
            finally:
                if process.poll() is None:
                    process.kill()
                    process.communicate(timeout=3)


if __name__ == "__main__":
    unittest.main()