vk_mavlink2_Gui/tests/test_sn_request.py

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

os.environ.setdefault('QT_QPA_PLATFORM', 'offscreen')
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
from PyQt6.QtWidgets import QApplication
from app import Link, Monitor
from generated import vkfly

app = QApplication.instance() or QApplication([])


class FakeLink:
    def __init__(self):
        self.requests = []

    def request_sn(self, target_system):
        self.requests.append(target_system)

    def stop(self):
        pass


class SnRequestTest(unittest.TestCase):
    def setUp(self):
        self.monitor = Monitor()
        self.link = FakeLink()
        self.monitor.link = self.link
        self.writer = vkfly.MAVLink(None, srcSystem=42, srcComponent=1)

    def tearDown(self):
        self.monitor.close()

    def decode(self, message, writer):
        reader = vkfly.MAVLink(None)
        decoded = None
        for byte in message.pack(writer):
            result = reader.parse_char(bytes([byte]))
            if result is not None:
                decoded = result
        self.assertIsNotNone(decoded)
        return decoded

    def test_auto_request_once_and_manual_retry(self):
        heartbeat = self.decode(self.writer.heartbeat_encode(2, 3, 0, 0, 4, 3), self.writer)
        self.monitor.receive(heartbeat)
        self.monitor.receive(heartbeat)
        self.assertEqual(self.link.requests, [42])
        self.assertTrue(self.monitor.sn_button.isEnabled())
        self.monitor.sn_button.click()
        self.assertEqual(self.link.requests, [42, 42])
        self.assertEqual(self.monitor.labels['sn'].text(), '已请求，等待飞控回复')

    def test_ignores_non_autopilot_heartbeat(self):
        other = vkfly.MAVLink(None, srcSystem=99, srcComponent=1)
        self.monitor.receive(self.decode(other.heartbeat_encode(6, vkfly.MAV_AUTOPILOT_INVALID, 0, 0, 0, 3), other))
        self.assertEqual(self.link.requests, [])
        self.assertFalse(self.monitor.sn_button.isEnabled())

    def test_displays_flight_controller_sn_not_other_component(self):
        self.monitor.receive(self.decode(self.writer.heartbeat_encode(2, 3, 0, 0, 4, 3), self.writer))
        zeros = [0] * 16
        sn = list(b'VK123') + [0] * 11
        msg = self.writer.vk_comp_version_encode(1, zeros, zeros, sn, zeros, zeros, zeros)
        reader = vkfly.MAVLink(None)
        decoded = None
        for byte in msg.pack(self.writer):
            result = reader.parse_char(bytes([byte]))
            if result is not None:
                decoded = result
        self.monitor.receive(decoded)
        self.assertEqual(self.monitor.labels['sn'].text(), 'VK123')
        other = self.writer.vk_comp_version_encode(3, zeros, zeros, zeros, zeros, zeros, zeros)
        self.monitor.receive(other)
        self.assertEqual(self.monitor.labels['sn'].text(), 'VK123')

    def test_serial_writes_only_sn_request(self):
        link = Link('fake', 115200)
        link.request_sn(42)
        writes = []

        class FakeSerial:
            in_waiting = 0
            def __init__(self, *args, **kwargs):
                pass
            def __enter__(self):
                return self
            def __exit__(self, *args):
                pass
            def write(self, data):
                writes.append(bytes(data))
                return len(data)
            def read(self, size):
                link.running = False
                return b''

        with patch('app.serial.Serial', FakeSerial):
            link.run()
        self.assertEqual(len(writes), 1)
        reader = vkfly.MAVLink(None)
        msg = None
        for byte in writes[0]:
            result = reader.parse_char(bytes([byte]))
            if result is not None:
                msg = result
        self.assertEqual(msg.get_type(), 'COMMAND_LONG')
        self.assertEqual((msg.target_system, msg.target_component), (42, 1))
        self.assertEqual(msg.command, vkfly.MAV_CMD_REQUEST_MESSAGE)
        self.assertEqual((msg.param1, msg.param2), (vkfly.MAVLINK_MSG_ID_VK_COMP_VERSION, 1))
        self.assertEqual((msg.param3, msg.param4, msg.param5, msg.param6, msg.param7), (0, 0, 0, 0, 0))


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