test_entrypoints.py

"""验证App只监听、用户脚本固定can0以及开发者回环的资源释放。"""

from contextlib import redirect_stdout, redirect_stderr
import io
import os
from pathlib import Path
import runpy
import sys
from types import SimpleNamespace
import unittest
from unittest.mock import Mock, patch, call

from fake_backend import FakeBackend
from generic_can import CANBackendError, CanBus


APP_ROOT = Path(__file__).resolve().parents[1]
SCRIPTS_ROOT = APP_ROOT / "scripts"


class EntrypointTests(unittest.TestCase):
    """运行入口代码并用内存对象替代硬件或App Lab调度器。"""

    def load_script(self, name):
        """
        @description         : 加载独立脚本但暂不执行main
        @param self          : 当前测试用例
        @param name          : scripts目录中的Python文件名
        @return              : 脚本的全局变量字典
        """
        with patch.object(sys, "path", [str(SCRIPTS_ROOT), *sys.path]):
            return runpy.run_path(str(SCRIPTS_ROOT / name))

    def test_app_startup_only_listens_and_ignores_old_interface_variable(self):
        """
        @description         : 验证App执行首次循环时只监听且旧环境变量不能选接口
        @param self          : 当前测试用例
        @return              : 无，使用FakeBackend且不启动真正App Lab
        """
        backend = FakeBackend()
        bus = CanBus(backend=backend)
        app = SimpleNamespace(run=Mock(side_effect=lambda user_loop: user_loop()))
        modules = {
            "arduino": SimpleNamespace(),
            "arduino.app_utils": SimpleNamespace(App=app),
        }
        try:
            with (
                patch.dict(sys.modules, modules),
                patch.dict(os.environ, {"GENERIC_CAN_INTERFACE": "can7"}),
                patch("generic_can.CanBus", return_value=bus) as constructor,
                patch("atexit.register"),
                patch("time.sleep"),
                redirect_stdout(io.StringIO()),
            ):
                runpy.run_path(str(APP_ROOT / "python" / "main.py"))
            app.run.assert_called_once()
            self.assertNotIn("interface", constructor.call_args.kwargs)
            self.assertEqual(bus.interface, "can0")
            self.assertEqual(backend.open_count, 1)
            self.assertEqual(backend.sent_frames, [])
        finally:
            bus.close()

    def test_user_scripts_reject_interface_selection_before_opening_bus(self):
        """
        @description         : 验证三个用户脚本不再接受接口选择且参数错误不会打开总线
        @param self          : 当前测试用例
        @return              : 无
        """
        for name, arguments in (
            ("can_monitor.py", []),
            (
                "send_test_frame.py",
                ["--id", "0x123", "--confirm", "SEND_GENERIC_CAN_FRAME"],
            ),
            ("named_message_test.py", ["receive", "status"]),
        ):
            with self.subTest(script=name):
                with patch("generic_can.CanBus") as constructor:
                    script = self.load_script(name)
                    with (
                        patch.object(
                            sys, "argv", [name, "--interface", "can1", *arguments]
                        ),
                        redirect_stderr(io.StringIO()),
                        self.assertRaises(SystemExit) as result,
                    ):
                        script["main"]()
                    self.assertEqual(result.exception.code, 2)
                    constructor.assert_not_called()

    def test_developer_loopback_uses_internal_backends_and_closes_them(self):
        """
        @description         : 验证开发者回环绕过正式CanBus且发送与接收后关闭两个后端
        @param self          : 当前测试用例
        @return              : 无，不打开真实虚拟接口
        """
        receiver, sender = FakeBackend(), FakeBackend()
        with (
            patch(
                "generic_can.backends.SocketCANBackend",
                side_effect=[receiver, sender],
            ) as constructor,
            patch("generic_can.CanBus") as public_bus,
            patch.object(sender, "send", side_effect=receiver.inject),
        ):
            script = self.load_script("loopback_test.py")
            with (
                patch.object(
                    sys, "argv",
                    ["loopback_test.py", "--confirm", script["CONFIRMATION"]],
                ),
                redirect_stdout(io.StringIO()),
            ):
                self.assertEqual(script["main"](), 0)
        self.assertEqual(
            constructor.call_args_list,
            [call(device="vcan0"), call(device="vcan0")],
        )
        public_bus.assert_not_called()
        self.assertEqual(receiver.close_count, 1)
        self.assertEqual(sender.close_count, 1)

    def test_developer_loopback_closes_resources_when_sender_open_fails(self):
        """
        @description         : 验证第二个后端打开失败时开发者回环仍释放已注册资源
        @param self          : 当前测试用例
        @return              : 无
        """
        receiver, sender = FakeBackend(), FakeBackend()
        with (
            patch(
                "generic_can.backends.SocketCANBackend",
                side_effect=[receiver, sender],
            ),
            patch.object(
                sender, "open", side_effect=CANBackendError("test open failed")
            ),
        ):
            script = self.load_script("loopback_test.py")
            with (
                patch.object(
                    sys, "argv",
                    ["loopback_test.py", "--confirm", script["CONFIRMATION"]],
                ),
                redirect_stdout(io.StringIO()),
            ):
                self.assertEqual(script["main"](), 1)
        self.assertEqual(receiver.close_count, 1)
        self.assertEqual(sender.close_count, 1)


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