Danil Iarosh
5513eb8a56
- Добавлен стартовый handshake: ожидание строки версии контроллера, пауза 2000 мс и подтверждение готовности только по READY!. - Исправлена работа COM-буфера и аппаратного сброса через RTS, добавлены модульные тесты протокола. - Добавлена нативная сборка для Debian 13 x86_64 и обновлена документация. - Улучшена очистка Windows-сборки и кроссплатформенная обработка devnull.
83 lines
2.2 KiB
Python
83 lines
2.2 KiB
Python
from __future__ import annotations
|
|
|
|
import unittest
|
|
from collections import deque
|
|
|
|
from app.services.com_service import ComConnection
|
|
|
|
|
|
class FakeSerial:
|
|
def __init__(self, chunks: list[bytes] | None = None) -> None:
|
|
self.is_open = True
|
|
self._timeout = 0.1
|
|
self._chunks = deque(chunks or [])
|
|
self.events: list[str] = []
|
|
self._dtr = True
|
|
self._rts = False
|
|
|
|
@property
|
|
def timeout(self) -> float:
|
|
return self._timeout
|
|
|
|
@timeout.setter
|
|
def timeout(self, _value: float) -> None:
|
|
raise AssertionError("read_response не должен перенастраивать timeout COM-порта")
|
|
|
|
@property
|
|
def in_waiting(self) -> int:
|
|
return len(self._chunks[0]) if self._chunks else 0
|
|
|
|
@property
|
|
def dtr(self) -> bool:
|
|
return self._dtr
|
|
|
|
@dtr.setter
|
|
def dtr(self, value: bool) -> None:
|
|
self._dtr = value
|
|
self.events.append(f"dtr:{value}")
|
|
|
|
@property
|
|
def rts(self) -> bool:
|
|
return self._rts
|
|
|
|
@rts.setter
|
|
def rts(self, value: bool) -> None:
|
|
self._rts = value
|
|
self.events.append(f"rts:{value}")
|
|
|
|
def read(self, _size: int) -> bytes:
|
|
return self._chunks.popleft() if self._chunks else b""
|
|
|
|
def reset_input_buffer(self) -> None:
|
|
self.events.append("reset_input_buffer")
|
|
|
|
|
|
class ComConnectionTests(unittest.TestCase):
|
|
def test_read_response_keeps_serial_timeout_unchanged(self) -> None:
|
|
serial = FakeSerial([b"boot log\r\nperiphery controller ver_2.4.2_MDB\r\n"])
|
|
conn = ComConnection()
|
|
conn._serial = serial
|
|
|
|
self.assertEqual("boot log", conn.read_response(timeout=0.2))
|
|
self.assertEqual(
|
|
"periphery controller ver_2.4.2_MDB",
|
|
conn.read_response(timeout=0.2),
|
|
)
|
|
self.assertEqual(0.1, serial.timeout)
|
|
|
|
def test_reset_controller_uses_rts_pulse_with_dtr_inactive(self) -> None:
|
|
serial = FakeSerial()
|
|
conn = ComConnection()
|
|
conn._serial = serial
|
|
|
|
conn.reset_controller(pulse_seconds=0)
|
|
|
|
self.assertEqual(
|
|
["reset_input_buffer", "dtr:False", "rts:True", "rts:False"],
|
|
serial.events,
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|