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()