Cognitive_Agent/tests/unit/test_native_can_integration.py

from __future__ import annotations

import unittest
from pathlib import Path
from unittest.mock import patch

from cognitive_agent.application.native_can_runtime import (
    NativeCanExecutor,
    NativeCanRuntime,
)
from cognitive_agent.application.seat_service import REPOSITORY_ROOT, SeatDemoService
from cognitive_agent.infrastructure.can.gateway import CanFrame
from cognitive_agent.infrastructure.can.hardware import VehicleCanSettings


class _FakeBus:
    def __init__(self, received: list[CanFrame] | None = None) -> None:
        self.received = list(received or [])
        self.sent: list[CanFrame] = []
        self.closed = False

    def send(
        self,
        arbitration_id: int,
        data: bytes,
        *,
        is_fd: bool = False,
        bitrate_switch: bool = False,
    ) -> CanFrame:
        frame = CanFrame(
            arbitration_id,
            data,
            channel=1,
            is_fd=is_fd,
            bitrate_switch=bitrate_switch,
        )
        self.sent.append(frame)
        return frame

    def receive(
        self, timeout_ms: int, *, include_fd: bool = False
    ) -> CanFrame | None:
        return self.received.pop(0) if self.received else None

    def close(self) -> None:
        self.closed = True


class NativeCanIntegrationTests(unittest.TestCase):
    def setUp(self) -> None:
        self.runtime = NativeCanRuntime(REPOSITORY_ROOT)

    def _compile(self, domain: str, actions: list[dict]) -> tuple[dict, dict]:
        config = self.runtime.build_function_config(
            domain, actions, "req-native-test", "local_rules"
        )
        return config, self.runtime.compile(config)

    def test_climate_frames_match_can_dev_layout_and_repeat_policy(self) -> None:
        actions = [
            {
                "domain": "climate",
                "capability": "climate.set_power",
                "target": "vehicle",
                "operation": "set",
                "parameters": {"state": "on"},
            },
            {
                "domain": "climate",
                "capability": "climate.set_temperature",
                "target": "driver",
                "operation": "set",
                "parameters": {"temperature_c": 24},
            },
            {
                "domain": "climate",
                "capability": "climate.set_fan_level",
                "target": "vehicle",
                "operation": "set",
                "parameters": {"level": 3},
            },
        ]
        _, plan = self._compile("climate", actions)
        commands = plan["spec"]["commands"]

        self.assertEqual([item["signals"]["frame_id"] for item in commands], ["0x1DE"] * 3)
        self.assertEqual(commands[0]["signals"]["data_hex"], "00 00 00 10 00 00 00 00")
        self.assertEqual(commands[1]["signals"]["data_hex"], "00 00 00 00 18 00 00 00")
        self.assertEqual(commands[2]["signals"]["data_hex"], "30 00 00 00 00 00 00 00")
        self.assertEqual(commands[0]["repeat"], {"count": 3, "period_ms": 100})
        self.assertFalse(commands[0]["signals"]["is_fd"])

    def test_window_motion_uses_can_fd_with_brs(self) -> None:
        _, plan = self._compile(
            "window",
            [
                {
                    "domain": "window",
                    "capability": "window.set_motion",
                    "target": "front_left",
                    "operation": "set",
                    "parameters": {"motion": "open"},
                }
            ],
        )
        command = plan["spec"]["commands"][0]
        self.assertEqual(command["signals"]["frame_id"], "0x431")
        self.assertEqual(command["signals"]["data_hex"], "00 00 00 00 00 00 01 00")
        self.assertTrue(command["signals"]["is_fd"])
        self.assertTrue(command["signals"]["bitrate_switch"])

    def test_executor_requires_configured_feedback_for_vehicle_success(self) -> None:
        config, plan = self._compile(
            "climate",
            [
                {
                    "domain": "climate",
                    "capability": "climate.set_power",
                    "target": "vehicle",
                    "operation": "set",
                    "parameters": {"state": "on"},
                }
            ],
        )
        feedback = CanFrame(0x404, bytes((0, 1, 0, 0, 0, 0, 0, 0)), channel=1)
        bus = _FakeBus([feedback])
        with patch("cognitive_agent.application.native_can_runtime.time.sleep"):
            report = NativeCanExecutor(lambda: bus, feedback_timeout_ms=1).execute(
                self.runtime, plan, config
            )
        self.assertEqual(report["spec"]["status"], "succeeded")
        self.assertEqual(len(bus.sent), 3)
        self.assertTrue(bus.closed)

        no_feedback_bus = _FakeBus()
        with patch("cognitive_agent.application.native_can_runtime.time.sleep"):
            report = NativeCanExecutor(
                lambda: no_feedback_bus, feedback_timeout_ms=1
            ).execute(self.runtime, plan, config)
        self.assertEqual(report["spec"]["status"], "failed")

    def test_one_sentence_builds_seat_and_vehicle_channel_plans(self) -> None:
        service = SeatDemoService(REPOSITORY_ROOT)
        result = service.process(
            "副驾座椅通风3档，同时打开空调，温度设为24度，风量3档",
            provider="local",
        )
        self.assertEqual(result["status"], "succeeded")
        self.assertEqual(set(result["function_configs"]), {"seat", "native_climate"})
        self.assertEqual(set(result["command_plans"]), {"seat", "native_climate"})
        body_commands = result["command_plans"]["native_climate"]["spec"]["commands"]
        self.assertTrue(body_commands)
        self.assertTrue(all(item["bus"] == "body_can" for item in body_commands))
        self.assertFalse(result["deferred_requests"])

    def test_vehicle_timing_contract_matches_requested_60mhz_values(self) -> None:
        settings = VehicleCanSettings.from_env(Path(REPOSITORY_ROOT))
        self.assertTrue(settings.timing_is_verified)
        info = settings.public_info()
        self.assertEqual(info["arbitration"]["bitrate"], 500_000)
        self.assertEqual(info["arbitration"]["sample_point_percent"], 80)
        self.assertEqual(info["data"]["bitrate"], 2_000_000)
        self.assertEqual(info["data"]["sample_point_percent"], 80)


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