- Добавлена обратная совместимость с прошивками, содержащими маркер no_auth. - Повторная авторизация после команды 04 и успешного обновления прошивки приведена к требуемому порядку. - В терминал команд добавлена ручная кнопка авторизации. - Реализовано OTA-обновление приложения с загрузкой платформенного артефакта, безопасной заменой и перезапуском. - Проверка обновлений учитывает только более новые версии по схеме major, minor и patch. - Обновлены имена Windows и Linux сборок, документация и тестовое покрытие.
96 lines
2.6 KiB
Python
96 lines
2.6 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")
|
|
|
|
def close(self) -> None:
|
|
self.is_open = False
|
|
|
|
|
|
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,
|
|
)
|
|
|
|
def test_close_clears_detected_firmware_mode(self) -> None:
|
|
conn = ComConnection()
|
|
conn._serial = FakeSerial()
|
|
conn.set_firmware_version("2.0.0_no_auth", auth_required=False)
|
|
|
|
conn.close()
|
|
|
|
self.assertIsNone(conn.firmware_version)
|
|
self.assertIsNone(conn.auth_required)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|