keita
477ad5f5de
- pod_bridge: 协议G版编解码/地理位置引导律/PX4网关/串口链路/闭环主循环 - 含三项安全修复: position_time_ms拆分、表2-2锁定门控、run()异常护栏 - 测试92条全绿; 附树莓派4B(Ubuntu 24.04)部署注意事项与最小依赖清单
90 lines
2.2 KiB
Python
90 lines
2.2 KiB
Python
"""pod_serial_test.py —— PodSerial 单元测试（注入 fake serial 模块，无硬件）。"""
|
|
|
|
import sys
|
|
import types
|
|
|
|
import pytest
|
|
|
|
from pod_bridge.pod_serial import PodLink, PodSerial
|
|
from pod_bridge.protocol import TargetGeo, encode_target
|
|
|
|
|
|
class _FakeSerialInstance:
|
|
def __init__(self, **kwargs):
|
|
self.kwargs = kwargs
|
|
self._buf = b""
|
|
self.written = []
|
|
self.closed = False
|
|
|
|
@property
|
|
def in_waiting(self):
|
|
return len(self._buf)
|
|
|
|
def read(self, n):
|
|
data, self._buf = self._buf[:n], self._buf[n:]
|
|
return data
|
|
|
|
def write(self, data):
|
|
self.written.append(data)
|
|
|
|
def close(self):
|
|
self.closed = True
|
|
|
|
|
|
@pytest.fixture
|
|
def fake_serial(monkeypatch):
|
|
mod = types.ModuleType("serial")
|
|
mod.EIGHTBITS = 8
|
|
mod.PARITY_NONE = "N"
|
|
mod.STOPBITS_ONE = 1
|
|
mod.Serial = _FakeSerialInstance
|
|
monkeypatch.setitem(sys.modules, "serial", mod)
|
|
return mod
|
|
|
|
|
|
def test_empty_port_raises():
|
|
with pytest.raises(ValueError, match="port 为空"):
|
|
PodSerial("")
|
|
|
|
|
|
def test_serial_opened_with_8n1(fake_serial):
|
|
ps = PodSerial("COM3", baudrate=115200)
|
|
kw = ps._ser.kwargs
|
|
assert kw["port"] == "COM3"
|
|
assert kw["baudrate"] == 115200
|
|
assert kw["bytesize"] == 8
|
|
assert kw["parity"] == "N"
|
|
assert kw["stopbits"] == 1
|
|
assert kw["timeout"] == 0 # 非阻塞轮询
|
|
|
|
|
|
def test_poll_parses_buffered_frames(fake_serial):
|
|
ps = PodSerial("COM3")
|
|
ps._ser._buf = encode_target(TargetGeo(latitude_deg=31.0, longitude_deg=121.0))
|
|
frames = ps.poll()
|
|
assert len(frames) == 1
|
|
assert isinstance(frames[0], TargetGeo)
|
|
assert frames[0].latitude_deg == pytest.approx(31.0, abs=1e-7)
|
|
|
|
|
|
def test_poll_empty_returns_empty_list(fake_serial):
|
|
ps = PodSerial("COM3")
|
|
assert ps.poll() == []
|
|
|
|
|
|
def test_send_writes_bytes(fake_serial):
|
|
ps = PodSerial("COM3")
|
|
ps.send(b"\xeb\x90\x01")
|
|
assert ps._ser.written == [b"\xeb\x90\x01"]
|
|
|
|
|
|
def test_close_closes_serial(fake_serial):
|
|
ps = PodSerial("COM3")
|
|
ps.close()
|
|
assert ps._ser.closed is True
|
|
|
|
|
|
def test_satisfies_podlink_protocol(fake_serial):
|
|
ps = PodSerial("COM3")
|
|
assert isinstance(ps, PodLink)
|